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:
This library abstracts the latter two scenarios. The main predicates are
MAXLEN ~ Count
option to the XADD command, capping the length of the
stream. See also
Redis as a message brokering system (section
1.5)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.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))
Options:
0 to start get all messages from the epoch
or $ to get messages starting with the last. Default is $.
Note that this predicate does not terminate. It is normally
executed in a thread. The following call listens to the streams
key1 and 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($)]),
_, [detached(true)]).
| 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. |
XACK
is sent to the server.Options processed:
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
application. Notably:
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.XCLAIM) a message max Count
times. Exceeding this calls xhook/2.
Default Count is 3.10.redis(stop(Leave)), which is
caught by xlisten_group/5.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."