xlisten/3

Availability::- use_module(library(redis_streams)).(can be autoloaded)

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

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.