Current section
Files
Jump to
Current section
Files
src/spoke@core.erl
-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.