-module(stratus). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]). -define(FILEPATH, "src/stratus.gleam"). -export([continue/1, with_selector/2, stop/0, stop_abnormal/1, websocket/3, with_connect_timeout/2, on_close/2, on_handshake_error/2, to_user_message/1, send_text_message/2, send_binary_message/2, send_ping/2, close_with_reason/2, close/1, initialize/1]). -export_type([connection/0, next/2, internal_message/1, message/1, builder/2, state/2, close_reason/0, handshake_error/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. -opaque connection() :: {connection, stratus@internal@socket:socket(), stratus@internal@transport:transport(), gleam@option:option(gramps@websocket@compression:context())}. -opaque next(HFH, HFI) :: {continue, HFH, gleam@option:option(gleam@erlang@process:selector(HFI))} | normal_stop | {abnormal_stop, binary()}. -opaque internal_message(HFJ) :: started | {user_message, HFJ} | {err, stratus@internal@socket:socket_reason()} | {data, bitstring()} | closed | shutdown. -type message(HFK) :: {text, binary()} | {binary, bitstring()} | {user, HFK}. -opaque builder(HFL, HFM) :: {builder, gleam@http@request:request(binary()), integer(), fun(() -> {HFL, gleam@option:option(gleam@erlang@process:selector(HFM))}), fun((HFL, message(HFM), connection()) -> next(HFL, HFM)), fun((HFL) -> nil), fun((gleam@http@response:response(bitstring())) -> nil)}. -type state(HFN, HFO) :: {state, bitstring(), gleam@option:option(gramps@websocket:frame()), gleam@erlang@process:subject(internal_message(HFO)), gleam@option:option(stratus@internal@socket:socket()), HFN, gleam@option:option(gramps@websocket@compression:compression())}. -type close_reason() :: {normal, bitstring()} | {going_away, bitstring()} | {protocol_error, bitstring()} | {unexpected_data_type, bitstring()} | {inconsistent_data_type, bitstring()} | {policy_violation, bitstring()} | {message_too_big, bitstring()} | {missing_extensions, bitstring()} | {unexpected_condition, bitstring()}. -type handshake_error() :: {sock, stratus@internal@socket:socket_reason()} | {protocol, bitstring()} | {upgrade_failed, gleam@http@response:response(bitstring())}. -file("src/stratus.gleam", 43). -spec from_socket_message(stratus@internal@socket:socket_message()) -> internal_message(any()). from_socket_message(Msg) -> case Msg of {data, Bits} -> {data, Bits}; {err, closed} -> closed; {err, Reason} -> {err, Reason} end. -file("src/stratus.gleam", 57). -spec continue(HFU) -> next(HFU, any()). continue(State) -> {continue, State, none}. -file("src/stratus.gleam", 61). -spec with_selector(next(HFY, HFZ), gleam@erlang@process:selector(HFZ)) -> next(HFY, HFZ). with_selector(Next, Selector) -> case Next of {continue, State, _} -> {continue, State, {some, Selector}}; _ -> Next end. -file("src/stratus.gleam", 71). -spec stop() -> next(any(), any()). stop() -> normal_stop. -file("src/stratus.gleam", 75). -spec stop_abnormal(binary()) -> next(any(), any()). stop_abnormal(Reason) -> {abnormal_stop, Reason}. -file("src/stratus.gleam", 114). ?DOC( " This creates a builder to set up a WebSocket actor. This will use default\n" " values for the connection initialization timeout, and provide an empty\n" " function to be called when the server closes the connection. If you want to\n" " customize either of those, see the helper functions `with_connect_timeout`\n" ). -spec websocket( gleam@http@request:request(binary()), fun(() -> {HGO, gleam@option:option(gleam@erlang@process:selector(HGP))}), fun((HGO, message(HGP), connection()) -> next(HGO, HGP)) ) -> builder(HGO, HGP). websocket(Req, Init, Loop) -> {builder, Req, 5000, Init, Loop, fun(_) -> nil end, fun(_) -> nil end}. -file("src/stratus.gleam", 134). ?DOC( " This sets the maximum amount of time you are willing to wait for both\n" " connecting to the server and receiving the upgrade response. This means\n" " that it may take up to `timeout * 2` to begin sending or receiving messages.\n" " This value defaults to 5 seconds.\n" ). -spec with_connect_timeout(builder(HGX, HGY), integer()) -> builder(HGX, HGY). with_connect_timeout(Builder, Timeout) -> _record = Builder, {builder, erlang:element(2, _record), Timeout, erlang:element(4, _record), erlang:element(5, _record), erlang:element(6, _record), erlang:element(7, _record)}. -file("src/stratus.gleam", 147). ?DOC( " You can provide a function to be called when the connection is closed. This\n" " function receives the last value for the state of the WebSocket.\n" "\n" " NOTE: If you manually call `stratus.close`, this function will not be\n" " called. I'm unsure right now if this is a bug or working as intended. But\n" " you will be in the loop with the state value handy.\n" ). -spec on_close(builder(HHD, HHE), fun((HHD) -> nil)) -> builder(HHD, HHE). on_close(Builder, On_close) -> _record = Builder, {builder, erlang:element(2, _record), erlang:element(3, _record), erlang:element(4, _record), erlang:element(5, _record), On_close, erlang:element(7, _record)}. -file("src/stratus.gleam", 156). ?DOC( " If the WebSocket handshake fails, this method will be called with the\n" " response received from the server. The process will stop after this.\n" ). -spec on_handshake_error( builder(HHJ, HHK), fun((gleam@http@response:response(bitstring())) -> nil) ) -> builder(HHJ, HHK). on_handshake_error(Builder, On_handshake_error) -> _record = Builder, {builder, erlang:element(2, _record), erlang:element(3, _record), erlang:element(4, _record), erlang:element(5, _record), erlang:element(6, _record), On_handshake_error}. -file("src/stratus.gleam", 420). -spec handle_frame( builder(HHZ, HIA), state(HHZ, HIA), connection(), gramps@websocket:frame() ) -> next(state(HHZ, HIA), internal_message(HIA)). handle_frame(Builder, State, Conn, Frame) -> case Frame of {data, {text_frame, Data}} -> Str@1 = case gleam@bit_array:to_string(Data) of {ok, Str} -> Str; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"handle_frame"/utf8>>, line => 428, value => _assert_fail, start => 13502, 'end' => 13548, pattern_start => 13513, pattern_end => 13520}) end, Res = stratus_ffi:rescue( fun() -> (erlang:element(5, Builder))( erlang:element(6, State), {text, Str@1}, Conn ) end ), case Res of {ok, {continue, User_state, User_selector}} -> New_state = begin _record = State, {state, erlang:element(2, _record), erlang:element(3, _record), erlang:element(4, _record), erlang:element(5, _record), User_state, erlang:element(7, _record)} end, case User_selector of {some, User_selector@1} -> Selector = begin _pipe = User_selector@1, _pipe@1 = gleam_erlang_ffi:map_selector( _pipe, fun(Field@0) -> {user_message, Field@0} end ), gleam_erlang_ffi:merge_selector( _pipe@1, gleam_erlang_ffi:map_selector( stratus@internal@socket:selector(), fun(Msg) -> Msg@2 = case Msg of {ok, Msg@1} -> Msg@1; _assert_fail@1 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"handle_frame"/utf8>>, line => 441, value => _assert_fail@1, start => 14107, 'end' => 14131, pattern_start => 14118, pattern_end => 14125} ) end, from_socket_message(Msg@2) end ) ) end, {continue, New_state, {some, Selector}}; _ -> continue(New_state) end; {ok, normal_stop} -> normal_stop; {ok, {abnormal_stop, Reason}} -> {abnormal_stop, Reason}; {error, Reason@1} -> logging:log( error, <<"Caught error in user handler: "/utf8, (gleam@string:inspect(Reason@1))/binary>> ), continue(State) end; {data, {binary_frame, Data@1}} -> Res@1 = stratus_ffi:rescue( fun() -> (erlang:element(5, Builder))( erlang:element(6, State), {binary, Data@1}, Conn ) end ), case Res@1 of {ok, {continue, User_state@1, User_selector@2}} -> New_state@1 = begin _record@1 = State, {state, erlang:element(2, _record@1), erlang:element(3, _record@1), erlang:element(4, _record@1), erlang:element(5, _record@1), User_state@1, erlang:element(7, _record@1)} end, case User_selector@2 of {some, User_selector@3} -> Selector@1 = begin _pipe@2 = User_selector@3, _pipe@3 = gleam_erlang_ffi:map_selector( _pipe@2, fun(Field@0) -> {user_message, Field@0} end ), gleam_erlang_ffi:merge_selector( _pipe@3, gleam_erlang_ffi:map_selector( stratus@internal@socket:selector(), fun(Msg@3) -> Msg@5 = case Msg@3 of {ok, Msg@4} -> Msg@4; _assert_fail@2 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"handle_frame"/utf8>>, line => 475, value => _assert_fail@2, start => 15261, 'end' => 15285, pattern_start => 15272, pattern_end => 15279} ) end, from_socket_message(Msg@5) end ) ) end, {continue, New_state@1, {some, Selector@1}}; _ -> continue(New_state@1) end; {ok, normal_stop} -> normal_stop; {ok, {abnormal_stop, Reason@2}} -> {abnormal_stop, Reason@2}; {error, Reason@3} -> logging:log( error, <<"Caught error in user handler: "/utf8, (gleam@string:inspect(Reason@3))/binary>> ), continue(State) end; {control, {ping_frame, Payload}} -> Frame@1 = case erlang:element(4, Conn) of {some, Context} -> gramps@websocket:compressed_frame_to_bytes_tree( {control, {pong_frame, Payload}}, Context, {some, <<0:4/unit:8>>} ); none -> gramps@websocket:frame_to_bytes_tree( {control, {pong_frame, Payload}}, {some, <<0:4/unit:8>>} ) end, _ = stratus@internal@transport:send( erlang:element(3, Conn), erlang:element(2, Conn), Frame@1 ), continue(State); {control, {pong_frame, _}} -> continue(State); {control, {close_frame, Reason@4}} -> logging:log( debug, <<"WebSocket closing: "/utf8, (gleam@string:inspect(Reason@4))/binary>> ), (erlang:element(6, Builder))(erlang:element(6, State)), normal_stop; {continuation, _, _} -> continue(State) end. -file("src/stratus.gleam", 543). ?DOC( " The `Subject` returned from `initialize` is an opaque type. In order to\n" " send custom messages to your process, you can do this mapping.\n" "\n" " For example:\n" " ```gleam\n" " // using `process.send`\n" " MyMessage(some_data)\n" " |> stratus.to_user_message\n" " |> process.send(stratus_subject, _)\n" " // using `process.call`\n" " process.call(stratus_subject, fn(subj) {\n" " stratus.to_user_message(MyMessage(some_data, subj))\n" " })\n" " ```\n" ). -spec to_user_message(HIK) -> internal_message(HIK). to_user_message(User_message) -> {user_message, User_message}. -file("src/stratus.gleam", 551). ?DOC( " From within the actor loop, this is how you send a WebSocket text frame.\n" " This must be valid UTF-8, so it is a `String`.\n" ). -spec send_text_message(connection(), binary()) -> {ok, nil} | {error, stratus@internal@socket:socket_reason()}. send_text_message(Conn, Msg) -> Frame = gramps@websocket:to_text_frame( Msg, none, {some, crypto:strong_rand_bytes(4)} ), stratus@internal@transport:send( erlang:element(3, Conn), erlang:element(2, Conn), Frame ). -file("src/stratus.gleam", 561). ?DOC(" From within the actor loop, this is how you send a WebSocket text frame.\n"). -spec send_binary_message(connection(), bitstring()) -> {ok, nil} | {error, stratus@internal@socket:socket_reason()}. send_binary_message(Conn, Msg) -> Frame = gramps@websocket:to_binary_frame( Msg, none, {some, crypto:strong_rand_bytes(4)} ), stratus@internal@transport:send( erlang:element(3, Conn), erlang:element(2, Conn), Frame ). -file("src/stratus.gleam", 571). ?DOC(" Send a ping frame with some data.\n"). -spec send_ping(connection(), bitstring()) -> {ok, nil} | {error, stratus@internal@socket:socket_reason()}. send_ping(Conn, Data) -> Size = erlang:byte_size(Data), Mask = case Size of 0 -> <<0:4>>; _ -> crypto:strong_rand_bytes(4) end, Frame = case erlang:element(4, Conn) of {some, Context} -> gramps@websocket:compressed_frame_to_bytes_tree( {control, {ping_frame, Data}}, Context, {some, Mask} ); none -> gramps@websocket:frame_to_bytes_tree( {control, {ping_frame, Data}}, {some, Mask} ) end, stratus@internal@transport:send( erlang:element(3, Conn), erlang:element(2, Conn), Frame ). -file("src/stratus.gleam", 610). -spec convert_close_reason(close_reason()) -> gramps@websocket:close_reason(). convert_close_reason(Reason) -> case Reason of {going_away, Body} -> {going_away, Body}; {inconsistent_data_type, Body@1} -> {inconsistent_data_type, Body@1}; {message_too_big, Body@2} -> {message_too_big, Body@2}; {missing_extensions, Body@3} -> {missing_extensions, Body@3}; {normal, Body@4} -> {normal, Body@4}; {policy_violation, Body@5} -> {policy_violation, Body@5}; {protocol_error, Body@6} -> {protocol_error, Body@6}; {unexpected_condition, Body@7} -> {unexpected_condition, Body@7}; {unexpected_data_type, Body@8} -> {unexpected_data_type, Body@8} end. -file("src/stratus.gleam", 625). ?DOC(" This closes the WebSocket connection with a particular close reason.\n"). -spec close_with_reason(connection(), close_reason()) -> {ok, nil} | {error, stratus@internal@socket:socket_reason()}. close_with_reason(Conn, Reason) -> Reason@1 = convert_close_reason(Reason), Mask = crypto:strong_rand_bytes(4), Frame = gramps@websocket:frame_to_bytes_tree( {control, {close_frame, Reason@1}}, {some, Mask} ), stratus@internal@transport:send( erlang:element(3, Conn), erlang:element(2, Conn), Frame ). -file("src/stratus.gleam", 594). ?DOC(" This will close the WebSocket connection.\n"). -spec close(connection()) -> {ok, nil} | {error, stratus@internal@socket:socket_reason()}. close(Conn) -> close_with_reason(Conn, {normal, <<>>}). -file("src/stratus.gleam", 639). -spec make_upgrade(gleam@http@request:request(binary())) -> gleam@bytes_tree:bytes_tree(). make_upgrade(Req) -> User_headers = case erlang:element(3, Req) of [] -> <<""/utf8>>; _ -> _pipe = erlang:element(3, Req), _pipe@1 = gleam@list:filter( _pipe, fun(Pair) -> {Key, _} = Pair, ((((Key /= <<"host"/utf8>>) andalso (Key /= <<"upgrade"/utf8>>)) andalso (Key /= <<"connection"/utf8>>)) andalso (Key /= <<"sec-websocket-key"/utf8>>)) andalso (Key /= <<"sec-websocket-version"/utf8>>) end ), _pipe@2 = gleam@list:map( _pipe@1, fun(Pair@1) -> {Key@1, Value} = Pair@1, <<<>/binary, Value/binary>> end ), _pipe@3 = gleam@string:join(_pipe@2, <<"\r\n"/utf8>>), gleam@string:append(_pipe@3, <<"\r\n"/utf8>>) end, Path@1 = case erlang:element(8, Req) of <<""/utf8>> -> <<"/"/utf8>>; Path -> Path end, Query = begin _pipe@4 = Req, _pipe@5 = gleam@http@request:get_query(_pipe@4), _pipe@6 = gleam@result:map(_pipe@5, fun gleam@uri:query_to_string/1), (fun(Str) -> case Str of {ok, <<""/utf8>>} -> <<""/utf8>>; {ok, Str@1} -> <<"?"/utf8, Str@1/binary>>; _ -> <<""/utf8>> end end)(_pipe@6) end, Port@1 = begin _pipe@7 = erlang:element(7, Req), _pipe@8 = gleam@option:map( _pipe@7, fun(Port) -> <<":"/utf8, (erlang:integer_to_binary(Port))/binary>> end ), gleam@option:unwrap(_pipe@8, <<""/utf8>>) end, _pipe@9 = gleam@bytes_tree:new(), _pipe@10 = gleam@bytes_tree:append_string( _pipe@9, <<<<<<"GET "/utf8, Path@1/binary>>/binary, Query/binary>>/binary, " HTTP/1.1\r\n"/utf8>> ), _pipe@11 = gleam@bytes_tree:append_string( _pipe@10, <<<<<<"host: "/utf8, (erlang:element(6, Req))/binary>>/binary, Port@1/binary>>/binary, "\r\n"/utf8>> ), _pipe@12 = gleam@bytes_tree:append_string( _pipe@11, <<"upgrade: websocket\r\n"/utf8>> ), _pipe@13 = gleam@bytes_tree:append_string( _pipe@12, <<"connection: upgrade\r\n"/utf8>> ), _pipe@14 = gleam@bytes_tree:append_string( _pipe@13, <<<<"sec-websocket-key: "/utf8, (gramps@websocket:make_client_key())/binary>>/binary, "\r\n"/utf8>> ), _pipe@15 = gleam@bytes_tree:append_string( _pipe@14, <<"sec-websocket-version: 13\r\n"/utf8>> ), _pipe@16 = gleam@bytes_tree:append_string( _pipe@15, <<"sec-websocket-extensions: permessage-deflate\r\n"/utf8>> ), _pipe@17 = gleam@bytes_tree:append_string(_pipe@16, User_headers), gleam@bytes_tree:append_string(_pipe@17, <<"\r\n"/utf8>>). -file("src/stratus.gleam", 789). -spec read_body( stratus@internal@transport:transport(), stratus@internal@socket:socket(), integer(), integer(), bitstring() ) -> {ok, {bitstring(), bitstring()}} | {error, stratus@internal@socket:socket_reason()}. read_body(Transport, Socket, Timeout, Length, Body) -> case Body of <> -> {ok, {Data, Rest}}; _ -> case stratus@internal@transport:receive_timeout( Transport, Socket, 0, Timeout ) of {ok, Data@1} -> read_body( Transport, Socket, Timeout, Length, <> ); {error, Reason} -> {error, Reason} end end. -file("src/stratus.gleam", 704). -spec perform_handshake( gleam@http@request:request(binary()), stratus@internal@transport:transport(), integer() ) -> {ok, {stratus@internal@socket:socket(), gleam@http@response:response(bitstring()), bitstring()}} | {error, handshake_error()}. perform_handshake(Req, Transport, Timeout) -> Certs = case erlang:element(5, Req) of https -> case stratus_ffi:ssl_start() of {ok, _} -> nil; _assert_fail -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"perform_handshake"/utf8>>, line => 711, value => _assert_fail, start => 22033, 'end' => 22065, pattern_start => 22044, pattern_end => 22051}) end, [{cacerts, public_key:cacerts_get()}, stratus_ffi:custom_sni_matcher()]; http -> [] end, Opts = stratus@internal@socket:convert_options( lists:append( [{packets_of, binary}, {send_timeout, 30000}, {send_timeout_close, true}, {reuseaddr, true}, {nodelay, true}], [{'receive', pull} | Certs] ) ), Port = gleam@option:lazy_unwrap( erlang:element(7, Req), fun() -> case Transport of ssl -> 443; tcp -> 80 end end ), logging:log( debug, <<<<<<"Making request to "/utf8, (erlang:element(6, Req))/binary>>/binary, " at "/utf8>>/binary, (erlang:integer_to_binary(Port))/binary>> ), gleam@result:'try'( gleam@result:map_error( stratus@internal@transport:connect( Transport, unicode:characters_to_list(erlang:element(6, Req)), Port, Opts, Timeout ), fun(Field@0) -> {sock, Field@0} end ), fun(Socket) -> Upgrade_req = make_upgrade(Req), gleam@result:'try'( gleam@result:map_error( stratus@internal@transport:send( Transport, Socket, Upgrade_req ), fun(Field@0) -> {sock, Field@0} end ), fun(_) -> logging:log( debug, <<"Sent upgrade request, waiting "/utf8, (erlang:integer_to_binary(Timeout))/binary>> ), gleam@result:'try'( gleam@result:map_error( stratus@internal@transport:receive_timeout( Transport, Socket, 0, Timeout ), fun(Field@0) -> {sock, Field@0} end ), fun(Resp) -> _pipe = Resp, _pipe@1 = gramps@http:read_response(_pipe), _pipe@2 = gleam@result:map_error( _pipe@1, fun(_) -> {protocol, Resp} end ), _pipe@6 = gleam@result:'try'( _pipe@2, fun(Pair) -> {Resp@1, Body} = Pair, Body_size = begin _pipe@3 = erlang:element(3, Resp@1), _pipe@4 = gleam@list:key_find( _pipe@3, <<"content-length"/utf8>> ), _pipe@5 = gleam@result:'try'( _pipe@4, fun gleam_stdlib:parse_int/1 ), gleam@result:unwrap(_pipe@5, 0) end, case read_body( Transport, Socket, Timeout, Body_size, Body ) of {ok, {Body@1, Rest}} -> {ok, {gleam@http@response:set_body( Resp@1, Body@1 ), Rest}}; {error, Reason} -> {error, {sock, Reason}} end end ), gleam@result:'try'( _pipe@6, fun(Pair@1) -> {Resp@2, Rest@1} = Pair@1, case erlang:element(2, Resp@2) of 101 -> {ok, {Socket, Resp@2, Rest@1}}; _ -> {error, {upgrade_failed, Resp@2}} end end ) end ) end ) end ). -file("src/stratus.gleam", 809). -spec close_contexts( gleam@option:option(gramps@websocket@compression:compression()) ) -> nil. close_contexts(Contexts) -> case Contexts of {some, Compression} -> zlib:close(erlang:element(3, Compression)), zlib:close(erlang:element(2, Compression)), nil; _ -> nil end. -file("src/stratus.gleam", 184). ?DOC( " This opens the WebSocket connection with the provided `Builder`. It makes\n" " some assumptions about the request if you do not provide it. It will use\n" " ports 80 or 443 for `ws` or `wss` respectively.\n" "\n" " It will open the connection and perform the WebSocket handshake. If this\n" " fails, the actor will fail to start with the given reason as a string value.\n" "\n" " After that, received messages will be passed to your loop, and you can use\n" " the helper functions to send messages to the server. The `close` method will\n" " send a close frame and end the connection.\n" ). -spec initialize(builder(any(), HHR)) -> {ok, gleam@otp@actor:started(gleam@erlang@process:subject(internal_message(HHR)))} | {error, gleam@otp@actor:start_error()}. initialize(Builder) -> Transport = case erlang:element(5, erlang:element(2, Builder)) of https -> ssl; _ -> tcp end, _pipe@8 = gleam@otp@actor:new_with_initialiser( 1000, fun(Subject) -> Started_selector = gleam@erlang@process:select( gleam_erlang_ffi:new_selector(), Subject ), logging:log(debug, <<"Calling user initializer"/utf8>>), {User_state, User_selector} = (erlang:element(4, Builder))(), Selector@1 = case User_selector of {some, Selector} -> _pipe = Selector, _pipe@1 = gleam_erlang_ffi:map_selector( _pipe, fun(Field@0) -> {user_message, Field@0} end ), _pipe@2 = gleam_erlang_ffi:merge_selector( _pipe@1, Started_selector ), gleam_erlang_ffi:merge_selector( _pipe@2, gleam_erlang_ffi:map_selector( stratus@internal@socket:selector(), fun(Msg) -> Msg@2 = case Msg of {ok, Msg@1} -> Msg@1; _assert_fail -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 206, value => _assert_fail, start => 6540, 'end' => 6564, pattern_start => 6551, pattern_end => 6558} ) end, from_socket_message(Msg@2) end ) ); _ -> _pipe@3 = Started_selector, gleam_erlang_ffi:merge_selector( _pipe@3, gleam_erlang_ffi:map_selector( stratus@internal@socket:selector(), fun(Msg@3) -> Msg@5 = case Msg@3 of {ok, Msg@4} -> Msg@4; _assert_fail@1 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 215, value => _assert_fail@1, start => 6777, 'end' => 6801, pattern_start => 6788, pattern_end => 6795} ) end, from_socket_message(Msg@5) end ) ) end, gleam@erlang@process:send(Subject, started), _pipe@4 = {state, <<>>, none, Subject, none, User_state, none}, _pipe@5 = gleam@otp@actor:initialised(_pipe@4), _pipe@6 = gleam@otp@actor:selecting(_pipe@5, Selector@1), _pipe@7 = gleam@otp@actor:returning(_pipe@6, Subject), {ok, _pipe@7} end ), _pipe@23 = gleam@otp@actor:on_message( _pipe@8, fun(State, Message) -> case Message of started -> logging:log( debug, <<"Attempting handshake to "/utf8, (gleam@uri:to_string( gleam@http@request:to_uri( erlang:element(2, Builder) ) ))/binary>> ), _pipe@9 = perform_handshake( erlang:element(2, Builder), Transport, erlang:element(3, Builder) ), _pipe@12 = gleam@result:'try'( _pipe@9, fun(Pair) -> logging:log(debug, <<"Handshake successful"/utf8>>), _pipe@10 = stratus@internal@transport:set_opts( Transport, erlang:element(1, Pair), stratus@internal@socket:convert_options( [{'receive', once}] ) ), _pipe@11 = gleam@result:replace(_pipe@10, Pair), gleam@result:map_error( _pipe@11, fun(Field@0) -> {sock, Field@0} end ) end ), _pipe@16 = gleam@result:map( _pipe@12, fun(Pair@1) -> {Socket, Resp, Buffer} = Pair@1, logging:log( debug, <<"WebSocket process ready to start receiving"/utf8>> ), _ = case Buffer of <<>> -> nil; Data -> gleam@erlang@process:send( erlang:element(4, State), {data, Data} ) end, Extensions = begin _pipe@13 = Resp, _pipe@14 = gleam@http@response:get_header( _pipe@13, <<"sec-websocket-extensions"/utf8>> ), _pipe@15 = gleam@result:map( _pipe@14, fun(_capture) -> gleam@string:split( _capture, <<";"/utf8>> ) end ), gleam@result:unwrap(_pipe@15, []) end, Context = case gramps@websocket:has_deflate( Extensions ) of true -> {some, gramps@websocket@compression:init()}; false -> none end, gleam@otp@actor:continue( begin _record = State, {state, Buffer, erlang:element(3, _record), erlang:element(4, _record), {some, Socket}, erlang:element(6, _record), Context} end ) end ), _pipe@17 = gleam@result:map_error( _pipe@16, fun(Err) -> case Err of {protocol, _} -> Msg@6 = <<"Failed to connect to server: "/utf8, (gleam@string:inspect(Err))/binary>>, logging:log(error, Msg@6), gleam@otp@actor:stop_abnormal(Msg@6); {sock, _} -> Msg@6 = <<"Failed to connect to server: "/utf8, (gleam@string:inspect(Err))/binary>>, logging:log(error, Msg@6), gleam@otp@actor:stop_abnormal(Msg@6); {upgrade_failed, Resp@1} -> (erlang:element(7, Builder))(Resp@1), logging:log( error, <<"WebSocket handshake failed with status "/utf8, (erlang:integer_to_binary( erlang:element(2, Resp@1) ))/binary>> ), gleam@otp@actor:stop_abnormal( <<"WebSocket handshake failed"/utf8>> ) end end ), gleam@result:unwrap_both(_pipe@17); {user_message, User_message} -> Socket@2 = case erlang:element(5, State) of {some, Socket@1} -> Socket@1; _assert_fail@2 -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 302, value => _assert_fail@2, start => 9397, 'end' => 9435, pattern_start => 9408, pattern_end => 9420}) end, Conn = {connection, Socket@2, Transport, gleam@option:map( erlang:element(7, State), fun(Context@1) -> erlang:element(3, Context@1) end )}, Res = stratus_ffi:rescue( fun() -> (erlang:element(5, Builder))( erlang:element(6, State), {user, User_message}, Conn ) end ), case Res of {ok, {continue, User_state@1, User_selector@1}} -> New_state = begin _record@1 = State, {state, erlang:element(2, _record@1), erlang:element(3, _record@1), erlang:element(4, _record@1), erlang:element(5, _record@1), User_state@1, erlang:element(7, _record@1)} end, case User_selector@1 of {some, User_selector@2} -> Selector@2 = begin _pipe@18 = User_selector@2, _pipe@19 = gleam_erlang_ffi:map_selector( _pipe@18, fun(Field@0) -> {user_message, Field@0} end ), gleam_erlang_ffi:merge_selector( _pipe@19, gleam_erlang_ffi:map_selector( stratus@internal@socket:selector( ), fun(Msg@7) -> Msg@9 = case Msg@7 of {ok, Msg@8} -> Msg@8; _assert_fail@3 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 324, value => _assert_fail@3, start => 10233, 'end' => 10257, pattern_start => 10244, pattern_end => 10251} ) end, from_socket_message(Msg@9) end ) ) end, _pipe@20 = New_state, _pipe@21 = gleam@otp@actor:continue( _pipe@20 ), gleam@otp@actor:with_selector( _pipe@21, Selector@2 ); _ -> gleam@otp@actor:continue(New_state) end; {ok, normal_stop} -> gleam@otp@actor:stop(); {ok, {abnormal_stop, Reason}} -> gleam@otp@actor:stop_abnormal(Reason); {error, Reason@1} -> logging:log( error, <<"Caught error in user handler: "/utf8, (gleam@string:inspect(Reason@1))/binary>> ), gleam@otp@actor:continue(State) end; {err, Reason@2} -> close_contexts(erlang:element(7, State)), gleam@otp@actor:stop_abnormal( gleam@string:inspect(Reason@2) ); {data, Bits} -> Socket@4 = case erlang:element(5, State) of {some, Socket@3} -> Socket@3; _assert_fail@4 -> erlang:error(#{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 351, value => _assert_fail@4, start => 11042, 'end' => 11080, pattern_start => 11053, pattern_end => 11065}) end, Conn@1 = {connection, Socket@4, Transport, gleam@option:map( erlang:element(7, State), fun(Context@2) -> erlang:element(3, Context@2) end )}, {Frames, Rest} = gramps@websocket:get_messages( gleam@bit_array:append(erlang:element(2, State), Bits), [], gleam@option:map( erlang:element(7, State), fun(Context@3) -> erlang:element(2, Context@3) end ) ), Frames@1 = gramps@websocket:aggregate_frames( Frames, erlang:element(3, State), [] ), _pipe@22 = case Frames@1 of {error, nil} -> continue(State); {ok, Frames@2} -> gleam@list:fold_until( Frames@2, continue(State), fun(Acc, Frame) -> Prev_state@1 = case Acc of {continue, Prev_state, _} -> Prev_state; _assert_fail@5 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 369, value => _assert_fail@5, start => 11721, 'end' => 11769, pattern_start => 11732, pattern_end => 11763} ) end, case handle_frame( Builder, Prev_state@1, Conn@1, Frame ) of {continue, _, _} = Next -> {continue, Next}; Err@1 -> {stop, Err@1} end end ) end, (fun(Next@1) -> case Next@1 of {continue, State@1, Selector@3} -> case stratus@internal@transport:set_opts( Transport, Socket@4, stratus@internal@socket:convert_options( [{'receive', once}] ) ) of {ok, _} -> nil; _assert_fail@6 -> erlang:error( #{gleam_error => let_assert, message => <<"Pattern match failed, no pattern matched the value."/utf8>>, file => <>, module => <<"stratus"/utf8>>, function => <<"initialize"/utf8>>, line => 380, value => _assert_fail@6, start => 12090, 'end' => 12276, pattern_start => 12101, pattern_end => 12106} ) end, Next@2 = gleam@otp@actor:continue( begin _record@2 = State@1, {state, Rest, erlang:element(3, _record@2), erlang:element(4, _record@2), erlang:element(5, _record@2), erlang:element(6, _record@2), erlang:element(7, _record@2)} end ), case Selector@3 of {some, Selector@4} -> gleam@otp@actor:with_selector( Next@2, Selector@4 ); _ -> Next@2 end; normal_stop -> close_contexts(erlang:element(7, State)), gleam@otp@actor:stop(); {abnormal_stop, Reason@3} -> close_contexts(erlang:element(7, State)), gleam@otp@actor:stop_abnormal(Reason@3) end end)(_pipe@22); closed -> logging:log(debug, <<"Received closed frame"/utf8>>), (erlang:element(6, Builder))(erlang:element(6, State)), close_contexts(erlang:element(7, State)), gleam@otp@actor:stop(); shutdown -> logging:log(debug, <<"Received shutdown messag"/utf8>>), close_contexts(erlang:element(7, State)), gleam@otp@actor:stop() end end ), gleam@otp@actor:start(_pipe@23).