diff --git a/event_bus_typed/Cargo.lock b/event_bus_typed/Cargo.lock new file mode 100644 index 0000000..950f0ae --- /dev/null +++ b/event_bus_typed/Cargo.lock @@ -0,0 +1,798 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + +[[package]] +name = "alloca" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5a7d05ea6aea7e9e64d25b9156ba2fee3fdd659e34e41063cd2fc7cd020d7f4" +dependencies = [ + "cc", +] + +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "autocfg" +version = "1.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53" + +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + +[[package]] +name = "bumpalo" +version = "3.20.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" + +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + +[[package]] +name = "cc" +version = "1.4.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54413ede23c2daf518f35156dfde027feb2374004d63bd497f983c8db9c0e313" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + +[[package]] +name = "clap" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa8876b300ab35ba921adea3dfd70157a46249b33f95c9084ae5709785478946" +dependencies = [ + "clap_builder", +] + +[[package]] +name = "clap_builder" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0797fb7aeb1406c84efac526901f7ec3ead2124f946b494e72879d4b54704d" +dependencies = [ + "anstyle", + "clap_lex", +] + +[[package]] +name = "clap_lex" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c133bc6a41be0d194c306b5506d15e6feeea7b1d6604bd3f8310dfb2ca96486" + +[[package]] +name = "criterion" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "950046b2aa2492f9a536f5f4f9a3de7b9e2476e575e05bd6c333371add4d98f3" +dependencies = [ + "alloca", + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "itertools", + "num-traits", + "oorandom", + "page_size", + "plotters", + "rayon", + "regex", + "serde", + "serde_json", + "tinytemplate", + "tokio", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8d80a2f4f5b554395e47b5d8305bc3d27813bacb73493eb1001e8f76dae29ea" +dependencies = [ + "cast", + "itertools", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "622f3fc73690be383c7214310406f28a90e6edeadc3cea882f9d71e495b9711a" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + +[[package]] +name = "crunchy" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" + +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + +[[package]] +name = "event_bus_typed" +version = "0.1.0" +dependencies = [ + "criterion", + "parking_lot", + "tokio", + "tracing", + "uuid", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef25905e51abafe4dcea6c15fec58c57b601cdbd0ee53d22ea1d3016c587d39b" + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", +] + +[[package]] +name = "half" +version = "2.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" +dependencies = [ + "cfg-if", + "crunchy", + "zerocopy", +] + +[[package]] +name = "itertools" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" +dependencies = [ + "either", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "js-sys" +version = "0.3.105" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce57d20d1ea864ce2ac172ab472d409214f4fd359f0b2a2775abdf522e2af99e" +dependencies = [ + "cfg-if", + "futures-util", + "wasm-bindgen", +] + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "num-traits" +version = "0.2.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + +[[package]] +name = "page_size" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30d5b2194ed13191c1999ae0704b7839fb18384fa22e49b57eeaa97d79ce40da" +dependencies = [ + "libc", + "winapi", +] + +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + +[[package]] +name = "rustversion" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" + +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + +[[package]] +name = "tokio" +version = "1.53.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "202caea871b69668250d242070849eb495be178ed697a3e98aebce5bc81a0bed" +dependencies = [ + "pin-project-lite", + "tokio-macros", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.6", +] + +[[package]] +name = "tracing" +version = "0.1.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" +dependencies = [ + "pin-project-lite", + "tracing-attributes", + "tracing-core", +] + +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "tracing-core" +version = "0.1.36" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" +dependencies = [ + "once_cell", +] + +[[package]] +name = "unicode-ident" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" + +[[package]] +name = "uuid" +version = "1.26.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ef6dac1e96601b4fb3acccccff2139741fcb757cb9a36089bf5be91cfb285ce" +dependencies = [ + "getrandom", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aecb87a33d3b0c5e3b7aa46336eaf486cffafbd281b195e4c8b80d50df2351bf" +dependencies = [ + "cfg-if", + "once_cell", + "rustversion", + "wasm-bindgen-macro", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-macro" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a690d511e3c1a8b3a55e33511e3c2c00c78415cd23650f32b808627f5696b9ed" +dependencies = [ + "quote", + "wasm-bindgen-macro-support", +] + +[[package]] +name = "wasm-bindgen-macro-support" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "411e4887f0071ef2d2164a9d5fdf2d20efbef78fccd3a78b0c10a1dc5295e48a" +dependencies = [ + "bumpalo", + "proc-macro2", + "quote", + "syn 3.0.6", + "wasm-bindgen-shared", +] + +[[package]] +name = "wasm-bindgen-shared" +version = "0.2.128" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81941cd78d0c92026c33e5e01312845a4cb1e9af3407f9134b100dd03144103e" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "web-sys" +version = "0.3.105" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fbddc4a036f00ec4f18c83445bd3115cb306a91da554919a099d9222fe4a7f8" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "zerocopy" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d35102a9f36d089ccae9e4c6802bc118be4487b80aaffc0ab4e0cf5ce92d2873" +dependencies = [ + "zerocopy-derive", +] + +[[package]] +name = "zerocopy-derive" +version = "0.8.57" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "146c01f5ab44258da43cf276c74a2763db2ff3969c9c652c3f2de07041d0b2bc" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/event_bus_typed/Cargo.toml b/event_bus_typed/Cargo.toml new file mode 100644 index 0000000..9e6f23f --- /dev/null +++ b/event_bus_typed/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "event_bus_typed" +version = "0.1.0" +edition = "2024" +publish = false + +[workspace] + +[lib] +name = "event_bus_typed" +path = "src/lib.rs" + +[[bench]] +name = "event_bus_throughput" +harness = false + +[dependencies] +tokio = { version = "1.53.1", default-features = false, features = ["rt", "sync"] } +parking_lot = "0.12.5" +tracing = "0.1" + +[dev-dependencies] +tokio = { version = "1.53.1", default-features = false, features = ["rt", "rt-multi-thread", "macros", "time", "sync"] } +criterion = { version = "0.8.2", features = ["async_tokio"] } +uuid = { version = "1.26.1", features = ["v4"] } diff --git a/event_bus_typed/README.md b/event_bus_typed/README.md new file mode 100644 index 0000000..3627081 --- /dev/null +++ b/event_bus_typed/README.md @@ -0,0 +1,148 @@ +# event_bus_typed + +Un bus d'événements asynchrone en mémoire pour Tokio, entièrement basé sur le typage fort en Rust (`TypeId`). + +--- + +## 🎯 Problématique résolue + +Dans l'implémentation initiale (`event_bus`), le modèle était inspiré de JavaScript (topics basés sur des chaînes de caractères) : +- Les événements étaient transportés via un pointeur générique `AnyEvent` (`Arc`). +- L'émission imposait de spécifier un topic string (`bus.emit("topic", event)`). +- La réception nécessitait de spécifier à la fois le type et le topic string (`bus.on_async::("topic", ...)`), puis d'effectuer un déréférencement / downcast dynamique (`downcast_ref::()` ou macro `match_event!`) sur chaque message reçu. + +**`event_bus_typed` résout entièrement cette complexité :** +- **Typage fort natif** : le routage est directement effectué par l'identifiant de type (`std::any::TypeId`), sans nom de topic requis. +- **Zéro downcast / déréférencement à la réception** : le callback reçoit directement la structure d'événement typée. +- **Syntaxe ergonomique** : support complet de la syntaxe turbofish demandée `bus.on_async::(|event| async move { ... })` ainsi que de l'inférence automatique `bus.on_async(|event: MessageUpdatedEvent| async move { ... })`. +- **Source unique de vérité** : plus besoin de `Scope` externe ; le contexte (ex: `channel_id`, `server_id`, `caller_id`) est directement transporté dans les champs de la structure typée. +- **Targeted wake-up** : canaux Tokio `broadcast` isolés par type d'événement, garantissant des performances optimales sans réveil inutile de tâches. + +--- + +## 🚀 Utilisation + +### 1. Définir des événements + +N'importe quelle structure Rust implémentant `Clone + Send + Sync + 'static` est automatiquement un `Event` valide (aucun macro derive supplémentaire nécessaire) : + +```rust +use uuid::Uuid; + +#[derive(Clone, Debug)] +pub struct MessageCreatedEvent { + pub server_id: Option, + pub channel_id: Uuid, + pub content: String, +} + +#[derive(Clone, Debug)] +pub struct MessageUpdatedEvent { + pub id: u64, + pub content: String, +} +``` + +--- + +### 2. Émission d'événements + +```rust +use event_bus_typed::EventBus; + +let bus = EventBus::new(); + +// Émission typée directe +bus.emit(MessageUpdatedEvent { + id: 42, + content: "Nouveau message".into(), +}); +``` + +--- + +### 3. Réception asynchrone (`on_async`) + +Syntaxe turbofish exacte demandée : + +```rust +bus.on_async::(|event| async move { + // `event` est directement de type MessageUpdatedEvent + println!("Message {} mis à jour : {}", event.id, event.content); +}); +``` + +Ou avec inférence sur l'argument de fermeture : + +```rust +bus.on_async(|event: MessageUpdatedEvent| async move { + println!("Contenu : {}", event.content); +}); +``` + +--- + +### 4. Réception synchrone (`on`) + +```rust +bus.on::(|event| { + println!("Nouveau message créé sur le salon : {:?}", event.channel_id); +}); +``` + +--- + +### 5. Utilisation avec contexte injecté (`on_async_with`) + +Pratique pour passer des services ou repositories sans clones manuels répétés : + +```rust +bus.on_async_with::(router, |router, event| async move { + router.gateway.send(...); +}); +``` + +--- + +### 6. Écouteurs One-Shot (`wait_next` et `wait_for`) + +Permet d'attendre un événement de manière linéaire avec une `Future` (sans boucle manuelle ni fuite de souscription) : + +```rust +// Attend le tout prochain événement de ce type +let event = bus.wait_next::().await?; + +// Ou attend un événement répondant à une condition précise +let confirmed = bus.wait_for::(|e| e.id == target_id).await?; +``` + +--- + +### 7. Flux direct / Récepteur sans callback (`subscribe`) + +Si vous préférez consommer les événements dans votre propre boucle de streaming : + +```rust +let mut rx = bus.subscribe::(); + +tokio::spawn(async move { + while let Ok(event) = rx.recv().await { + // `event` est directement MessageUpdatedEvent, aucun `match_event!` requis ! + println!("Reçu : {}", event.content); + } +}); +``` + +--- + +## 🧪 Tests et Benchmarks + +Exécuter les tests du crate : +```bash +cargo test --manifest-path event_bus_typed/Cargo.toml +``` + +Exécuter les benchmarks Criterion : +```bash +cargo bench --manifest-path event_bus_typed/Cargo.toml +``` diff --git a/event_bus_typed/benches/event_bus_throughput.rs b/event_bus_typed/benches/event_bus_throughput.rs new file mode 100644 index 0000000..38e8ff2 --- /dev/null +++ b/event_bus_typed/benches/event_bus_throughput.rs @@ -0,0 +1,488 @@ +#![allow(dead_code)] + +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, +}; +use std::time::Instant; + +use criterion::{Criterion, Throughput, criterion_group, criterion_main}; +use event_bus_typed::EventBus; +use tokio::runtime::Runtime; + +#[derive(Clone)] +struct SmallEvent { + value: u64, +} + +#[derive(Clone)] +struct StringEvent { + id: u64, + name: String, + message: String, +} + +#[derive(Clone)] +struct VecEvent { + id: u64, + payload: Vec, +} + +#[derive(Clone)] +struct ArcPayloadEvent { + id: u64, + payload: Arc<[u8]>, +} + +fn runtime() -> Runtime { + Runtime::new().expect("failed to create tokio runtime") +} + +async fn wait_until_received(received: &AtomicU64, expected: u64) { + while received.load(Ordering::Relaxed) < expected { + tokio::task::yield_now().await; + } +} + +fn bench_emit_no_subscriber(c: &mut Criterion) { + let mut group = c.benchmark_group("event_bus_typed/no_subscriber"); + group.throughput(Throughput::Elements(1)); + + let bus = EventBus::with_capacity(1024); + + group.bench_function("u64", |b| { + b.iter(|| { + bus.emit(42_u64); + }); + }); + + group.bench_function("small_struct", |b| { + b.iter(|| { + bus.emit(SmallEvent { value: 42 }); + }); + }); + + group.bench_function("string_struct", |b| { + b.iter(|| { + bus.emit(StringEvent { + id: 42, + name: "Alice".to_string(), + message: "hello from benchmark".to_string(), + }); + }); + }); + + group.bench_function("vec_payload_1kb", |b| { + b.iter(|| { + bus.emit(VecEvent { + id: 42, + payload: vec![7_u8; 1024], + }); + }); + }); + + let shared_payload: Arc<[u8]> = Arc::from(vec![7_u8; 1024].into_boxed_slice()); + + group.bench_function("arc_payload_1kb", |b| { + b.iter(|| { + bus.emit(ArcPayloadEvent { + id: 42, + payload: Arc::clone(&shared_payload), + }); + }); + }); + + group.finish(); +} + +fn bench_raw_subscriber(c: &mut Criterion) { + let rt = runtime(); + + let mut group = c.benchmark_group("event_bus_typed/raw_subscriber"); + group.throughput(Throughput::Elements(1)); + + group.bench_function("u64", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + let mut rx = bus.subscribe::(); + + let received = Arc::new(AtomicU64::new(0)); + let receiver_count = Arc::clone(&received); + + let receiver = tokio::spawn(async move { + while receiver_count.load(Ordering::Relaxed) < iters { + if rx.recv().await.is_ok() { + receiver_count.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(i); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + receiver.abort(); + elapsed + }); + }); + + group.bench_function("small_struct", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + let mut rx = bus.subscribe::(); + + let received = Arc::new(AtomicU64::new(0)); + let receiver_count = Arc::clone(&received); + + let receiver = tokio::spawn(async move { + while receiver_count.load(Ordering::Relaxed) < iters { + if rx.recv().await.is_ok() { + receiver_count.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(SmallEvent { value: i }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + receiver.abort(); + elapsed + }); + }); + + group.bench_function("string_struct", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + let mut rx = bus.subscribe::(); + + let received = Arc::new(AtomicU64::new(0)); + let receiver_count = Arc::clone(&received); + + let receiver = tokio::spawn(async move { + while receiver_count.load(Ordering::Relaxed) < iters { + if rx.recv().await.is_ok() { + receiver_count.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(StringEvent { + id: i, + name: "Alice".to_string(), + message: "hello from benchmark".to_string(), + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + receiver.abort(); + elapsed + }); + }); + + group.bench_function("vec_payload_1kb", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + let mut rx = bus.subscribe::(); + + let received = Arc::new(AtomicU64::new(0)); + let receiver_count = Arc::clone(&received); + + let receiver = tokio::spawn(async move { + while receiver_count.load(Ordering::Relaxed) < iters { + if rx.recv().await.is_ok() { + receiver_count.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(VecEvent { + id: i, + payload: vec![7_u8; 1024], + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + receiver.abort(); + elapsed + }); + }); + + let shared_payload: Arc<[u8]> = Arc::from(vec![7_u8; 1024].into_boxed_slice()); + + group.bench_function("arc_payload_1kb", |b| { + let shared = Arc::clone(&shared_payload); + b.to_async(&rt).iter_custom(|iters| { + let payload = Arc::clone(&shared); + async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + let mut rx = bus.subscribe::(); + + let received = Arc::new(AtomicU64::new(0)); + let receiver_count = Arc::clone(&received); + + let receiver = tokio::spawn(async move { + while receiver_count.load(Ordering::Relaxed) < iters { + if rx.recv().await.is_ok() { + receiver_count.fetch_add(1, Ordering::Relaxed); + } + } + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(ArcPayloadEvent { + id: i, + payload: Arc::clone(&payload), + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + receiver.abort(); + elapsed + } + }); + }); + + group.finish(); +} + +fn bench_typed_callback(c: &mut Criterion) { + let rt = runtime(); + + let mut group = c.benchmark_group("event_bus_typed/typed_callback"); + group.throughput(Throughput::Elements(1)); + + group.bench_function("u64", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let received = Arc::new(AtomicU64::new(0)); + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event; + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(i); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + subscription.abort(); + elapsed + }); + }); + + group.bench_function("small_struct", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let received = Arc::new(AtomicU64::new(0)); + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event.value; + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(SmallEvent { value: i }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + subscription.abort(); + elapsed + }); + }); + + group.bench_function("string_struct", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let received = Arc::new(AtomicU64::new(0)); + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event.id; + let _ = event.name.len(); + let _ = event.message.len(); + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(StringEvent { + id: i, + name: "Alice".to_string(), + message: "hello from benchmark".to_string(), + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + subscription.abort(); + elapsed + }); + }); + + group.bench_function("vec_payload_1kb", |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let received = Arc::new(AtomicU64::new(0)); + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event.id; + let _ = event.payload.len(); + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(VecEvent { + id: i, + payload: vec![7_u8; 1024], + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + subscription.abort(); + elapsed + }); + }); + + let shared_payload: Arc<[u8]> = Arc::from(vec![7_u8; 1024].into_boxed_slice()); + + group.bench_function("arc_payload_1kb", |b| { + let shared = Arc::clone(&shared_payload); + b.to_async(&rt).iter_custom(|iters| { + let payload = Arc::clone(&shared); + async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let received = Arc::new(AtomicU64::new(0)); + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event.id; + let _ = event.payload.len(); + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(ArcPayloadEvent { + id: i, + payload: Arc::clone(&payload), + }); + } + + wait_until_received(&received, iters).await; + + let elapsed = start.elapsed(); + subscription.abort(); + elapsed + } + }); + }); + + group.finish(); +} + +fn bench_multiple_subscribers(c: &mut Criterion) { + let rt = runtime(); + + let mut group = c.benchmark_group("event_bus_typed/multiple_subscribers"); + group.throughput(Throughput::Elements(1)); + + for subscriber_count in [1_u64, 2, 4, 8, 16, 32] { + group.bench_function(format!("{subscriber_count}_subscribers"), |b| { + b.to_async(&rt).iter_custom(|iters| async move { + let bus = EventBus::with_capacity(iters as usize + 1024); + + let expected = iters * subscriber_count; + let received = Arc::new(AtomicU64::new(0)); + let mut subscriptions = Vec::with_capacity(subscriber_count as usize); + + for _ in 0..subscriber_count { + let handler_count = Arc::clone(&received); + + let subscription = bus.on::(move |event| { + let _ = event.value; + handler_count.fetch_add(1, Ordering::Relaxed); + }); + + subscriptions.push(subscription); + } + + let start = Instant::now(); + + for i in 0..iters { + bus.emit(SmallEvent { value: i }); + } + + wait_until_received(&received, expected).await; + + let elapsed = start.elapsed(); + + for subscription in subscriptions { + subscription.abort(); + } + + elapsed + }); + }); + } + + group.finish(); +} + +criterion_group!( + benches, + bench_emit_no_subscriber, + bench_raw_subscriber, + bench_typed_callback, + bench_multiple_subscribers, +); + +criterion_main!(benches); diff --git a/event_bus_typed/src/bus.rs b/event_bus_typed/src/bus.rs new file mode 100644 index 0000000..2ad23a1 --- /dev/null +++ b/event_bus_typed/src/bus.rs @@ -0,0 +1,433 @@ +use std::any::{Any, TypeId}; +use std::collections::HashMap; +use std::fmt::Debug; +use std::sync::Arc; + +use parking_lot::RwLock; +use tokio::sync::broadcast; +use tokio::task::JoinHandle; +use tracing::{debug, trace, warn}; + +use crate::handler::{AsyncHandler, AsyncHandlerWith}; + +/// Default buffer capacity for each broadcast channel. +pub const DEFAULT_CAPACITY: usize = 1024; + +/// Marker trait for events dispatched through [`EventBus`]. +/// +/// Any type implementing `Clone + Send + Sync + 'static` automatically +/// implements `Event`. +pub trait Event: Clone + Send + Sync + 'static {} + +impl Event for T {} + +#[derive(Debug)] +struct EventBusInner { + channels: RwLock>>, + capacity: usize, +} + +/// A central, strongly-typed broadcast event bus. +/// +/// Unlike string/topic-based event buses, [`EventBus`] dispatches events directly +/// based on the concrete Rust type of the event (using [`TypeId`]). +/// +/// Under the hood, each event type is backed by an independent [`tokio::sync::broadcast`] +/// ring buffer. Dispatching an event via [`emit`](Self::emit) is $O(1)$ and non-blocking. +/// +/// # Sync callback example +/// ```rust,no_run +/// use event_bus_typed::EventBus; +/// +/// #[derive(Clone, Debug)] +/// struct MessageCreatedEvent { +/// content: String, +/// } +/// +/// #[tokio::main] +/// async fn main() { +/// let bus = EventBus::new(); +/// +/// bus.on::(|event| { +/// println!("Created message: {:?}", event); +/// }); +/// +/// bus.emit(MessageCreatedEvent { +/// content: "Hello!".into(), +/// }); +/// } +/// ``` +/// +/// # Async callback example +/// ```rust,no_run +/// use event_bus_typed::EventBus; +/// +/// #[derive(Clone, Debug)] +/// struct MessageUpdatedEvent { +/// id: u64, +/// content: String, +/// } +/// +/// #[tokio::main] +/// async fn main() { +/// let bus = EventBus::new(); +/// +/// bus.on_async::(|event| async move { +/// println!("(async) Updated message: {:?}", event); +/// }); +/// +/// bus.emit(MessageUpdatedEvent { +/// id: 42, +/// content: "World!".into(), +/// }); +/// } +/// ``` +#[derive(Debug, Clone)] +pub struct EventBus { + inner: Arc, +} + +impl Default for EventBus { + fn default() -> Self { + Self::new() + } +} + +impl EventBus { + /// Creates a new `EventBus` with default capacity (1024 messages per channel). + pub fn new() -> Self { + Self::with_capacity(DEFAULT_CAPACITY) + } + + /// Creates a new `EventBus` with custom buffer capacity per channel. + pub fn with_capacity(capacity: usize) -> Self { + debug!(capacity, "EventBus created"); + Self { + inner: Arc::new(EventBusInner { + channels: RwLock::new(HashMap::new()), + capacity, + }), + } + } + + // ───────────────────────────────────────────────────────────────────────── + // Internal Channel Management + // ───────────────────────────────────────────────────────────────────────── + + fn get_or_create_sender(&self) -> broadcast::Sender { + let type_id = TypeId::of::(); + + if let Some(tx) = self + .inner + .channels + .read() + .get(&type_id) + .and_then(|entry| entry.downcast_ref::>()) + { + return tx.clone(); + } + + let mut channels = self.inner.channels.write(); + if let Some(tx) = channels + .get(&type_id) + .and_then(|entry| entry.downcast_ref::>()) + { + return tx.clone(); + } + + let (tx, _) = broadcast::channel::(self.inner.capacity); + debug!( + event_type = std::any::type_name::(), + capacity = self.inner.capacity, + "New broadcast channel created" + ); + channels.insert(type_id, Box::new(tx.clone())); + tx + } + + // ───────────────────────────────────────────────────────────────────────── + // Emission + // ───────────────────────────────────────────────────────────────────────── + + /// Emits a strongly-typed event to all subscribers of `E` without blocking ($O(1)$). + /// + /// If no subscribers exist for this event type, the event is dropped immediately + /// without cloning or allocating. + /// + /// # Example + /// ```rust,no_run + /// # use event_bus_typed::EventBus; + /// # #[derive(Clone)] struct UserConnected { id: u64 } + /// # let bus = EventBus::new(); + /// bus.emit(UserConnected { id: 1 }); + /// ``` + pub fn emit(&self, event: E) { + trace!( + event_type = std::any::type_name::(), + "Emitting event" + ); + + let type_id = TypeId::of::(); + + if let Some(tx) = self + .inner + .channels + .read() + .get(&type_id) + .and_then(|entry| entry.downcast_ref::>()) + .filter(|tx| tx.receiver_count() > 0) + { + let _ = tx.send(event); + } + } + + // ───────────────────────────────────────────────────────────────────────── + // Subscriptions — Callbacks + // ───────────────────────────────────────────────────────────────────────── + + /// Registers a synchronous callback for events of type `E`. + /// + /// The handler runs in a dedicated background Tokio task. + /// Returns a [`JoinHandle`] allowing to cancel the subscription via [`.abort()`](JoinHandle::abort). + pub fn on(&self, handler: impl Fn(E) + Send + Sync + 'static) -> JoinHandle<()> { + let mut rx = self.subscribe::(); + let type_name = std::any::type_name::(); + + debug!(event_type = type_name, "Sync subscriber registered"); + + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + trace!(event_type = type_name, "Sync handler invoked"); + handler(event); + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + event_type = type_name, + skipped, + "Subscriber lagged behind and skipped messages" + ); + } + Err(broadcast::error::RecvError::Closed) => { + debug!(event_type = type_name, "Channel closed, subscriber exiting"); + break; + } + } + } + }) + } + + /// Registers an asynchronous callback for events of type `E`. + /// + /// The handler runs in a dedicated background Tokio task. + /// Returns a [`JoinHandle`] allowing to cancel the subscription via [`.abort()`](JoinHandle::abort). + /// + /// # Example + /// ```rust,no_run + /// # use event_bus_typed::EventBus; + /// # #[derive(Clone)] struct MessageUpdatedEvent { id: u64 } + /// # let bus = EventBus::new(); + /// bus.on_async::(|event| async move { + /// println!("Updated: {:?}", event.id); + /// }); + /// ``` + pub fn on_async(&self, handler: impl AsyncHandler) -> JoinHandle<()> { + let mut rx = self.subscribe::(); + let type_name = std::any::type_name::(); + + debug!(event_type = type_name, "Async subscriber registered"); + + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + trace!(event_type = type_name, "Async handler invoked"); + handler(event).await; + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + event_type = type_name, + skipped, + "Subscriber lagged behind and skipped messages" + ); + } + Err(broadcast::error::RecvError::Closed) => { + debug!(event_type = type_name, "Channel closed, subscriber exiting"); + break; + } + } + } + }) + } + + /// Registers an asynchronous callback for events of type `E` with an injected context `C`. + pub fn on_async_with( + &self, + context: C, + handler: impl AsyncHandlerWith, + ) -> JoinHandle<()> { + let mut rx = self.subscribe::(); + let type_name = std::any::type_name::(); + + debug!(event_type = type_name, "Async subscriber with context registered"); + + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + trace!(event_type = type_name, "Async handler with context invoked"); + handler(context.clone(), event).await; + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + event_type = type_name, + skipped, + "Subscriber lagged behind and skipped messages" + ); + } + Err(broadcast::error::RecvError::Closed) => { + debug!(event_type = type_name, "Channel closed, subscriber exiting"); + break; + } + } + } + }) + } + + // ───────────────────────────────────────────────────────────────────────── + // Low-Level Subscription (Direct Stream / Receiver) + // ───────────────────────────────────────────────────────────────────────── + + /// Returns a direct [`broadcast::Receiver`] for events of type `E`. + /// + /// Allows writing custom event processing loops without callback wrappers. + /// + /// # Example + /// ```rust,no_run + /// use event_bus_typed::EventBus; + /// + /// #[derive(Clone)] + /// struct MyEvent; + /// + /// #[tokio::main] + /// async fn main() { + /// let bus = EventBus::new(); + /// let mut rx = bus.subscribe::(); + /// bus.emit(MyEvent); + /// if let Ok(event) = rx.recv().await { + /// // direct typed `event` + /// } + /// } + /// ``` + pub fn subscribe(&self) -> broadcast::Receiver { + self.get_or_create_sender::().subscribe() + } + + /// Waits for the next event of type `E` to be emitted. + /// + /// Creates a temporary one-shot subscription and resolves as soon as an event of type `E` + /// is emitted. The subscription is automatically dropped after receiving the event. + /// + /// # Example + /// ```rust,no_run + /// # use event_bus_typed::EventBus; + /// # #[derive(Clone)] struct MyEvent { id: u64 } + /// #[tokio::main] + /// async fn main() { + /// let bus = EventBus::new(); + /// let event = bus.wait_next::().await.unwrap(); + /// println!("Received next event: {}", event.id); + /// } + /// ``` + pub async fn wait_next(&self) -> Result { + let mut rx = self.subscribe::(); + rx.recv().await + } + + /// Waits for an event of type `E` satisfying the given predicate to be emitted. + /// + /// Creates a temporary subscription, receives events of type `E`, and resolves + /// when the predicate returns `true`. The subscription is automatically dropped afterwards. + /// + /// # Example + /// ```rust,no_run + /// # use event_bus_typed::EventBus; + /// # #[derive(Clone)] struct MessageSaved { id: u64 } + /// #[tokio::main] + /// async fn main() { + /// let bus = EventBus::new(); + /// let target_id = 42; + /// let event = bus.wait_for::(|e| e.id == target_id).await.unwrap(); + /// println!("Saved message confirmed: {}", event.id); + /// } + /// ``` + pub async fn wait_for( + &self, + mut predicate: impl FnMut(&E) -> bool, + ) -> Result { + let mut rx = self.subscribe::(); + loop { + match rx.recv().await { + Ok(event) => { + if predicate(&event) { + return Ok(event); + } + } + Err(broadcast::error::RecvError::Lagged(skipped)) => { + warn!( + event_type = std::any::type_name::(), + skipped, + "wait_for subscriber lagged behind and skipped messages" + ); + } + Err(broadcast::error::RecvError::Closed) => { + return Err(broadcast::error::RecvError::Closed); + } + } + } + } + + // ───────────────────────────────────────────────────────────────────────── + // Metrics & Utilities + // ───────────────────────────────────────────────────────────────────────── + + /// Returns the total number of active subscribers for event type `E`. + pub fn subscriber_count(&self) -> usize { + let type_id = TypeId::of::(); + + self.inner + .channels + .read() + .get(&type_id) + .and_then(|e| e.downcast_ref::>()) + .map(|tx| tx.receiver_count()) + .unwrap_or(0) + } + + /// Returns `true` if there are any active subscribers for event type `E`. + pub fn has_subscribers(&self) -> bool { + self.subscriber_count::() > 0 + } + + /// Returns the total number of broadcast channels currently instantiated in the bus. + pub fn channel_count(&self) -> usize { + self.inner.channels.read().len() + } + + /// Returns `true` if no channels are currently registered in the bus. + pub fn is_empty(&self) -> bool { + self.inner.channels.read().is_empty() + } + + /// Removes and drops all channels in the event bus. + pub fn clear(&self) { + self.inner.channels.write().clear(); + } + + /// Removes and drops the channel associated with event type `E`. + pub fn clear_type(&self) { + let type_id = TypeId::of::(); + self.inner.channels.write().remove(&type_id); + } +} diff --git a/event_bus_typed/src/handler.rs b/event_bus_typed/src/handler.rs new file mode 100644 index 0000000..b9c5871 --- /dev/null +++ b/event_bus_typed/src/handler.rs @@ -0,0 +1,47 @@ +use std::future::Future; + +/// Handler trait for asynchronous event callbacks. +/// +/// This trait is automatically implemented for any closure matching +/// `Fn(E) -> Future + Send + Sync + 'static`. +/// +/// By using `Fn(E) -> >::Fut` as a supertrait, +/// Rust propagates the event type `E` directly into the closure's parameter, +/// enabling seamless turbofish syntax with clean field access: +/// +/// ```rust,no_run +/// # use event_bus_typed::EventBus; +/// # #[derive(Clone)] struct MessageUpdatedEvent { id: u64 } +/// # let bus = EventBus::new(); +/// bus.on_async::(|event| async move { +/// println!("Updated: {:?}", event.id); +/// }); +/// ``` +pub trait AsyncHandler: Fn(E) -> >::Fut + Send + Sync + 'static { + type Fut: Future + Send + 'static; +} + +impl AsyncHandler for F +where + F: Fn(E) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, +{ + type Fut = Fut; +} + +/// Handler trait for asynchronous event callbacks with an injected context. +/// +/// Automatically implemented for closures matching +/// `Fn(C, E) -> Future + Send + Sync + 'static`. +pub trait AsyncHandlerWith: Fn(C, E) -> >::Fut + Send + Sync + 'static { + type Fut: Future + Send + 'static; +} + +impl AsyncHandlerWith for F +where + C: Clone + Send + Sync + 'static, + F: Fn(C, E) -> Fut + Send + Sync + 'static, + Fut: Future + Send + 'static, +{ + type Fut = Fut; +} diff --git a/event_bus_typed/src/lib.rs b/event_bus_typed/src/lib.rs new file mode 100644 index 0000000..2165cce --- /dev/null +++ b/event_bus_typed/src/lib.rs @@ -0,0 +1,49 @@ +//! # event_bus_typed +//! +//! A strongly-typed, high-performance in-memory event bus for Tokio. +//! +//! ## Overview +//! +//! Unlike string/topic-based event buses, `event_bus_typed` routes events using +//! their concrete Rust types ([`std::any::TypeId`]). +//! +//! - **Strong typing**: No string keys required for event types, no manual `match_event!` +//! macros, and no runtime downcasting (`downcast_ref`) inside the subscriber loops. +//! - **Ergonomic async subscribers**: Handlers can be registered with clean turbofish syntax: +//! `bus.on_async::(|event| async move { ... })`. +//! - **Targeted wake-up**: Tokio broadcast channels are isolated per event type. +//! +//! ## Example +//! +//! ```rust,no_run +//! use event_bus_typed::EventBus; +//! +//! #[derive(Clone, Debug, PartialEq)] +//! struct MessageCreatedEvent { +//! content: String, +//! } +//! +//! #[tokio::main] +//! async fn main() { +//! let bus = EventBus::new(); +//! +//! // Async subscriber +//! bus.on_async::(|event| async move { +//! println!("Received message: {}", event.content); +//! }); +//! +//! // Emit event +//! bus.emit(MessageCreatedEvent { +//! content: "Hello from typed event bus!".into(), +//! }); +//! } +//! ``` + +mod bus; +mod handler; + +pub use bus::{DEFAULT_CAPACITY, Event, EventBus}; +pub use handler::{AsyncHandler, AsyncHandlerWith}; + +#[cfg(test)] +mod tests; diff --git a/event_bus_typed/src/tests.rs b/event_bus_typed/src/tests.rs new file mode 100644 index 0000000..89342b4 --- /dev/null +++ b/event_bus_typed/src/tests.rs @@ -0,0 +1,364 @@ +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; +use uuid::Uuid; + +use crate::EventBus; + +#[derive(Clone, Debug, PartialEq)] +struct MessageCreatedEvent { + channel_id: Uuid, + content: String, +} + +#[derive(Clone, Debug, PartialEq)] +struct MessageUpdatedEvent { + id: u64, + content: String, +} + +#[derive(Clone, Debug, PartialEq)] +struct MessageDeletedEvent { + id: u64, +} + +#[derive(Clone, Debug, PartialEq)] +struct UdpMetricEvent { + value: f32, +} + +// ── Sync Callbacks ────────────────────────────────────────────────────────── + +#[tokio::test] +async fn test_on_callback_sync() { + let bus = EventBus::new(); + let received = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&received); + + bus.on::(move |event| { + if event.content == "Hello" { + flag.store(true, Ordering::SeqCst); + } + }); + + bus.emit(MessageCreatedEvent { + channel_id: Uuid::new_v4(), + content: "Hello".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(received.load(Ordering::SeqCst)); +} + +// ── Async Callbacks (Exact User Requirement) ──────────────────────────────── + +#[tokio::test] +async fn test_on_async_callback_turbofish() { + let bus = EventBus::new(); + let received_content = Arc::new(tokio::sync::Mutex::new(String::new())); + let rc = Arc::clone(&received_content); + + // Exact syntax specified by the user: + // event_bus.on_async::(|event| async move { ... }); + bus.on_async::(move |event| { + let rc = Arc::clone(&rc); + async move { + let mut lock = rc.lock().await; + *lock = event.content; + } + }); + + // Exact syntax specified by the user: + // event_bus.emit(MessageUpdatedEvent { ... }); + bus.emit(MessageUpdatedEvent { + id: 42, + content: "Updated message content".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + let result = received_content.lock().await.clone(); + assert_eq!(result, "Updated message content"); +} + +#[tokio::test] +async fn test_on_async_callback_type_inferred() { + let bus = EventBus::new(); + let flag = Arc::new(AtomicBool::new(false)); + let f = Arc::clone(&flag); + + // Also supports inferring the event type from closure parameter + bus.on_async(move |event: MessageUpdatedEvent| { + let f = Arc::clone(&f); + async move { + if event.id == 99 { + f.store(true, Ordering::SeqCst); + } + } + }); + + bus.emit(MessageUpdatedEvent { + id: 99, + content: "Inferred".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(flag.load(Ordering::SeqCst)); +} + +// ── Targeted Wake-Up & Type Isolation ─────────────────────────────────────── + +#[tokio::test] +async fn test_on_targeted_wakeup() { + let bus = EventBus::new(); + let metric_called = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&metric_called); + + bus.on::(move |_| { + flag.store(true, Ordering::SeqCst); + }); + + // Emitting MessageCreatedEvent must never wake up UdpMetricEvent subscribers + bus.emit(MessageCreatedEvent { + channel_id: Uuid::new_v4(), + content: "Ignore me".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(!metric_called.load(Ordering::SeqCst)); +} + +#[tokio::test] +async fn test_multiple_subscribers_same_type() { + let bus = EventBus::new(); + let count = Arc::new(AtomicU32::new(0)); + + for _ in 0..3 { + let c = Arc::clone(&count); + bus.on::(move |_| { + c.fetch_add(1, Ordering::SeqCst); + }); + } + + bus.emit(MessageCreatedEvent { + channel_id: Uuid::new_v4(), + content: "Broadcast".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert_eq!(count.load(Ordering::SeqCst), 3); +} + +#[tokio::test] +async fn test_multiple_different_types_same_bus() { + let bus = EventBus::new(); + let msg_ok = Arc::new(AtomicBool::new(false)); + let metric_ok = Arc::new(AtomicBool::new(false)); + let m_flag = Arc::clone(&msg_ok); + let u_flag = Arc::clone(&metric_ok); + + bus.on::(move |event| { + if event.content == "Test" { + m_flag.store(true, Ordering::SeqCst); + } + }); + + bus.on::(move |metric| { + if (metric.value - 42.5).abs() < 0.001 { + u_flag.store(true, Ordering::SeqCst); + } + }); + + bus.emit(MessageCreatedEvent { + channel_id: Uuid::new_v4(), + content: "Test".into(), + }); + bus.emit(UdpMetricEvent { value: 42.5 }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert!(msg_ok.load(Ordering::SeqCst)); + assert!(metric_ok.load(Ordering::SeqCst)); +} + +// ── Direct Stream / Receiver (No match_event! needed) ─────────────────────── + +#[tokio::test] +async fn test_subscribe_direct_typed_receiver() { + let bus = EventBus::new(); + let mut rx = bus.subscribe::(); + + bus.emit(MessageUpdatedEvent { + id: 123, + content: "Direct typed".into(), + }); + + let event = rx.recv().await.expect("failed to receive event"); + // event is directly of type MessageUpdatedEvent, no downcast needed! + assert_eq!(event.id, 123); + assert_eq!(event.content, "Direct typed"); +} + +// ── In-Handler Filtering (Direct Field Access) ────────────────────────────── + +#[tokio::test] +async fn test_filter_by_field_in_subscriber() { + let bus = EventBus::new(); + let channel_a = Uuid::new_v4(); + let channel_b = Uuid::new_v4(); + + let count_a = Arc::new(AtomicU32::new(0)); + let count_b = Arc::new(AtomicU32::new(0)); + let count_global = Arc::new(AtomicU32::new(0)); + + let ca = Arc::clone(&count_a); + bus.on_async::(move |event| { + let ca = Arc::clone(&ca); + async move { + if event.channel_id == channel_a { + ca.fetch_add(1, Ordering::SeqCst); + } + } + }); + + let cb = Arc::clone(&count_b); + bus.on_async::(move |event| { + let cb = Arc::clone(&cb); + async move { + if event.channel_id == channel_b { + cb.fetch_add(1, Ordering::SeqCst); + } + } + }); + + let cg = Arc::clone(&count_global); + bus.on::(move |_| { + cg.fetch_add(1, Ordering::SeqCst); + }); + + // Emit event with channel_a + bus.emit(MessageCreatedEvent { + channel_id: channel_a, + content: "For A".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + // Both channel_a handler and global handler processed it, but not channel_b + assert_eq!(count_a.load(Ordering::SeqCst), 1); + assert_eq!(count_b.load(Ordering::SeqCst), 0); + assert_eq!(count_global.load(Ordering::SeqCst), 1); +} + +// ── Async with Context ────────────────────────────────────────────────────── + +#[tokio::test] +async fn test_on_async_with_context() { + let bus = EventBus::new(); + let prefix = Arc::new("Prefix: ".to_string()); + let result = Arc::new(tokio::sync::Mutex::new(String::new())); + let r = Arc::clone(&result); + + bus.on_async_with::(prefix, move |ctx, event| { + let r = Arc::clone(&r); + async move { + let mut lock = r.lock().await; + *lock = format!("{}{}", ctx, event.content); + } + }); + + bus.emit(MessageCreatedEvent { + channel_id: Uuid::new_v4(), + content: "Hello Context".into(), + }); + + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + let final_str = result.lock().await.clone(); + assert_eq!(final_str, "Prefix: Hello Context"); +} + +// ── Metrics, Utilities & Cleanup ──────────────────────────────────────────── + +#[test] +fn test_subscriber_count_and_clear() { + let bus = EventBus::new(); + assert_eq!(bus.subscriber_count::(), 0); + assert!(!bus.has_subscribers::()); + + let _sub = bus.subscribe::(); + assert_eq!(bus.subscriber_count::(), 1); + assert!(bus.has_subscribers::()); + assert_eq!(bus.channel_count(), 1); + + bus.clear_type::(); + assert_eq!(bus.subscriber_count::(), 0); + assert_eq!(bus.channel_count(), 0); +} + +#[tokio::test] +async fn test_subscription_abort() { + let bus = EventBus::new(); + let count = Arc::new(AtomicU32::new(0)); + let c = Arc::clone(&count); + + let handle = bus.on::(move |_| { + c.fetch_add(1, Ordering::SeqCst); + }); + + bus.emit(MessageDeletedEvent { id: 1 }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + assert_eq!(count.load(Ordering::SeqCst), 1); + + // Cancel the subscription + handle.abort(); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + + bus.emit(MessageDeletedEvent { id: 2 }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + // Count should not increase after abort + assert_eq!(count.load(Ordering::SeqCst), 1); +} + +// ── One-Shot Listeners (wait_next & wait_for) ──────────────────────────────── + +#[tokio::test] +async fn test_wait_next() { + let bus = EventBus::new(); + let b = bus.clone(); + + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + b.emit(MessageUpdatedEvent { + id: 777, + content: "Next event".into(), + }); + }); + + let event = bus.wait_next::().await.unwrap(); + assert_eq!(event.id, 777); + assert_eq!(event.content, "Next event"); +} + +#[tokio::test] +async fn test_wait_for() { + let bus = EventBus::new(); + let b = bus.clone(); + + tokio::spawn(async move { + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + b.emit(MessageUpdatedEvent { + id: 1, + content: "Ignore".into(), + }); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + b.emit(MessageUpdatedEvent { + id: 2, + content: "Target".into(), + }); + }); + + let event = bus + .wait_for::(|e| e.id == 2) + .await + .unwrap(); + assert_eq!(event.id, 2); + assert_eq!(event.content, "Target"); +}