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
- See also
- - https://redis.io/topics/streams-intro
- xstream_set(+Redis, +Key, +Option)
- Set an option on for Key on Redis. Currently supports:
- Make xadd/4 add a
MAXLEN ~ Count option to the
command, capping the length of the stream. See also
Redis as a message brokering system
- 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
- xlisten(+Redis, +Streams, +Options)
- Listen using
XREAD on 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
0 to start get all messages from the epoch
$ 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
key2 on 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($)]),
|Redis||- is either a Redis server name (see redis_server/3) or
an open connection. If it is a server name, a new connection is
opened that is closed if xlisten/3 completes.|
- See also
- - redis_subscribe/2 implements the classical pub/sub system of
Redis that does not have any memory.
- 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
XREADGROUP has 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
- block +Seconds
XREADGROUP to 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
- 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.
- 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
- xidle_group(+Redis, +Streams, +Starts, +NewStarts, +Options) is det[private]
- Called after
XREADGROUP returns and the returned messages (if
any) have been processed. If
Start == NewStarts no messages have
been processed, indicating a timeout.
This implementation looks for idle messages on other consumer and
will try to claim them.
- To be done
- - : max time to work on other consumers messages?
- 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.
- To be done
XGROUP DELCONSUMER only takes a single stream. Why?
- May be called from a consumer listener to stop the consumer. This
predicate throws the exception
redis(stop(Leave)), which is caught
- 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
"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."