An event bus over Redis Streams.
Streams rather than Redis pub/sub, and the difference is the whole reason this is usable. Pub/sub is fire-and-forget: a subscriber that is restarting when an event is published never learns it happened, and there is no way to find out afterwards. A stream keeps its entries, hands each consumer group its own cursor, and tracks what has not been acknowledged — which is what makes at-least-once possible at all.
At-least-once, specifically: an event is delivered until a consumer
acknowledges it, so a consumer that crashes mid-handler sees the event again
when it comes back. Consumers must be idempotent. :id on the envelope is
there for that — it is assigned by the publisher, so it is the same across
redeliveries, unlike the stream entry id.
An event bus over Redis Streams. Streams rather than Redis pub/sub, and the difference is the whole reason this is usable. Pub/sub is fire-and-forget: a subscriber that is restarting when an event is published never learns it happened, and there is no way to find out afterwards. A stream keeps its entries, hands each consumer group its own cursor, and tracks what has not been acknowledged — which is what makes at-least-once possible at all. At-least-once, specifically: an event is delivered until a consumer acknowledges it, so a consumer that crashes mid-handler sees the event again when it comes back. Consumers must be idempotent. `:id` on the envelope is there for that — it is assigned by the publisher, so it is the same across redeliveries, unlike the stream entry id.
(create-redis-streams-bus
pool
&
[{:keys [prefix group max-len max-deliveries min-idle-ms] :as opts}])An event bus backed by Redis Streams.
Args: pool - a JedisPool opts - :prefix stream name prefix (default "wagoe:events:") :group consumer group name — one per logical consumer, since a group shares a cursor and each event goes to exactly one member of it. Two services that both need every event need two groups, not two consumers in one. :max-len approximate stream length to keep (default 10000) :max-deliveries attempts before an event is dead-lettered (default 5; nil retries forever, which stalls the topic behind a poison message) :min-idle-ms how long an entry must be untouched before this consumer reclaims it from another (default 30000)
An event bus backed by Redis Streams.
Args:
pool - a JedisPool
opts - :prefix stream name prefix (default "wagoe:events:")
:group consumer group name — one per logical consumer, since a
group shares a cursor and each event goes to exactly one
member of it. Two services that both need every event
need two groups, not two consumers in one.
:max-len approximate stream length to keep (default 10000)
:max-deliveries attempts before an event is dead-lettered
(default 5; nil retries forever, which stalls the
topic behind a poison message)
:min-idle-ms how long an entry must be untouched before this
consumer reclaims it from another (default 30000)(stop! {:keys [executor subscriptions]})Stop every subscription and release the threads.
Does not close the pool: this takes one rather than making it, so whoever
created it closes it. module-wiring builds a pool solely for the bus and
closes it in halt-key!; a caller passing a shared pool keeps theirs.
Stop every subscription and release the threads. Does not close the pool: this takes one rather than making it, so whoever created it closes it. `module-wiring` builds a pool solely for the bus and closes it in `halt-key!`; a caller passing a shared pool keeps theirs.
cljdoc builds & hosts documentation for Clojure/Script libraries
| Ctrl+k | Jump to recent docs |
| ← | Move to previous article |
| → | Move to next article |
| Ctrl+/ | Jump to the search field |