eredis_streams

%%%-------------------------------------------------------------------
%% @doc eredis_streams public API
%% @end
%%%-------------------------------------------------------------------

-module(eredis_streams).

-export(
  [
    start_stream/1,
    push_message/3,
    xack_message/2,
    xack_message/3,
    get_message_payload/1,
    get_message_id/1
  ]
).

start_stream(StreamDeclaration) -> eredis_streams_sup:add_stream(StreamDeclaration).

xack_message(Conn, Message) -> xack_message(Conn, Message, #{}).

xack_message(
  #{kind := eredis_streams, pool := Pool, stream := Stream, group := Group} = _Conn,
  Message,
  Opts
) ->
  AutoXdel = maps:get(no_xdel, Opts, false) == false,
  MsgId = get_message_id(Message),
  case {eredis_streams_db:xack_message(Pool, Stream, Group, MsgId), AutoXdel} of
    {ok, false} -> ok;

    {ok, true} ->
      case eredis_streams_db:xdel(Pool, Stream, MsgId) of
        {ok, _} -> ok;
        ErrXdel -> ErrXdel
      end;

    {ErrXack, _} -> ErrXack
  end.


get_message_payload({_MsgId, MsgFields} = _Message) ->
  proplists:get_value(<<"message">>, MsgFields);

get_message_payload(_) -> {error, wrong_message_format}.

get_message_id({MsgId, _MsgFields} = _Message) -> MsgId;
get_message_id(_) -> {error, wrong_message_format}.

push_message(Pool, Stream, Payload) -> eredis_streams_db:push_message(Pool, Stream, Payload).