feat: redis-style pub/sub - #55
Merged
Merged
Conversation
add SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, and PUBLISH commands to the protocol layer. parsing only — no execution yet.
introduces PubSubManager backed by DashMap + tokio broadcast for lock-free fan-out messaging. supports channel subscriptions, glob pattern matching (PSUBSCRIBE), and concurrent publish. includes comprehensive unit tests for glob matching and pub/sub semantics.
- server.rs: create and pass PubSubManager to all connections - connection.rs: subscriber mode state machine with select! loop, push message serialization, subscription cleanup on disconnect - concurrent_handler.rs: PUBLISH support, sub commands return error
- remove unused imports in pubsub.rs - remove unused parameters from handle_subscriber_mode - fix formatting in protocol parse functions
covers subscribe, unsubscribe, psubscribe, punsubscribe, publish with edge cases: no args, multiple channels, wrong arity, case insensitivity.
completes the pub/sub command set with redis-compatible introspection: - PUBSUB CHANNELS [pattern] — list active channels with optional glob - PUBSUB NUMSUB [channel ...] — subscriber counts per channel - PUBSUB NUMPAT — count of active pattern subscriptions includes parsing tests and end-to-end verification with redis-cli.
- fix race condition: unsubscribe/punsubscribe now only decrement subscription_count when the channel actually exists in the registry - replace busy-wait polling in recv_any_message with FuturesUnordered for proper async multiplexing across all broadcast receivers - remove unnecessary String allocations in serialize functions (use Bytes::copy_from_slice and Bytes::from_static instead) - simplify is_subscribe_frame using eq_ignore_ascii_case on byte slices instead of allocating a String for case comparison - simplify enter_sub detection with iter().any() - organize imports (group std::time together)
- README.md: 65+ → 76 commands, 609 → 639 tests, ~14k → ~21k LOC - README.md: add pub/sub to feature list - bench/README.md: update command count, remove pub/sub from dragonfly advantages (ember now has it)
kacy
added a commit
that referenced
this pull request
Feb 11, 2026
* feat: add pub/sub command parsing add SUBSCRIBE, UNSUBSCRIBE, PSUBSCRIBE, PUNSUBSCRIBE, and PUBLISH commands to the protocol layer. parsing only — no execution yet. * feat: add pubsub manager with broadcast channels introduces PubSubManager backed by DashMap + tokio broadcast for lock-free fan-out messaging. supports channel subscriptions, glob pattern matching (PSUBSCRIBE), and concurrent publish. includes comprehensive unit tests for glob matching and pub/sub semantics. * feat: wire pub/sub into connection handlers - server.rs: create and pass PubSubManager to all connections - connection.rs: subscriber mode state machine with select! loop, push message serialization, subscription cleanup on disconnect - concurrent_handler.rs: PUBLISH support, sub commands return error * chore: fix clippy warnings and formatting - remove unused imports in pubsub.rs - remove unused parameters from handle_subscriber_mode - fix formatting in protocol parse functions * test: add pub/sub command parsing tests covers subscribe, unsubscribe, psubscribe, punsubscribe, publish with edge cases: no args, multiple channels, wrong arity, case insensitivity. * feat: add PUBSUB CHANNELS/NUMSUB/NUMPAT introspection commands completes the pub/sub command set with redis-compatible introspection: - PUBSUB CHANNELS [pattern] — list active channels with optional glob - PUBSUB NUMSUB [channel ...] — subscriber counts per channel - PUBSUB NUMPAT — count of active pattern subscriptions includes parsing tests and end-to-end verification with redis-cli. * refactor: audit fixes for pub/sub implementation - fix race condition: unsubscribe/punsubscribe now only decrement subscription_count when the channel actually exists in the registry - replace busy-wait polling in recv_any_message with FuturesUnordered for proper async multiplexing across all broadcast receivers - remove unnecessary String allocations in serialize functions (use Bytes::copy_from_slice and Bytes::from_static instead) - simplify is_subscribe_frame using eq_ignore_ascii_case on byte slices instead of allocating a String for case comparison - simplify enter_sub detection with iter().any() - organize imports (group std::time together) * docs: update command count to 76, add pub/sub to feature list - README.md: 65+ → 76 commands, 609 → 639 tests, ~14k → ~21k LOC - README.md: add pub/sub to feature list - bench/README.md: update command count, remove pub/sub from dragonfly advantages (ember now has it)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
summary
adds redis-compatible pub/sub messaging with full command support:
tokio::select!to multiplex incoming commands and pushed messages*,?,[abc],[^abc],\x(matches redis PSUBSCRIBE behavior)also updates READMEs: 65+ → 76 commands, 609 → 639 tests, ~14k → ~21k LOC.
what was tested
design considerations
broadcastchannel per subscription for fan-out. buffer capacity is 256 messages — lagged subscribers miss messages (matching redis behavior)FuturesUnorderedfor efficient async multiplexing across all broadcast receivers (no busy-wait polling)Bytes::from_static/Bytes::copy_from_sliceused in serialization to avoid unnecessary String allocationsArc, same as ServerContext