/* Part of SWI-Prolog Author: Jan Wielemaker E-mail: jan@swi-prolog.org WWW: http://www.swi-prolog.org Copyright (c) 2020, SWI-Prolog Solutions b.v. All rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. */ :- module(redis_streams, [ xstream_set/3, % +Redis, +Key, +Option xadd/4, % +Redis, +Key, ?Id, +Data:dict xlisten/3, % +Redis, +Streams, +Options xlisten_group/5, % +Redis, +Group, +Consumer, % +Streams, +Options xconsumer_stop/1 % +Leave ]). :- use_module(library(redis)). :- use_module(library(error)). :- use_module(library(apply)). :- use_module(library(broadcast)). :- use_module(library(lists)). :- use_module(library(option)). :- use_module(library(debug)). :- meta_predicate xlisten(+, +, 5, 5, +). :- multifile xhook/2. % +Stream, +Event :- predicate_options(xlisten/3, 3, [ count(nonneg), start(one_of([$,0])), starts(list) ]). :- predicate_options(xlisten_group/5, 5, [ block(number), max_deliveries(nonneg), max_claim(nonneg) ]). /** 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 https://redis.io/topics/streams-intro */ :- dynamic xstream_option/3. %! 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](#redis-brokering) xstream_set(Redis, KeyS, Option) :- must_be(atom, Redis), atom_string(Key, KeyS), valid_option(Option), functor(Option, Name, Arity), functor(Gen, Name, Arity), retractall(xstream_option(Redis, Key, Gen)), asserta(xstream_option(Redis, Key, Option)). valid_option(Option) :- stream_option(Option), !. valid_option(Option) :- domain_error(redis_stream_option, Option). stream_option(maxlen(X)) :- must_be(integer, X). %! 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. xadd(DB, Key, Id, Dict) :- redis_array_dict(Array, _, Dict), ( var(Id) -> IdIn = '*' ; IdIn = Id ), ( xstream_option(DB, Key, maxlen(MaxLen)) -> Command =.. [xadd, Key, maxlen, ~, MaxLen, IdIn|Array] ; Command =.. [xadd, Key, IdIn|Array] ), redis(DB, Command, Reply), return_id(Id, Reply). return_id(Id, Reply) :- var(Id), !, Id = Time-Seq, split_string(Reply, "-", "", [TimeS,SeqS]), number_string(Time, TimeS), number_string(Seq, SeqS). return_id(_, _). /******************************* * SUBSCRIBE * *******************************/ %! 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)]). % ``` % % @arg 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 redis_subscribe/2 implements the classical pub/sub system of % Redis that does not have any memory. xlisten(Redis, Streams, Options) :- xlisten(Redis, Streams, xbroadcast, xidle, Options). xbroadcast(Redis, Stream, Id, Dict, _Options) :- redis_id(Redis, RedisId), catch(broadcast(redis(RedisId, Stream, Id, Dict)), Error, print_message(error, Error)). redis_id(redis(Id, _, _), Id) :- !. redis_id(Id, Id). xidle(_Redis, _Streams, _Starts, _NewStarts, _Options). %! xlisten(+Redis, +Streams, +OnBroadCast, +OnIdle, +Options). % % 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. xlisten(Redis, Streams, OnBroadcast, OnIdle, Options) :- atom(Redis), !, setup_call_cleanup( redis_connect(Redis, Conn, [reconnect(true)|Options]), xlisten_(Conn, Streams, OnBroadcast, OnIdle, Options), redis_disconnect(Conn)). xlisten(Redis, Streams, OnBroadcast, OnIdle, Options) :- xlisten_(Redis, Streams, OnBroadcast, OnIdle, Options). xlisten_(Redis, Streams, OnBroadcast, OnIdle, Options) :- xread_command(Streams, Starts0, CommandTempl, Options), listen_loop(Redis, Starts0, CommandTempl, OnBroadcast, OnIdle, Streams, Options). xread_command(Streams, Starts0, Starts-Command, Options) :- option(group(Group-Consumer), Options), !, xread_command_args(Streams, Starts0, Starts, CmdArgs, Options), Command =.. [xreadgroup, group, Group, Consumer | CmdArgs]. xread_command(Streams, Starts0, Starts-Command, Options) :- xread_command_args(Streams, Starts0, Starts, CmdArgs, Options), Command =.. [xread|CmdArgs]. xread_command_args(Streams, Starts0, Starts, CmdArgs, Options) :- option(count(Count), Options), !, option(block(Block), Options, 0), BlockMS is integer(Block*1000), start_templ(Streams, Starts0, Starts, StreamArgs, Options), CmdArgs = [count, Count, block, BlockMS, streams | StreamArgs]. xread_command_args(Streams, Starts0, Starts, CmdArgs, Options) :- option(block(Block), Options, 0), BlockMS is integer(Block*1000), start_templ(Streams, Starts0, Starts, StreamArgs, Options), CmdArgs = [block, BlockMS, streams | StreamArgs]. start_templ(Streams, Starts0, Starts, StreamArgs, Options) :- option(starts(Starts0), Options), !, length(Streams, Len), length(Starts, Len), length(Starts0, LenS), ( LenS =:= Len -> true ; domain_error(xread_starts, Starts0) ), append(Streams, Starts, StreamArgs). start_templ(Streams, Starts0, Starts, StreamArgs, Options) :- option(start(Start), Options, $), !, length(Streams, Len), length(Starts, Len), length(Starts0, Len), maplist(=(Start), Starts0), append(Streams, Starts, StreamArgs). listen_loop(Redis, Starts, CommandTempl, OnBroadcast, OnIdle, Streams, Options) :- copy_term(CommandTempl, Starts-Command), ( redis(Redis, Command, Reply), Reply \== nil -> dispatch_streams(Reply, Redis, Streams, Starts, NewStarts, OnBroadcast, OnIdle, Options) ; NewStarts = Starts ), call(OnIdle, Redis, Streams, Starts, NewStarts, Options), listen_loop(Redis, NewStarts, CommandTempl, OnBroadcast, OnIdle, Streams, Options). dispatch_streams([], _, _, Starts, NewStarts, _, _, _) :- maplist(copy_start, Starts, NewStarts). dispatch_streams([Tuple|T], Redis, Streams, Starts, NewStarts, OnBroadcast, OnIdle, Options) :- stream_tuple(Tuple, StreamS, []), atom_string(Stream, StreamS), !, % xreadgroup: no more old pending stuff set_start(Stream, _Start, >, Streams, Starts, NewStarts), dispatch_streams(T, Redis, Streams, Starts, NewStarts, OnBroadcast, OnIdle, Options). dispatch_streams([Tuple|T], Redis, Streams, Starts, NewStarts, OnBroadcast, OnIdle, Options) :- stream_tuple(Tuple, StreamS, Messages), atom_string(Stream, StreamS), set_start(Stream, Start, NewStart, Streams, Starts, NewStarts), dispatch_messages(Messages, Stream, Redis, Start, NewStart, OnBroadcast, Options), dispatch_streams(T, Redis, Streams, Starts, NewStarts, OnBroadcast, OnIdle, Options). stream_tuple(Stream-Messages, Stream, Messages) :- !. stream_tuple([Stream,Messages], Stream, Messages). set_start(Stream, Old, New, [Stream|_], [Old|_], [New|_]) :- !. set_start(Stream, Old, New, [_|Streams], [_|OldStarts], [_|NewStarts]) :- set_start(Stream, Old, New, Streams, OldStarts, NewStarts). copy_start(Old, New) :- ( var(New) -> Old = New ; true ). %! dispatch_messages(+Messages, +Stream, +Redis, +Start0, -Start) is det dispatch_messages([], _, _, Start, Start, _, _). dispatch_messages([[Start,Data]|T], Stream, Redis, Start0, NewStart, OnBroadcast, Options) :- dispatch_message(Data, Stream, Redis, Start, OnBroadcast, Options), join_starts(Start0, Start, Start1), dispatch_messages(T, Stream, Redis, Start1, NewStart, OnBroadcast, Options). dispatch_message(Data, Stream, Redis, Id, OnBroadcast, Options) :- ( Data == nil % when does this happen? -> Dict = redis{} ; redis_array_dict(Data, redis, Dict) ), call(OnBroadcast, Redis, Stream, Id, Dict, Options). join_starts(>, _Start, >) :- !. join_starts(_Start0, Start, Start). /******************************* * CONSUMERS * *******************************/ %! 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`. xlisten_group(Redis, Group, Consumer, Streams, Options) :- catch(xlisten(Redis, Streams, xbroadcast_group, xidle_group, [ group(Group-Consumer), start(0) | Options ]), redis(stop(Leave)), true), ( Leave == true -> xleave_group(Redis, Group, Consumer, Streams) ; true ). xbroadcast_group(Connection, Stream, Id, Dict, Options) :- option(group(Group-Consumer), Options), redis_id(Connection, RedisId), ( catch(broadcast_request(redis_consume(Stream, Dict, _{redis:RedisId, message:Id, group:Group, consumer:Consumer})), Error, xbroadcast_error(Error, Connection, Stream, Group, Id)) -> redis(Connection, xack(Stream, Group, Id)) ; true ). xbroadcast_error(redis(stop(Unregister)), Connection, Stream, Group, Id) :- !, redis(Connection, xack(Stream, Group, Id), _), throw(redis(stop(Unregister))). xbroadcast_error(Error, _Connection, _Stream, _Group, _Id) :- print_message(error, Error), fail. %! xidle_group(+Redis, +Streams, +Starts, +NewStarts, +Options) is det. % % 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. % % @tbd: max time to work on other consumers messages? xidle_group(Redis, Streams, Starts, Starts, Options) :- % Idle time !, option(group(Group-_Consumer), Options), claim(Streams, Redis, Group, 0, _Claimed, Options). xidle_group(_Redis, _Streams, _Starts, _NewStarts, _Options). claim([], _, _, Claimed, Claimed, _). claim([Stream|ST], Redis, Group, Claimed0, Claimed, Options) :- claim_for_stream(Redis, Stream, Group, Claimed0, Claimed1, Options), claim(ST, Redis, Group, Claimed1, Claimed, Options). claim_for_stream(Redis, Stream, Group, Claimed0, Claimed, Options) :- check_limit_claimed(Claimed0, Options), redis(Redis, xpending(Stream, Group), [Count,_Oldest,_Newest, Cons]), Count > 0, !, debug(redis(claim), '~D pending messages on ~p for ~p (Consumers = ~p)', [Count, Stream, Group, Cons]), claim_for_stream_for_consumers(Cons, Redis, Stream, Group, Claimed0, Claimed, Options). claim_for_stream(_Redis, _Stream, _Group, Claimed, Claimed, _Options). claim_for_stream_for_consumers([], _Redis, _Stream, _Group, Claimed, Claimed, _Options). claim_for_stream_for_consumers([C|T], Redis, Stream, Group, Claimed0, Claimed, Options) :- claim_for_stream_for_consumer(Redis, Stream, Group, C, Claimed0, Claimed1, Options), claim_for_stream_for_consumers(T, Redis, Stream, Group, Claimed1, Claimed, Options). claim_for_stream_for_consumer(Redis, Stream, Group, [Consumer,_Pending], Claimed0, Claimed, Options) :- redis(Redis, xpending(Stream, Group, -, +, 10, Consumer), Reply), claim_messages(Reply, Redis, Stream, Group, Claimed0, Claimed, Options). claim_messages([], _Redis, _Stream, _Group, Claimed, Claimed, _Options). claim_messages([H|T], Redis, Stream, Group, Claimed0, Claimed, Options) :- debug(redis(claim), 'Found pending ~p', [H]), claim_message(H, Redis, Stream, Group, Claimed0, Claimed1, Options), claim_messages(T, Redis, Stream, Group, Claimed1, Claimed, Options). claim_message([Id,Consumer,Idle,Delivered], Redis, Stream, Group, Claimed0, Claimed, Options) :- option(block(Block), Options), BlockMS is integer(Block*1000), Idle > BlockMS, check_limit_deliveries(Redis, Stream, Delivered, Id, Options), check_limit_claimed(Claimed0, Options), option(group(Group-Me), Options), debug(redis(claim), '~p: trying to claim ~p from ~p (idle ~D)', [Me, Id, Consumer, Idle]), redis(Redis, xclaim(Stream, Group, Me, BlockMS, Id), ClaimedMsgs), !, Claimed is Claimed0 + 1, debug(redis(claimed), '~p: claimed ~p', [Me, ClaimedMsgs]), dispatch_claimed(ClaimedMsgs, Redis, Stream, Options). claim_message(_Msg, _Redis, _Stream, _Group, Claimed, Claimed, _Options). dispatch_claimed([], _Redis, _Stream, _Options). dispatch_claimed([[MsgId,MsgArray]|Msgs], Redis, Stream, Options) :- redis_array_dict(MsgArray, _, Dict), xbroadcast_group(Redis, Stream, MsgId, Dict, Options), dispatch_claimed(Msgs, Redis, Stream, Options). %! check_limit_deliveries(+Redis, +Stream, +Delivered, +Id, +Options) % % 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. check_limit_deliveries(_Redis, _Stream, Delivered, _Id, Options) :- option(max_deliveries(Max), Options, 3), Delivered =< Max, !. check_limit_deliveries(Redis, Stream, Delivered, Id, Options) :- option(group(Group-_Me), Options), ( xhook(Stream, delivery_failed(Id,Group,Delivered)) -> true ; print_message(warning, redis(delivery_failed(Id,Group,Delivered))), redis(Redis, xack(Stream, Group, Id)) ), fail. check_limit_claimed(Claimed0, Options) :- option(max_claim(Max), Options, 10), Claimed0 < Max. %! xleave_group(+Redis, +Group, +Consumer, +Streams) is det. % % Remove Consumer from Group. % % @tbd ``XGROUP DELCONSUMER`` only takes a single stream. Why? xleave_group(Redis, Group, Consumer, [Stream|_]) :- redis(Redis, xgroup(delconsumer, Stream, Group, Consumer), _). %! 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. xconsumer_stop(Leave) :- throw(redis(stop(Leave))). /******************************* * HOOKS * *******************************/ %! xhook(+Stream, +Event) % % 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](https://redis.io/topics/streams-intro): % % > "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." /******************************* * MESSAGES * *******************************/ :- multifile prolog:message//1. prolog:message(redis(Message)) --> [ 'REDIS: '-[] ], redis_message(Message). redis_message(delivery_failed(Id,Group,Delivered)) --> [ 'Failed to deliver ~p to group ~p (tried ~D times)'- [Id, Group, Delivered] ].