-module(spoke@core). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/spoke/core.gleam"). -export([restore_state/2, new_state/1, handle_input/3]). -export_type([command/0, timed_action/0, transport_event/0, input/0, output/0, state/0, options/0, connection_state/0, pending_subscription/0, pending_unsubscribe/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. -type command() :: {subscribe_to_updates, drift:effect(spoke@mqtt:update())} | {unsubscribe_from_updates, drift:effect(spoke@mqtt:update())} | {connect, boolean(), gleam@option:option(spoke@mqtt:publish_data())} | {disconnect, drift:effect(nil)} | {subscribe, list(spoke@mqtt:subscribe_request()), drift:effect({ok, list(spoke@mqtt:subscription())} | {error, spoke@mqtt:operation_error()})} | {unsubscribe, list(binary()), drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()})} | {publish_message, spoke@mqtt:publish_data()} | {get_pending_publishes, drift:effect(integer())} | {wait_for_publishes_to_finish, drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}), integer()}. -opaque timed_action() :: send_ping | ping_resp_timed_out | connect_timed_out | {subscribe_timed_out, integer()} | {unsubscribe_timed_out, integer()} | {wait_for_publishes_timeout, drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()})}. -type transport_event() :: transport_established | {transport_failed, binary()} | transport_closed | {received_data, bitstring()}. -type input() :: {perform, command()} | {handle, transport_event()} | {timeout, timed_action()}. -type output() :: {publish, drift:action(spoke@mqtt:update())} | open_transport | close_transport | {send_data, gleam@bytes_tree:bytes_tree()} | {return_pending_publishes, drift:action(integer())} | {publishes_completed, drift:action({ok, nil} | {error, spoke@mqtt:operation_error()})} | {subscribe_completed, drift:action({ok, list(spoke@mqtt:subscription())} | {error, spoke@mqtt:operation_error()})} | {unsubscribe_completed, drift:action({ok, nil} | {error, spoke@mqtt:operation_error()})} | {complete_disconnect, drift:action(nil)} | {update_persisted_session, spoke@core@session_state:storage_update()}. -opaque state() :: {state, options(), spoke@core@internal@session:session(), connection_state(), gleam@dict:dict(integer(), pending_subscription()), gleam@dict:dict(integer(), pending_unsubscribe()), gleam@option:option(drift@internal@timer:timer()), gleam@option:option(drift@internal@timer:timer()), gleam@option:option(drift@internal@timer:timer()), gleam@set:set(drift:effect(spoke@mqtt:update())), gleam@dict:dict(drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}), drift@internal@timer:timer())}. -type options() :: {options, binary(), gleam@option:option(spoke@packet:auth_options()), integer(), integer()}. -type connection_state() :: not_connected | {connecting, spoke@packet:connect_options()} | {waiting_for_conn_ack, spoke@core@internal@connection:connection()} | {connected, spoke@core@internal@connection:connection()} | disconnecting. -type pending_subscription() :: {pending_subscription, list(spoke@mqtt:subscribe_request()), drift:effect({ok, list(spoke@mqtt:subscription())} | {error, spoke@mqtt:operation_error()}), drift@internal@timer:timer()}. -type pending_unsubscribe() :: {pending_unsubscribe, drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}), drift@internal@timer:timer()}. -file("src/spoke/core.gleam", 176). -spec new( spoke@core@internal@session:session(), spoke@mqtt:connect_options(any()) ) -> state(). new(Session, Options) -> Options@1 = {options, erlang:element(3, Options), spoke@core@internal@convert:to_auth_options(erlang:element(4, Options)), erlang:element(5, Options) * 1000, erlang:element(6, Options)}, {state, Options@1, Session, not_connected, maps:new(), maps:new(), none, none, none, gleam@set:new(), maps:new()}. -file("src/spoke/core.gleam", 128). ?DOC(" Creates a new state from an existing session.\n"). -spec restore_state( spoke@mqtt:connect_options(any()), spoke@core@session_state:session_state() ) -> state(). restore_state(Options, State) -> _pipe = spoke@core@internal@session:from_state(State), new(_pipe, Options). -file("src/spoke/core.gleam", 137). ?DOC(" Creates new, clean state.\n"). -spec new_state(spoke@mqtt:connect_options(any())) -> state(). new_state(Options) -> _pipe = spoke@core@internal@session:new(false), new(_pipe, Options). -file("src/spoke/core.gleam", 283). -spec wait_for_publishes( drift:context(input(), output()), state(), drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}), integer() ) -> drift:step(state(), input(), output(), binary()). wait_for_publishes(Context, State, Complete, Timeout) -> case spoke@core@internal@session:pending_publishes(erlang:element(3, State)) of 0 -> _pipe = Context, _pipe@1 = drift:perform( _pipe, fun(Field@0) -> {publishes_completed, Field@0} end, Complete, {ok, nil} ), drift:continue(_pipe@1, State); _ -> {Context@1, Timer} = drift:start_timer( Context, Timeout, {timeout, {wait_for_publishes_timeout, Complete}} ), Publish_completion_listeners = gleam@dict:insert( erlang:element(11, State), Complete, Timer ), drift:continue( Context@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), Publish_completion_listeners} ) end. -file("src/spoke/core.gleam", 364). -spec time_out_wait_for_publish( drift:context(input(), output()), state(), drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}) ) -> drift:step(state(), input(), output(), binary()). time_out_wait_for_publish(Context, State, Complete) -> Publish_completion_listeners = gleam@dict:delete( erlang:element(11, State), Complete ), _pipe = Context, _pipe@1 = drift:perform( _pipe, fun(Field@0) -> {publishes_completed, Field@0} end, Complete, {error, operation_timed_out} ), drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), Publish_completion_listeners} ). -file("src/spoke/core.gleam", 419). -spec subscribe_to_updates( drift:context(input(), output()), state(), drift:effect(spoke@mqtt:update()) ) -> drift:step(state(), input(), output(), binary()). subscribe_to_updates(Context, State, Publish) -> Update_listeners = gleam@set:insert(erlang:element(10, State), Publish), drift:continue( Context, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), Update_listeners, erlang:element(11, State)} ). -file("src/spoke/core.gleam", 428). -spec unsubscribe_from_updates( drift:context(input(), output()), state(), drift:effect(spoke@mqtt:update()) ) -> drift:step(state(), input(), output(), binary()). unsubscribe_from_updates(Context, State, Publish) -> Update_listeners = gleam@set:delete(erlang:element(10, State), Publish), drift:continue( Context, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), Update_listeners, erlang:element(11, State)} ). -file("src/spoke/core.gleam", 538). -spec get_pending_publishes( drift:context(input(), output()), state(), drift:effect(integer()) ) -> drift:step(state(), input(), output(), binary()). get_pending_publishes(Context, State, Complete) -> _pipe = Context, _pipe@1 = drift:perform( _pipe, fun(Field@0) -> {return_pending_publishes, Field@0} end, Complete, spoke@core@internal@session:pending_publishes(erlang:element(3, State)) ), drift:continue(_pipe@1, State). -file("src/spoke/core.gleam", 913). -spec check_publish_completion_with( drift:context(input(), output()), state(), {ok, nil} | {error, spoke@mqtt:operation_error()} ) -> drift:step(state(), input(), output(), binary()). check_publish_completion_with(Context, State, Result) -> case spoke@core@internal@session:pending_publishes(erlang:element(3, State)) of 0 -> _pipe@1 = begin gleam@dict:fold( erlang:element(11, State), Context, fun(Context@1, Complete, Timer) -> _pipe = erlang:element( 1, drift:cancel_timer(Context@1, Timer) ), drift:perform( _pipe, fun(Field@0) -> {publishes_completed, Field@0} end, Complete, Result ) end ) end, drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), maps:new()} ); _ -> drift:continue(Context, State) end. -file("src/spoke/core.gleam", 905). -spec check_publish_completion(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). check_publish_completion(Context, State) -> check_publish_completion_with(Context, State, {ok, nil}). -file("src/spoke/core.gleam", 909). -spec reset_publish_completion(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). reset_publish_completion(Context, State) -> check_publish_completion_with(Context, State, {error, session_reset}). -file("src/spoke/core.gleam", 973). -spec send(spoke@packet@client@outgoing:packet()) -> output(). send(Packet) -> {send_data, spoke@packet@client@outgoing:encode_packet(Packet)}. -file("src/spoke/core.gleam", 482). -spec disconnect(drift:context(input(), output()), state(), drift:effect(nil)) -> drift:step(state(), input(), output(), binary()). disconnect(Context, State, Complete) -> Outputs = case erlang:element(4, State) of {connecting, _} -> {some, [close_transport]}; {waiting_for_conn_ack, _} -> {some, [close_transport]}; {connected, _} -> {some, [send(disconnect), close_transport]}; disconnecting -> none; not_connected -> none end, Result = {complete_disconnect, drift:bind_effect(Complete, nil)}, case Outputs of {some, Outputs@1} -> _pipe = Context, _pipe@1 = drift:output_many(_pipe, Outputs@1), _pipe@2 = drift:output(_pipe@1, Result), drift:continue( _pipe@2, {state, erlang:element(2, State), erlang:element(3, State), disconnecting, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ); none -> _pipe@3 = Context, _pipe@4 = drift:output(_pipe@3, Result), drift:continue(_pipe@4, State) end. -file("src/spoke/core.gleam", 552). -spec transport_established(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). transport_established(Context, State) -> case erlang:element(4, State) of {connecting, Options} -> Connection = {waiting_for_conn_ack, spoke@core@internal@connection:new()}, _pipe = Context, _pipe@1 = drift:output(_pipe, send({connect, Options})), drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), Connection, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ); _ -> drift:stop_with_error( Context, <<"Unexpected connection state when establishing transport: "/utf8, (case erlang:element(4, State) of {connected, _} -> <<"Connected"/utf8>>; {connecting, _} -> <<"Connecting"/utf8>>; disconnecting -> <<"Disconnecting"/utf8>>; not_connected -> <<"Not connected"/utf8>>; {waiting_for_conn_ack, _} -> <<"Waiting for CONNACK"/utf8>> end)/binary>> ) end. -file("src/spoke/core.gleam", 977). -spec publish_update( drift:context(input(), output()), state(), spoke@mqtt:update() ) -> drift:context(input(), output()). publish_update(Context, State, Update) -> Updates = begin gleam@list:map( gleam@set:to_list(erlang:element(10, State)), fun(Listener) -> {publish, drift:bind_effect(Listener, Update)} end ) end, drift:output_many(Context, Updates). -file("src/spoke/core.gleam", 575). -spec transport_closed(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). transport_closed(Context, State) -> Change = case erlang:element(4, State) of disconnecting -> disconnected; _ -> {disconnected_unexpectedly, <<"Transport closed"/utf8>>} end, _pipe = Context, _pipe@1 = drift:cancel_all_timers(_pipe), _pipe@2 = publish_update(_pipe@1, State, {connection_state_changed, Change}), drift:continue( _pipe@2, {state, erlang:element(2, State), erlang:element(3, State), not_connected, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 587). -spec disconnect_unexpectedly( drift:context(input(), output()), state(), binary() ) -> drift:step(state(), input(), output(), binary()). disconnect_unexpectedly(Context, State, Error) -> Change = case erlang:element(4, State) of not_connected -> none; {connected, _} -> {some, {disconnected_unexpectedly, Error}}; disconnecting -> {some, {disconnected_unexpectedly, Error}}; {connecting, _} -> {some, {connect_failed, Error}}; {waiting_for_conn_ack, _} -> {some, {connect_failed, Error}} end, Context@1 = case Change of none -> Context; {some, Change@1} -> _pipe = Context, _pipe@1 = drift:output(_pipe, close_transport), publish_update(_pipe@1, State, {connection_state_changed, Change@1}) end, _pipe@2 = Context@1, _pipe@3 = drift:cancel_all_timers(_pipe@2), drift:continue( _pipe@3, {state, erlang:element(2, State), erlang:element(3, State), not_connected, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 937). -spec kill_connection(drift:context(input(), output()), state(), binary()) -> drift:step(state(), input(), output(), binary()). kill_connection(Context, State, Error) -> Sub_errors = begin gleam@list:map( maps:values(erlang:element(5, State)), fun(Pending_sub) -> {subscribe_completed, drift:bind_effect( erlang:element(3, Pending_sub), {error, protocol_violation} )} end ) end, Unsub_errors = begin gleam@list:map( maps:values(erlang:element(6, State)), fun(Pending_unsub) -> {unsubscribe_completed, drift:bind_effect( erlang:element(2, Pending_unsub), {error, protocol_violation} )} end ) end, _pipe = Context, _pipe@1 = drift:cancel_all_timers(_pipe), _pipe@2 = drift:output_many(_pipe@1, Sub_errors), _pipe@3 = drift:output_many(_pipe@2, Unsub_errors), _pipe@4 = drift:output(_pipe@3, close_transport), _pipe@5 = publish_update( _pipe@4, State, {connection_state_changed, {disconnected_unexpectedly, Error}} ), drift:continue( _pipe@5, {state, erlang:element(2, State), erlang:element(3, State), not_connected, maps:new(), maps:new(), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 377). -spec time_out_subscription( drift:context(input(), output()), state(), integer() ) -> drift:step(state(), input(), output(), binary()). time_out_subscription(Context, State, Id) -> Pending_subs = erlang:element(5, State), case gleam_stdlib:map_get(Pending_subs, Id) of {error, _} -> drift:continue(Context, State); {ok, {pending_subscription, _, Complete, _}} -> kill_connection( drift:perform( Context, fun(Field@0) -> {subscribe_completed, Field@0} end, Complete, {error, operation_timed_out} ), {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), gleam@dict:delete(Pending_subs, Id), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)}, <<"Subscribe timed out"/utf8>> ) end. -file("src/spoke/core.gleam", 398). -spec time_out_unsubscribe(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). time_out_unsubscribe(Context, State, Id) -> Pending_unsubs = erlang:element(6, State), case gleam_stdlib:map_get(Pending_unsubs, Id) of {error, _} -> drift:continue(Context, State); {ok, {pending_unsubscribe, Complete, _}} -> kill_connection( drift:perform( Context, fun(Field@0) -> {unsubscribe_completed, Field@0} end, Complete, {error, operation_timed_out} ), {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), gleam@dict:delete(Pending_unsubs, Id), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)}, <<"Unsubscribe timed out"/utf8>> ) end. -file("src/spoke/core.gleam", 791). -spec handle_suback( drift:context(input(), output()), state(), integer(), list({ok, spoke@packet:qo_s()} | {error, nil}) ) -> drift:step(state(), input(), output(), binary()). handle_suback(Context, State, Id, Return_codes) -> Result = begin Subs = erlang:element(5, State), gleam@result:'try'( gleam@result:replace_error( gleam_stdlib:map_get(Subs, Id), <<"Received invalid packet id in subscribe ack"/utf8>> ), fun(Pending_sub) -> gleam@result:'try'( gleam@result:replace_error( gleam@list:strict_zip( erlang:element(2, Pending_sub), Return_codes ), <<"Received invalid number of results in subscribe ack"/utf8>> ), fun(Pairs) -> Results = gleam@list:map( Pairs, fun spoke@core@internal@convert:to_subscription/1 ), {ok, begin _pipe = erlang:element( 1, drift:cancel_timer( Context, erlang:element(4, Pending_sub) ) ), _pipe@1 = drift:perform( _pipe, fun(Field@0) -> {subscribe_completed, Field@0} end, erlang:element(3, Pending_sub), {ok, Results} ), drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), gleam@dict:delete(Subs, Id), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ) end} end ) end ) end, case Result of {error, E} -> kill_connection(Context, State, E); {ok, Step} -> Step end. -file("src/spoke/core.gleam", 823). -spec handle_unsuback(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). handle_unsuback(Context, State, Id) -> Unsubs = erlang:element(6, State), case gleam_stdlib:map_get(Unsubs, Id) of {ok, Pending_unsub} -> {Context@1, _} = drift:cancel_timer( Context, erlang:element(3, Pending_unsub) ), _pipe = Context@1, _pipe@1 = drift:perform( _pipe, fun(Field@0) -> {unsubscribe_completed, Field@0} end, erlang:element(2, Pending_unsub), {ok, nil} ), drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), gleam@dict:delete(Unsubs, Id), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ); {error, _} -> kill_connection( Context, State, <<"Received invalid packet id in unsubscribe ack"/utf8>> ) end. -file("src/spoke/core.gleam", 989). -spec maybe_cancel_timer( drift:context(input(), output()), gleam@option:option(drift@internal@timer:timer()) ) -> drift:context(input(), output()). maybe_cancel_timer(Context, Timer) -> case Timer of {some, Timer@1} -> erlang:element(1, drift:cancel_timer(Context, Timer@1)); none -> Context end. -file("src/spoke/core.gleam", 686). -spec handle_ping_resp(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). handle_ping_resp(Context, State) -> _pipe = Context, _pipe@1 = maybe_cancel_timer(_pipe, erlang:element(8, State)), drift:continue( _pipe@1, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), none, erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 883). -spec start_send_ping_timer(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). start_send_ping_timer(Context, State) -> Context@1 = maybe_cancel_timer(Context, erlang:element(7, State)), {Context@2, Timer} = drift:start_timer( Context@1, erlang:element(4, erlang:element(2, State)), {timeout, send_ping} ), drift:continue( Context@2, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), {some, Timer}, erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 692). -spec handle_first_packet( drift:context(input(), output()), state(), spoke@core@internal@connection:connection(), spoke@packet@client@incoming:packet() ) -> drift:step(state(), input(), output(), binary()). handle_first_packet(Context, State, Connection, Packet) -> case Packet of {conn_ack, Result} -> Update = {connection_state_changed, spoke@core@internal@convert:to_connection_state(Result)}, case Result of {ok, _} -> _pipe = Context, _pipe@1 = publish_update(_pipe, State, Update), _pipe@4 = drift:output_many( _pipe@1, begin _pipe@2 = erlang:element(3, State), _pipe@3 = spoke@core@internal@session:packets_to_send_after_connect( _pipe@2 ), gleam@list:map(_pipe@3, fun send/1) end ), _pipe@5 = maybe_cancel_timer( _pipe@4, erlang:element(9, State) ), _pipe@6 = start_send_ping_timer( _pipe@5, {state, erlang:element(2, State), erlang:element(3, State), {connected, Connection}, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@6, fun reset_publish_completion/2); {error, _} -> _pipe@7 = Context, _pipe@8 = publish_update(_pipe@7, State, Update), _pipe@9 = drift:output(_pipe@8, close_transport), _pipe@10 = drift:cancel_all_timers(_pipe@9), drift:continue( _pipe@10, {state, erlang:element(2, State), erlang:element(3, State), not_connected, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ) end; _ -> kill_connection( Context, State, <<"The first packet sent from the Server to the Client MUST be a CONNACK Packet"/utf8>> ) end. -file("src/spoke/core.gleam", 892). -spec start_ping_timeout_timer(drift:context(input(), output()), state()) -> drift:step(state(), input(), output(), binary()). start_ping_timeout_timer(Context, State) -> Context@1 = maybe_cancel_timer(Context, erlang:element(8, State)), {Context@2, Timer} = drift:start_timer( Context@1, erlang:element(5, erlang:element(2, State)), {timeout, ping_resp_timed_out} ), drift:continue( Context@2, {state, erlang:element(2, State), erlang:element(3, State), erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), {some, Timer}, erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ). -file("src/spoke/core.gleam", 338). -spec handle_timer(drift:context(input(), output()), state(), timed_action()) -> drift:step(state(), input(), output(), binary()). handle_timer(Context, State, Action) -> case Action of send_ping -> case erlang:element(4, State) of {connected, _} -> _pipe = Context, _pipe@1 = drift:output(_pipe, send(ping_req)), _pipe@2 = start_ping_timeout_timer(_pipe@1, State), drift:chain(_pipe@2, fun start_send_ping_timer/2); _ -> drift:continue(Context, State) end; ping_resp_timed_out -> disconnect_unexpectedly( Context, State, <<"Ping response timed out"/utf8>> ); connect_timed_out -> disconnect_unexpectedly( Context, State, <<"Connecting timed out"/utf8>> ); {wait_for_publishes_timeout, Ref} -> time_out_wait_for_publish(Context, State, Ref); {subscribe_timed_out, Id} -> time_out_subscription(Context, State, Id); {unsubscribe_timed_out, Id@1} -> time_out_unsubscribe(Context, State, Id@1) end. -file("src/spoke/core.gleam", 996). -spec output_storage_updates( drift:context(input(), output()), list(spoke@core@session_state:storage_update()) ) -> drift:context(input(), output()). output_storage_updates(Context, Updates) -> drift:output_many( Context, gleam@list:map( Updates, fun(Field@0) -> {update_persisted_session, Field@0} end ) ). -file("src/spoke/core.gleam", 198). -spec subscribe( drift:context(input(), output()), state(), list(spoke@mqtt:subscribe_request()), drift:effect({ok, list(spoke@mqtt:subscription())} | {error, spoke@mqtt:operation_error()}) ) -> drift:step(state(), input(), output(), binary()). subscribe(Context, State, All_requests, Complete) -> case {All_requests, erlang:element(4, State)} of {[Request | Requests], {connected, _}} -> {Session, Id, Id_updates} = spoke@core@internal@session:reserve_packet_id( erlang:element(3, State) ), {Context@1, Timer} = drift:start_timer( Context, erlang:element(5, erlang:element(2, State)), {timeout, {subscribe_timed_out, Id}} ), Pending_subs = gleam@dict:insert( erlang:element(5, State), Id, {pending_subscription, All_requests, Complete, Timer} ), _pipe = Context@1, _pipe@1 = drift:output( _pipe, send( {subscribe, Id, spoke@core@internal@convert:to_packet_subscribe_request( Request ), gleam@list:map( Requests, fun spoke@core@internal@convert:to_packet_subscribe_request/1 )} ) ), _pipe@2 = output_storage_updates(_pipe@1, Id_updates), _pipe@3 = drift:continue( _pipe@2, {state, erlang:element(2, State), Session, erlang:element(4, State), Pending_subs, erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@3, fun start_send_ping_timer/2); {[], {connected, _}} -> _pipe@4 = Context, _pipe@5 = drift:perform( _pipe@4, fun(Field@0) -> {subscribe_completed, Field@0} end, Complete, {ok, []} ), drift:continue(_pipe@5, State); {_, _} -> _pipe@6 = Context, _pipe@7 = drift:perform( _pipe@6, fun(Field@0) -> {subscribe_completed, Field@0} end, Complete, {error, not_connected} ), drift:continue(_pipe@7, State) end. -file("src/spoke/core.gleam", 246). -spec unsubscribe( drift:context(input(), output()), state(), list(binary()), drift:effect({ok, nil} | {error, spoke@mqtt:operation_error()}) ) -> drift:step(state(), input(), output(), binary()). unsubscribe(Context, State, Topics, Complete) -> case {Topics, erlang:element(4, State)} of {[Topic | Topics@1], {connected, _}} -> {Session, Id, Id_updates} = spoke@core@internal@session:reserve_packet_id( erlang:element(3, State) ), {Context@1, Timer} = drift:start_timer( Context, erlang:element(5, erlang:element(2, State)), {timeout, {unsubscribe_timed_out, Id}} ), Pending_unsubs = begin _pipe = erlang:element(6, State), gleam@dict:insert( _pipe, Id, {pending_unsubscribe, Complete, Timer} ) end, _pipe@1 = Context@1, _pipe@2 = drift:output( _pipe@1, send({unsubscribe, Id, Topic, Topics@1}) ), _pipe@3 = output_storage_updates(_pipe@2, Id_updates), _pipe@4 = drift:continue( _pipe@3, {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), Pending_unsubs, erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@4, fun start_send_ping_timer/2); {[], {connected, _}} -> _pipe@5 = Context, _pipe@6 = drift:perform( _pipe@5, fun(Field@0) -> {unsubscribe_completed, Field@0} end, Complete, {ok, nil} ), drift:continue(_pipe@6, State); {_, _} -> _pipe@7 = Context, _pipe@8 = drift:perform( _pipe@7, fun(Field@0) -> {unsubscribe_completed, Field@0} end, Complete, {error, not_connected} ), drift:continue(_pipe@8, State) end. -file("src/spoke/core.gleam", 437). -spec connect( drift:context(input(), output()), state(), boolean(), gleam@option:option(spoke@mqtt:publish_data()) ) -> drift:step(state(), input(), output(), binary()). connect(Context, State, Clean_session, Will) -> case erlang:element(4, State) of not_connected -> {Session, Storage_updates} = spoke@core@internal@session:connect( erlang:element(3, State), Clean_session ), Options = {connect_options, Clean_session, erlang:element(2, erlang:element(2, State)), erlang:element(4, erlang:element(2, State)) div 1000, erlang:element(3, erlang:element(2, State)), gleam@option:map( Will, fun spoke@core@internal@convert:to_will/1 )}, {Context@1, Timer} = drift:start_timer( Context, erlang:element(5, erlang:element(2, State)), {timeout, connect_timed_out} ), _pipe = Context@1, _pipe@1 = output_storage_updates(_pipe, Storage_updates), _pipe@2 = drift:output(_pipe@1, open_transport), drift:continue( _pipe@2, {state, erlang:element(2, State), Session, {connecting, Options}, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), {some, Timer}, erlang:element(10, State), erlang:element(11, State)} ); _ -> drift:continue(Context, State) end. -file("src/spoke/core.gleam", 506). -spec publish( drift:context(input(), output()), state(), spoke@mqtt:publish_data() ) -> drift:step(state(), input(), output(), binary()). publish(Context, State, Data) -> Message = {message_data, erlang:element(2, Data), erlang:element(3, Data), erlang:element(5, Data)}, {Session, Packet, Storage_updates} = case erlang:element(4, Data) of at_most_once -> {erlang:element(3, State), {publish, {publish_data_qo_s0, Message}}, []}; at_least_once -> spoke@core@internal@session:start_qos1_publish( erlang:element(3, State), Message ); exactly_once -> spoke@core@internal@session:start_qos2_publish( erlang:element(3, State), Message ) end, State@1 = {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)}, case erlang:element(4, State@1) of {connected, _} -> _pipe = Context, _pipe@1 = drift:output(_pipe, send(Packet)), _pipe@2 = output_storage_updates(_pipe@1, Storage_updates), _pipe@3 = drift:continue(_pipe@2, State@1), drift:chain(_pipe@3, fun start_send_ping_timer/2); _ -> drift:continue(Context, State@1) end. -file("src/spoke/core.gleam", 738). -spec handle_publish( drift:context(input(), output()), state(), spoke@packet:publish_data() ) -> drift:step(state(), input(), output(), binary()). handle_publish(Context, State, Data) -> {Msg@4, Session@1, Packet, Storage_updates@1} = case Data of {publish_data_qo_s0, Msg} -> {{some, Msg}, erlang:element(3, State), none, []}; {publish_data_qo_s1, Msg@1, _, Id} -> {{some, Msg@1}, erlang:element(3, State), {some, {pub_ack, Id}}, []}; {publish_data_qo_s2, Msg@2, _, Id@1} -> {Session, Publish_result, Storage_updates} = spoke@core@internal@session:start_qos2_receive( erlang:element(3, State), Id@1 ), Msg@3 = case Publish_result of true -> {some, Msg@2}; false -> none end, {Msg@3, Session, {some, {pub_rec, Id@1}}, Storage_updates} end, _pipe@2 = case {Packet, erlang:element(4, State)} of {{some, Packet@1}, {connected, _}} -> _pipe = Context, _pipe@1 = drift:output(_pipe, send(Packet@1)), start_send_ping_timer(_pipe@1, State); {_, _} -> drift:continue(Context, State) end, drift:chain( _pipe@2, fun(Context@1, State@1) -> Context@2 = case Msg@4 of {some, Msg@5} -> publish_update( Context@1, State@1, {received_message, erlang:element(2, Msg@5), erlang:element(3, Msg@5), erlang:element(4, Msg@5)} ); none -> Context@1 end, _pipe@3 = Context@2, _pipe@4 = output_storage_updates(_pipe@3, Storage_updates@1), drift:continue( _pipe@4, {state, erlang:element(2, State@1), Session@1, erlang:element(4, State@1), erlang:element(5, State@1), erlang:element(6, State@1), erlang:element(7, State@1), erlang:element(8, State@1), erlang:element(9, State@1), erlang:element(10, State@1), erlang:element(11, State@1)} ) end ). -file("src/spoke/core.gleam", 841). -spec handle_puback(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). handle_puback(Context, State, Id) -> _pipe@2 = case spoke@core@internal@session:handle_puback( erlang:element(3, State), Id ) of {publish_finished, Session, Storage_updates} -> _pipe = Context, _pipe@1 = output_storage_updates(_pipe, Storage_updates), drift:continue( _pipe@1, {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ); invalid_pub_ack_id -> drift:continue(Context, State) end, drift:chain(_pipe@2, fun check_publish_completion/2). -file("src/spoke/core.gleam", 855). -spec handle_pubrec(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). handle_pubrec(Context, State, Id) -> {Session, Storage_updates} = spoke@core@internal@session:handle_pubrec( erlang:element(3, State), Id ), _pipe = Context, _pipe@1 = drift:output(_pipe, send({pub_rel, Id})), _pipe@2 = output_storage_updates(_pipe@1, Storage_updates), _pipe@3 = drift:continue( _pipe@2, {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@3, fun start_send_ping_timer/2). -file("src/spoke/core.gleam", 864). -spec handle_pubcomp(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). handle_pubcomp(Context, State, Id) -> {Session, Storage_updates} = spoke@core@internal@session:handle_pubcomp( erlang:element(3, State), Id ), _pipe = Context, _pipe@1 = output_storage_updates(_pipe, Storage_updates), _pipe@2 = drift:continue( _pipe@1, {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@2, fun check_publish_completion/2). -file("src/spoke/core.gleam", 872). -spec handle_pubrel(drift:context(input(), output()), state(), integer()) -> drift:step(state(), input(), output(), binary()). handle_pubrel(Context, State, Id) -> {Session, Storage_updates} = spoke@core@internal@session:handle_pubrel( erlang:element(3, State), Id ), _pipe = Context, _pipe@1 = drift:output(_pipe, send({pub_comp, Id})), _pipe@2 = output_storage_updates(_pipe@1, Storage_updates), _pipe@3 = drift:continue( _pipe@2, {state, erlang:element(2, State), Session, erlang:element(4, State), erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), drift:chain(_pipe@3, fun start_send_ping_timer/2). -file("src/spoke/core.gleam", 665). -spec handle_packets_while_connected( drift:step(state(), input(), output(), binary()), list(spoke@packet@client@incoming:packet()) ) -> drift:step(state(), input(), output(), binary()). handle_packets_while_connected(Step, Packets) -> gleam@list:fold( Packets, Step, fun(Step@1, Packet) -> drift:chain(Step@1, fun(Context, State) -> case Packet of {conn_ack, _} -> kill_connection( Context, State, <<"Got CONNACK while already connected"/utf8>> ); ping_resp -> handle_ping_resp(Context, State); {pub_ack, Id} -> handle_puback(Context, State, Id); {pub_rec, Id@1} -> handle_pubrec(Context, State, Id@1); {pub_comp, Id@2} -> handle_pubcomp(Context, State, Id@2); {pub_rel, Id@3} -> handle_pubrel(Context, State, Id@3); {publish, Data} -> handle_publish(Context, State, Data); {sub_ack, Id@4, Return_codes} -> handle_suback(Context, State, Id@4, Return_codes); {unsub_ack, Id@5} -> handle_unsuback(Context, State, Id@5) end end) end ). -file("src/spoke/core.gleam", 614). -spec 'receive'(drift:context(input(), output()), state(), bitstring()) -> drift:step(state(), input(), output(), binary()). 'receive'(Context, State, Data) -> case erlang:element(4, State) of {waiting_for_conn_ack, Connection} -> case spoke@core@internal@connection:receive_one(Connection, Data) of {error, E} -> kill_connection( Context, State, <<"Received invalid data while connecting: "/utf8, (gleam@string:inspect(E))/binary>> ); {ok, {Connection@1, {some, Packet}}} -> Step = handle_first_packet( Context, State, Connection@1, Packet ), drift:chain( Step, fun(Context@1, State@1) -> 'receive'(Context@1, State@1, <<>>) end ); {ok, {Connection@2, none}} -> drift:continue( Context, {state, erlang:element(2, State), erlang:element(3, State), {waiting_for_conn_ack, Connection@2}, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ) end; {connected, Connection@3} -> case spoke@core@internal@connection:receive_all(Connection@3, Data) of {error, E@1} -> kill_connection( Context, State, <<"Received invalid data while connected: "/utf8, (gleam@string:inspect(E@1))/binary>> ); {ok, {Connection@4, Packets}} -> _pipe = Context, _pipe@1 = drift:continue( _pipe, {state, erlang:element(2, State), erlang:element(3, State), {connected, Connection@4}, erlang:element(5, State), erlang:element(6, State), erlang:element(7, State), erlang:element(8, State), erlang:element(9, State), erlang:element(10, State), erlang:element(11, State)} ), handle_packets_while_connected(_pipe@1, Packets) end; {connecting, _} -> kill_connection( Context, State, <<"Received data before sending CONNECT"/utf8>> ); not_connected -> drift:continue(Context, State); disconnecting -> drift:continue(Context, State) end. -file("src/spoke/core.gleam", 143). ?DOC(" Runs one step, handing the given input.\n"). -spec handle_input(drift:context(input(), output()), state(), input()) -> drift:step(state(), input(), output(), binary()). handle_input(Context, State, Input) -> case Input of {handle, Event} -> case Event of {received_data, Data} -> 'receive'(Context, State, Data); transport_established -> transport_established(Context, State); {transport_failed, Error} -> disconnect_unexpectedly(Context, State, Error); transport_closed -> transport_closed(Context, State) end; {perform, Action} -> case Action of {subscribe_to_updates, Publish} -> subscribe_to_updates(Context, State, Publish); {unsubscribe_from_updates, Publish@1} -> unsubscribe_from_updates(Context, State, Publish@1); {connect, Options, Will} -> connect(Context, State, Options, Will); {disconnect, Complete} -> disconnect(Context, State, Complete); {subscribe, Requests, Effect} -> subscribe(Context, State, Requests, Effect); {unsubscribe, Topics, Effect@1} -> unsubscribe(Context, State, Topics, Effect@1); {publish_message, Data@1} -> publish(Context, State, Data@1); {get_pending_publishes, Complete@1} -> get_pending_publishes(Context, State, Complete@1); {wait_for_publishes_to_finish, Complete@2, Timeout} -> wait_for_publishes(Context, State, Complete@2, Timeout) end; {timeout, Action@1} -> handle_timer(Context, State, Action@1) end.