redis_streams.pl -- Using Redis streams

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.

See also: - https://redis.io/topics/streams-intro

xstream_set(+Redis, +Key, +Option)

Set an option on for Key on Redis. Currently supports:

maxlen(+Count): Make xadd/4 add a MAXLEN ~ Count option to the XADD 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 details.

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))

Options:

count(+Count): Process at most Count messages per stream for each request.
start(+Start): Normally either 0 to start get all messages from the epoch or $ to get messages starting with the last. Default is $.
starts(+List): 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 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)]).

Arguments:

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 XREAD or 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 following differences:

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 XACK is sent to the server.

Options processed:

block(+Seconds)

Causes 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:

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.

max_deliveries(+Count)

Re-deliver (using XCLAIM) a message max Count times. Exceeding this calls xhook/2. Default Count is 3.

max_claim(+Count)

Do not claim more than Count messages during a single idle action. Default is 10.

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?

xconsumer_stop(+Leave)

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 events are:

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."