-module(glats@handler). -compile([no_auto_import, nowarn_unused_vars]). -export([handle_subscription/3]). -export_type([state/0]). -type state() :: {subscription_state, gleam@erlang@process:subject(glats@connection:command()), fun((glats@message:message(), gleam@erlang@process:subject(glats@connection:command())) -> {ok, nil} | {error, binary()})}. -spec map_message(gleam@dynamic:dynamic()) -> glats@message:message(). map_message(Msg) -> _pipe = Msg, _pipe@1 = 'Elixir.Glats':convert_msg(_pipe), gleam@result:unwrap( _pipe@1, {message, <<"error"/utf8>>, gleam@map:new(), none, <<"body"/utf8>>} ). -spec handle_subscription( gleam@erlang@process:subject(glats@connection:command()), binary(), fun((glats@message:message(), gleam@erlang@process:subject(glats@connection:command())) -> {ok, nil} | {error, binary()}) ) -> {ok, gleam@erlang@process:subject(glats@message:message())} | {error, gleam@otp@actor:start_error()}. handle_subscription(Conn, Subject, Handler) -> gleam@otp@actor:start_spec( {spec, fun() -> Receiver = gleam@erlang@process:new_subject(), Selector = begin _pipe = gleam_erlang_ffi:new_selector(), gleam@erlang@process:selecting_anything( _pipe, fun map_message/1 ) end, case glats@connection:subscribe(Conn, Receiver, Subject) of {ok, _} -> {ready, {subscription_state, Conn, Handler}, Selector}; {error, Err} -> {failed, Err} end end, 10000, fun(Msg, State) -> case (erlang:element(3, State))(Msg, erlang:element(2, State)) of {ok, _} -> {continue, State}; {error, _} -> {stop, {abnormal, <<"handler returned error!"/utf8>>}} end end} ).