A Redis stream is a set of messages consisting of key-value pairs that are identified by a time and sequence number. Streams are powerful objects that can roughly be used for three purposes:
- Maintain and query a log of events, i.e., a timeline.
- Provide an alternative to Redis' publish/subscribe API that ensures messages get delivered by all clients even if they are offline at the moment an event is published.
- Distribute messages over a group of clients. This mode assigns messages to clients in a round-robin fashion. Clients confirm a specific message is handled. Living clients can inspect the stream for possibly dead clients and migrate the pending messages to other clients.
This library abstracts the latter two scenarios. The main predicates are
- xadd/4 to add to a stream
- xlisten/3 to read and broadcast messages from a stream
- xlisten_group/5 to act as a consumer in a consumer group.
- xstream_set(+Redis, +Key, +Option)
- Set an option on for Key on Redis. Currently supports:
- xadd(+Redis, +Key, ?Id, +Data:dict) is det
- Add a message to a the stream Key on Redis. The length of the stream
can be capped using the xstream_set/3 option
maxlen(Count). If Id is unbound, generating the id is left to the server and Id is unified with the returned id. The returned id is a string consisting of the time stamp in milliseconds and a sequence number. See Redis docs for details.
- xlisten(+Redis, +Streams, +Options)
- Listen using
XREADon one or more Streams on the server Redis. For each message that arrives, call broadcast/1, where Data is a dict representing the message.
broadcast(redis(Redis, Stream, Id, Data))
- Process at most Count messages per stream for each request.
- Normally either
0to start get all messages from the epoch or
$to get messages starting with the last. Default is
- May be used as an alternative to the start/1 option to specify the start for each stream. This may be used to restart listening if the application remembers the last processed id.
Note that this predicate does not terminate. It is normally executed in a thread. The following call listens to the streams
key2on the default Redis server. Using
reconnect(true), the client will try to re-establish a connection if the collection got lost.
?- redis_connect(default, C, [reconnect(true)]), thread_create(xlisten(C, [key1, key2], [start($)]), _, [detached(true)]).
- xlisten(+Redis, +Streams, +OnBroadCast, +OnIdle, +Options)[private]
- Generalized version of xlisten/3 that is provided two callbacks: one
to handle a message and one after each time the underlying
XREADGROUPhas returned and the messages are processed. These callbacks are called as follows:
call(OnBroadCast, +Redis, +Stream, +MessageId, +Dict) call(OnIdle, +Redis, +Streams, +Starts, +NewStarts, +Options)
Both callbacks must succeeds and not leave any open choice points. Failure or exception causes xlisten/5 to stop.
- dispatch_messages(+Messages, +Stream, +Redis, +Start0, -Start) is det[private]
- xlisten_group(+Redis, +Group, +Consumer, +Streams, +Options)
- Listen as Consumer to Group. This is similar to xlisten/3, with the
- Instead of using broadcast/1, broadcast_request/1 is used and
the message is only considered processed if broadcast_request/1
succeeds. If the message is handled with success, an
XACKis sent to the server.
XREADGROUPto return with timeout when no messages arrive within Seconds. On a timeout, xidle_group/5 is called which will try to handle messages to other consumers pending longer than Seconds. Choosing the time depends on the application. Notably:
- Using a time shorter than the required processing time
will make the job migrate from consumer to consumer until
max_deliveries(Count)is exceeded. Note that the original receiver does not notice that the job is claimed and thus multiple consumers may ultimately answer the message.
- Using a too long time causes an unnecessarily long delay if a node fails.
- Using a time shorter than the required processing time will make the job migrate from consumer to consumer until
- Re-deliver (using
XCLAIM) a message max Count times. Exceeding this calls xhook/2. Default Count is
- Do not claim more than Count messages during a single idle
action. Default is
- Instead of using broadcast/1, broadcast_request/1 is used and the message is only considered processed if broadcast_request/1 succeeds. If the message is handled with success, an
- xidle_group(+Redis, +Streams, +Starts, +NewStarts, +Options) is det[private]
- Called after
XREADGROUPreturns and the returned messages (if any) have been processed. If
Start == NewStartsno messages have been processed, indicating a timeout.
This implementation looks for idle messages on other consumer and will try to claim them.
- check_limit_deliveries(+Redis, +Stream, +Delivered, +Id, +Options)[private]
- If a message gets delivered to several nodes and none of the nodes is able to process it, we should stop trying to do so at some point because the failure is most likely due to persistent error and piling up such messages will harm the cluster.
- xleave_group(+Redis, +Group, +Consumer, +Streams) is det[private]
- Remove Consumer from Group.
- May be called from a consumer listener to stop the consumer. This
predicate throws the exception
redis(stop(Leave)), which is caught by xlisten_group/5.
- xhook(+Stream, +Event)[multifile]
- This multifile predicate is called on certain stream events. Defined
- delivery_failed(Id, Group, Delivered)
- A message was delivered more than specified by max_deliveries/1
of xlisten_group/5. Id is the message id, Group the group and
Delivered the current delivery count. If the hooks fails, the
message is acknowledged using
XACK. From introduction to streams:
"So once the deliveries counter reaches a given large number that you chose, it is probably wiser to put such messages in another stream and send a notification to the system administrator. This is basically the way that Redis streams implement the concept of the dead letter."