Packages
mist
0.15.0
6.0.3
6.0.2
6.0.1
6.0.0
5.0.4
5.0.3
5.0.2
5.0.1
5.0.0
5.0.0-rc1
4.0.7
4.0.6
4.0.5
4.0.4
4.0.3
4.0.2
4.0.1
4.0.0
3.0.0
2.0.0
1.2.0
1.1.0
1.0.0
1.0.0-rc3
1.0.0-rc2
1.0.0-rc1
0.17.0
0.15.0
0.14.3
0.14.2
0.14.1
0.14.0
0.13.2
0.13.1
0.13.0
0.12.0
0.11.0
0.10.0
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.3
0.8.2
0.8.1
0.8.0
0.7.1
0.7.0
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.1
0.2.0
0.1.3
a misty Gleam web server
Current section
Files
Jump to
Current section
Files
src/mist@internal@websocket.erl
-module(mist@internal@websocket).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function]).
-export([frame_to_bytes_builder/1, to_text_frame/1, to_binary_frame/1, aggregate_frames/3, initialize_connection/5]).
-export_type([data_frame/0, control_frame/0, frame/0, frame_parse_error/0, parsed_frame/0, valid_message/1, websocket_message/1, websocket_connection/0, handler_message/1, websocket_state/1]).
-type data_frame() :: {text_frame, integer(), bitstring()} |
{binary_frame, integer(), bitstring()}.
-type control_frame() :: {close_frame, integer(), bitstring()} |
{ping_frame, integer(), bitstring()} |
{pong_frame, integer(), bitstring()}.
-type frame() :: {data, data_frame()} |
{control, control_frame()} |
{continuation, integer(), bitstring()}.
-type frame_parse_error() :: {need_more_data, bitstring()} | invalid_frame.
-type parsed_frame() :: {complete, frame()} | {incomplete, frame()}.
-type valid_message(JMO) :: {socket_message, bitstring()} |
socket_closed_message |
{user_message, JMO}.
-type websocket_message(JMP) :: {valid, valid_message(JMP)} | invalid.
-type websocket_connection() :: {websocket_connection,
glisten@socket:socket(),
glisten@socket@transport:transport()}.
-type handler_message(JMQ) :: {internal, frame()} | {user, JMQ}.
-type websocket_state(JMR) :: {websocket_state, bitstring(), JMR}.
-spec unmask_data(bitstring(), list(bitstring()), integer(), bitstring()) -> bitstring().
unmask_data(Data, Masks, Index, Resp) ->
case Data of
<<>> ->
Resp;
<<Masked:8/bitstring, Rest/bitstring>> ->
_assert_subject = gleam@list:at(Masks, Index rem 4),
{ok, Mask_value} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"mist/internal/websocket"/utf8>>,
function => <<"unmask_data"/utf8>>,
line => 48})
end,
Unmasked = crypto:exor(Mask_value, Masked),
unmask_data(
Rest,
Masks,
Index + 1,
<<Resp/bitstring, Unmasked/bitstring>>
)
end.
-spec frame_from_message(bitstring(), websocket_connection()) -> {ok,
{parsed_frame(), bitstring()}} |
{error, frame_parse_error()}.
frame_from_message(Message, Conn) ->
case Message of
<<Complete:1,
_:3,
Opcode:4/integer,
1:1,
Payload_length:7/integer,
Rest/bitstring>> ->
Payload_size = case Payload_length of
126 ->
16;
127 ->
64;
_ ->
0
end,
case Rest of
<<Length:Payload_size/integer,
Mask1:1/binary,
Mask2:1/binary,
Mask3:1/binary,
Mask4:1/binary,
Rest@1/bitstring>> ->
Payload_byte_size = case Length of
0 ->
Payload_length;
N ->
N
end,
case Rest@1 of
<<Payload:Payload_byte_size/binary, Rest@2/bitstring>> ->
Data = unmask_data(
Payload,
[Mask1, Mask2, Mask3, Mask4],
0,
<<>>
),
_pipe = case Opcode of
0 ->
{ok, {continuation, Payload_length, Data}};
1 ->
{ok,
{data,
{text_frame, Payload_length, Data}}};
2 ->
{ok,
{data,
{binary_frame, Payload_length, Data}}};
8 ->
{ok,
{control,
{close_frame, Payload_length, Data}}};
9 ->
{ok,
{control,
{ping_frame, Payload_length, Data}}};
10 ->
{ok,
{control,
{pong_frame, Payload_length, Data}}};
_ ->
{error, invalid_frame}
end,
gleam@result:map(
_pipe,
fun(Frame) -> case Complete of
1 ->
{{complete, Frame}, Rest@2};
0 ->
{{incomplete, Frame}, Rest@2}
end end
);
_ ->
_assert_subject = (erlang:element(
10,
erlang:element(3, Conn)
))(erlang:element(2, Conn), 0),
{ok, Data@1} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"mist/internal/websocket"/utf8>>,
function => <<"frame_from_message"/utf8>>,
line => 117})
end,
frame_from_message(
<<Message/bitstring, Data@1/bitstring>>,
Conn
)
end;
_ ->
{error, invalid_frame}
end;
_ ->
{error, invalid_frame}
end.
-spec make_frame(integer(), integer(), bitstring()) -> gleam@bytes_builder:bytes_builder().
make_frame(Opcode, Length, Payload) ->
Length_section = case Length of
Length@1 when Length@1 > 65535 ->
<<127:7, Length@1:64/integer>>;
Length@2 when Length@2 >= 126 ->
<<126:7, Length@2:16/integer>>;
_ ->
<<Length:7>>
end,
_pipe = <<1:1,
0:3,
Opcode:4,
0:1,
Length_section/bitstring,
Payload/bitstring>>,
gleam_stdlib:wrap_list(_pipe).
-spec frame_to_bytes_builder(frame()) -> gleam@bytes_builder:bytes_builder().
frame_to_bytes_builder(Frame) ->
case Frame of
{data, {text_frame, Payload_length, Payload}} ->
make_frame(1, Payload_length, Payload);
{control, {close_frame, Payload_length@1, Payload@1}} ->
make_frame(8, Payload_length@1, Payload@1);
{data, {binary_frame, Payload_length@2, Payload@2}} ->
make_frame(2, Payload_length@2, Payload@2);
{control, {pong_frame, Payload_length@3, Payload@3}} ->
make_frame(10, Payload_length@3, Payload@3);
{control, {ping_frame, Payload_length@4, Payload@4}} ->
make_frame(9, Payload_length@4, Payload@4);
{continuation, Length, Payload@5} ->
make_frame(0, Length, Payload@5)
end.
-spec to_text_frame(bitstring()) -> gleam@bytes_builder:bytes_builder().
to_text_frame(Data) ->
Size = erlang:byte_size(Data),
frame_to_bytes_builder({data, {text_frame, Size, Data}}).
-spec to_binary_frame(bitstring()) -> gleam@bytes_builder:bytes_builder().
to_binary_frame(Data) ->
Size = erlang:byte_size(Data),
frame_to_bytes_builder({data, {binary_frame, Size, Data}}).
-spec message_selector() -> gleam@erlang@process:selector(websocket_message(any())).
message_selector() ->
_pipe = gleam_erlang_ffi:new_selector(),
_pipe@6 = gleam@erlang@process:selecting_record3(
_pipe,
erlang:binary_to_atom(<<"tcp"/utf8>>),
fun(_, Data) -> _pipe@1 = Data,
_pipe@2 = gleam@dynamic:bit_array(_pipe@1),
_pipe@3 = gleam@result:replace_error(_pipe@2, nil),
_pipe@4 = gleam@result:map(
_pipe@3,
fun(Field@0) -> {socket_message, Field@0} end
),
_pipe@5 = gleam@result:map(
_pipe@4,
fun(Field@0) -> {valid, Field@0} end
),
gleam@result:unwrap(_pipe@5, invalid) end
),
_pipe@12 = gleam@erlang@process:selecting_record3(
_pipe@6,
erlang:binary_to_atom(<<"ssl"/utf8>>),
fun(_, Data@1) -> _pipe@7 = Data@1,
_pipe@8 = gleam@dynamic:bit_array(_pipe@7),
_pipe@9 = gleam@result:replace_error(_pipe@8, nil),
_pipe@10 = gleam@result:map(
_pipe@9,
fun(Field@0) -> {socket_message, Field@0} end
),
_pipe@11 = gleam@result:map(
_pipe@10,
fun(Field@0) -> {valid, Field@0} end
),
gleam@result:unwrap(_pipe@11, invalid) end
),
_pipe@13 = gleam@erlang@process:selecting_record2(
_pipe@12,
erlang:binary_to_atom(<<"ssl_closed"/utf8>>),
fun(_) -> {valid, socket_closed_message} end
),
gleam@erlang@process:selecting_record2(
_pipe@13,
erlang:binary_to_atom(<<"tcp_closed"/utf8>>),
fun(_) -> {valid, socket_closed_message} end
).
-spec get_messages(bitstring(), websocket_connection(), list(parsed_frame())) -> {list(parsed_frame()),
bitstring()}.
get_messages(Data, Conn, Frames) ->
case frame_from_message(Data, Conn) of
{ok, {Frame, <<>>}} ->
{gleam@list:reverse([Frame | Frames]), <<>>};
{ok, {Frame@1, Rest}} ->
get_messages(Rest, Conn, [Frame@1 | Frames]);
{error, {need_more_data, Rest@1}} ->
{Frames, Rest@1};
{error, invalid_frame} ->
{Frames, Data}
end.
-spec set_active(websocket_connection()) -> nil.
set_active(Connection) ->
_assert_subject = (erlang:element(13, erlang:element(3, Connection)))(
erlang:element(2, Connection),
[{active_mode, once}]
),
{ok, _} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"mist/internal/websocket"/utf8>>,
function => <<"set_active"/utf8>>,
line => 447})
end,
nil.
-spec map_user_selector(gleam@option:option(gleam@erlang@process:selector(JOG))) -> gleam@option:option(gleam@erlang@process:selector(websocket_message(JOG))).
map_user_selector(Selector) ->
gleam@option:map(
Selector,
fun(_capture) ->
gleam_erlang_ffi:map_selector(
_capture,
fun(Msg) -> {valid, {user_message, Msg}} end
)
end
).
-spec apply_frames(
list(frame()),
fun((JNQ, websocket_connection(), handler_message(JNR)) -> gleam@otp@actor:next(JNR, JNQ)),
websocket_connection(),
gleam@otp@actor:next(websocket_message(JNR), JNQ),
fun((JNQ) -> nil)
) -> gleam@otp@actor:next(websocket_message(JNR), JNQ).
apply_frames(Frames, Handler, Connection, Next, On_close) ->
case {Frames, Next} of
{_, {stop, Reason}} ->
{stop, Reason};
{[], Next@1} ->
set_active(Connection),
Next@1;
{[{control, {close_frame, _, _}} = Frame | _], {continue, State, _}} ->
_ = (erlang:element(12, erlang:element(3, Connection)))(
erlang:element(2, Connection),
frame_to_bytes_builder(Frame)
),
On_close(State),
{stop, normal};
{[{control, {ping_frame, Length, Payload}} | _], {continue, State@1, _}} ->
_pipe = (erlang:element(12, erlang:element(3, Connection)))(
erlang:element(2, Connection),
frame_to_bytes_builder({control, {pong_frame, Length, Payload}})
),
_pipe@1 = gleam@result:map(
_pipe,
fun(_) ->
set_active(Connection),
gleam@otp@actor:continue(State@1)
end
),
gleam@result:lazy_unwrap(
_pipe@1,
fun() ->
On_close(State@1),
{stop, {abnormal, <<"Failed to send pong frame"/utf8>>}}
end
);
{[Frame@1 | Rest], {continue, State@2, Prev_selector}} ->
case gleam_erlang_ffi:rescue(
fun() -> Handler(State@2, Connection, {internal, Frame@1}) end
) of
{ok, {continue, State@3, Selector}} ->
Next_selector = begin
_pipe@2 = Selector,
_pipe@3 = map_user_selector(_pipe@2),
_pipe@4 = gleam@option:'or'(_pipe@3, Prev_selector),
gleam@option:map(
_pipe@4,
fun(With_user) ->
gleam_erlang_ffi:merge_selector(
message_selector(),
With_user
)
end
)
end,
apply_frames(
Rest,
Handler,
Connection,
{continue, State@3, Next_selector},
On_close
);
{ok, {stop, Reason@1}} ->
On_close(State@2),
{stop, Reason@1};
{error, _} ->
mist@internal@logger:error(
<<"Caught error in websocket handler"/utf8>>
),
On_close(State@2),
{stop,
{abnormal, <<"Crash in user websocket handler"/utf8>>}}
end;
{_, _} ->
{stop, {abnormal, <<"Unexpected frame type or user state"/utf8>>}}
end.
-spec append_frame(frame(), integer(), bitstring()) -> frame().
append_frame(Left, Length, Data) ->
case Left of
{data, {text_frame, Len, Payload}} ->
{data,
{text_frame,
Len + Length,
<<Payload/bitstring, Data/bitstring>>}};
{data, {binary_frame, Len@1, Payload@1}} ->
{data,
{binary_frame,
Len@1 + Length,
<<Payload@1/bitstring, Data/bitstring>>}};
{control, {close_frame, Len@2, Payload@2}} ->
{control,
{close_frame,
Len@2 + Length,
<<Payload@2/bitstring, Data/bitstring>>}};
{control, {ping_frame, Len@3, Payload@3}} ->
{control,
{ping_frame,
Len@3 + Length,
<<Payload@3/bitstring, Data/bitstring>>}};
{control, {pong_frame, Len@4, Payload@4}} ->
{control,
{pong_frame,
Len@4 + Length,
<<Payload@4/bitstring, Data/bitstring>>}};
{continuation, _, _} ->
Left
end.
-spec aggregate_frames(
list(parsed_frame()),
gleam@option:option(frame()),
list(frame())
) -> {ok, list(frame())} | {error, nil}.
aggregate_frames(Frames, Previous, Joined) ->
case {Frames, Previous} of
{[], _} ->
{ok, gleam@list:reverse(Joined)};
{[{complete, {continuation, Length, Data}} | Rest], {some, Prev}} ->
Next = append_frame(Prev, Length, Data),
aggregate_frames(Rest, none, [Next | Joined]);
{[{incomplete, {continuation, Length@1, Data@1}} | Rest@1],
{some, Prev@1}} ->
Next@1 = append_frame(Prev@1, Length@1, Data@1),
aggregate_frames(Rest@1, {some, Next@1}, Joined);
{[{incomplete, Frame} | Rest@2], none} ->
aggregate_frames(Rest@2, {some, Frame}, Joined);
{[{complete, Frame@1} | Rest@3], none} ->
aggregate_frames(Rest@3, none, [Frame@1 | Joined]);
{_, _} ->
{error, nil}
end.
-spec initialize_connection(
fun((websocket_connection()) -> {JND,
gleam@option:option(gleam@erlang@process:selector(JNE))}),
fun((JND) -> nil),
fun((JND, websocket_connection(), handler_message(JNE)) -> gleam@otp@actor:next(JNE, JND)),
glisten@socket:socket(),
glisten@socket@transport:transport()
) -> {ok, gleam@erlang@process:subject(websocket_message(JNE))} | {error, nil}.
initialize_connection(On_init, On_close, Handler, Socket, Transport) ->
Connection = {websocket_connection, Socket, Transport},
_pipe@10 = gleam@otp@actor:start_spec(
{spec,
fun() ->
{Initial_state, User_selector} = On_init(Connection),
Selector = case User_selector of
{some, User_selector@1} ->
_pipe = User_selector@1,
_pipe@1 = gleam_erlang_ffi:map_selector(
_pipe,
fun(Field@0) -> {user_message, Field@0} end
),
_pipe@2 = gleam_erlang_ffi:map_selector(
_pipe@1,
fun(Field@0) -> {valid, Field@0} end
),
gleam_erlang_ffi:merge_selector(
_pipe@2,
message_selector()
);
_ ->
message_selector()
end,
{ready, {websocket_state, <<>>, Initial_state}, Selector}
end,
500,
fun(Msg, State) -> case Msg of
{valid, {socket_message, Data}} ->
{Frames, Rest} = get_messages(
<<(erlang:element(2, State))/bitstring,
Data/bitstring>>,
Connection,
[]
),
_pipe@3 = Frames,
_pipe@4 = aggregate_frames(_pipe@3, none, []),
_pipe@5 = gleam@result:map(
_pipe@4,
fun(Frames@1) ->
Next = apply_frames(
Frames@1,
Handler,
Connection,
gleam@otp@actor:continue(
erlang:element(3, State)
),
On_close
),
case Next of
{continue, User_state, Selector@1} ->
{continue,
{websocket_state, Rest, User_state},
Selector@1};
{stop, Reason} ->
{stop, Reason}
end
end
),
gleam@result:lazy_unwrap(
_pipe@5,
fun() ->
mist@internal@logger:error(
{<<"Received a malformed WebSocket frame"/utf8>>}
),
On_close(erlang:element(3, State)),
{stop,
{abnormal,
<<"WebSocket received a malformed message"/utf8>>}}
end
);
{valid, {user_message, Msg@1}} ->
_pipe@6 = gleam_erlang_ffi:rescue(
fun() ->
Handler(
erlang:element(3, State),
Connection,
{user, Msg@1}
)
end
),
_pipe@9 = gleam@result:map(
_pipe@6,
fun(Cont) -> case Cont of
{continue, User_state@1, Selector@2} ->
Selector@3 = begin
_pipe@7 = Selector@2,
_pipe@8 = map_user_selector(_pipe@7),
gleam@option:map(
_pipe@8,
fun(With_user) ->
gleam_erlang_ffi:merge_selector(
message_selector(),
With_user
)
end
)
end,
{continue,
erlang:setelement(
3,
State,
User_state@1
),
Selector@3};
{stop, Reason@1} ->
On_close(erlang:element(3, State)),
{stop, Reason@1}
end end
),
gleam@result:lazy_unwrap(
_pipe@9,
fun() ->
mist@internal@logger:error(
<<"Caught error in websocket handler"/utf8>>
),
On_close(erlang:element(3, State)),
{stop,
{abnormal,
<<"Crash in user websocket handler"/utf8>>}}
end
);
{valid, socket_closed_message} ->
On_close(erlang:element(3, State)),
{stop, normal};
invalid ->
mist@internal@logger:error(
{<<"Received a malformed WebSocket frame"/utf8>>}
),
On_close(erlang:element(3, State)),
{stop,
{abnormal,
<<"WebSocket received a malformed message"/utf8>>}}
end end}
),
_pipe@11 = gleam@result:replace_error(_pipe@10, nil),
_pipe@12 = gleam@result:map(
_pipe@11,
fun(Subj) ->
Websocket_pid = gleam@erlang@process:subject_owner(Subj),
_assert_subject = (erlang:element(5, erlang:element(3, Connection)))(
erlang:element(2, Connection),
Websocket_pid
),
{ok, _} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Assertion pattern match failed"/utf8>>,
value => _assert_fail,
module => <<"mist/internal/websocket"/utf8>>,
function => <<"initialize_connection"/utf8>>,
line => 330})
end,
set_active(Connection),
Subj
end
),
gleam@result:replace_error(_pipe@12, nil).