eredis_streams

-module(eredis_streams_test_handler).

-behaviour(eredis_stream_handler).

-include_lib("eredis_streams/include/telemetry.hrl").

-export([init/1, handle_messages/3, handle_failover/3, handle_info/2]).

init(_Params) -> ok.

handle_messages(Conn, Messages, Config) ->
  _ = do(Conn, Messages, Config),
  #{processed => length(Messages)}.


handle_failover(Conn, Messages, Config) ->
  _ = do(Conn, Messages, Config),
  #{processed => length(Messages)}.


handle_info(_Info, State) -> {noreply, State}.

%%%===================================================================
%%% helpers
%%%===================================================================

do(_, [], _) -> ok;

do(Conn, [Message | Messages], #{outstream := #{name := OutStream, pool := OutPool}} = Config) ->
  % 1. get message payload
  Payload = eredis_streams:get_message_payload(Message),
  % 2. push message to out stream
  {ok, _} = eredis_streams:push_message(OutPool, OutStream, Payload),
  % 3. xack message with auto xdel
  ok = eredis_streams:xack_message(Conn, Message),
  % 4. emit telemetry event for each message (optional)
  ?EXEC_TELEMETRY(
    ?EREDIS_STREAMS_WORKER_HANDLE_MESSAGE,
    #{},
    Conn#{message => Message, config => Config}
  ),
  do(Conn, Messages, Config).