diff --git a/CHANGELOG.md b/CHANGELOG.md index 97e322e..914e5e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +# v2.0.3 + +- Replace `gramps` with `websocks` package +- Remove alias names for internal stream modules +- Improve script that change documentation + # v2.0.2 - Refactor internal handler code diff --git a/autobahn/config.json b/autobahn/config.json index 10f44b0..2603281 100644 --- a/autobahn/config.json +++ b/autobahn/config.json @@ -2,8 +2,8 @@ "outdir": "./reports/server", "servers": [ { - "url": "ws://localhost:8080", - "agent": "ewe" + "url": "ws://localhost:8081", + "agent": "unreleased" } ], "cases": [ diff --git a/gleam.toml b/gleam.toml index d2134d9..32a43ea 100644 --- a/gleam.toml +++ b/gleam.toml @@ -19,6 +19,7 @@ logging = ">= 1.3.0 and < 2.0.0" gleam_erlang = ">= 1.3.0 and < 2.0.0" gleam_crypto = ">= 1.5.1 and < 2.0.0" compresso = "0.1.0" +websocks = ">= 1.0.0 and < 2.0.0" # gramps = ">= 6.0.0 and < 7.0.0" [dev-dependencies] diff --git a/manifest.toml b/manifest.toml index 6d50e6a..81eacdd 100644 --- a/manifest.toml +++ b/manifest.toml @@ -16,6 +16,7 @@ packages = [ { name = "glisten", version = "8.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging", "telemetry"], otp_app = "glisten", source = "hex", outer_checksum = "534BB27C71FB9E506345A767C0D76B17A9E9199934340C975DC003C710E3692D" }, { name = "logging", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "1098FBF10B54B44C2C7FDF0B01C1253CAFACDACABEFB4B0D027803246753E06D" }, { name = "telemetry", version = "1.3.0", build_tools = ["rebar3"], requirements = [], otp_app = "telemetry", source = "hex", outer_checksum = "7015FC8919DBE63764F4B4B87A95B7C0996BD539E0D499BE6EC9D7F3875B79E6" }, + { name = "websocks", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "2046226F484F61D284879080AD2B968E57F411FDE8701C182E47DE85153C4B6B" }, ] [requirements] @@ -30,3 +31,4 @@ gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } glisten = { version = ">= 8.0.1 and < 9.0.0" } logging = { version = ">= 1.3.0 and < 2.0.0" } +websocks = { version = ">= 1.0.0 and < 2.0.0" } diff --git a/src/ewe.gleam b/src/ewe.gleam index 0579bd1..3dfb240 100644 --- a/src/ewe.gleam +++ b/src/ewe.gleam @@ -1,94 +1,139 @@ -//// //// -//// #### IP Address -//// - [ip_address_to_string](#ip_address_to_string) -//// #### Information -//// - [get_client_info](#get_client_info) -//// - [get_server_info](#get_server_info) -//// #### Builder -//// - [new](#new) -//// - [bind](#bind) -//// - [bind_all](#bind_all) -//// - [listening](#listening) -//// - [listening_random](#listening_random) -//// - [enable_ipv6](#enable_ipv6) -//// - [enable_tls](#enable_tls) -//// - [with_name](#with_name) -//// - [quiet](#quiet) -//// - [idle_timeout](#idle_timeout) -//// - [on_start](#on_start) -//// - [on_crash](#on_crash) -//// #### Server -//// - [start](#start) -//// - [supervised](#supervised) -//// #### Request -//// - [read_body](#read_body) -//// - [stream_body](#stream_body) -//// #### Response -//// - [file](#file) -//// #### Chunked Response -//// - [chunked_body](#chunked_body) -//// - [send_chunk](#send_chunk) -//// - [chunked_continue](#chunked_continue) -//// - [chunked_stop](#chunked_stop) -//// - [chunked_stop_abnormal](#chunked_stop_abnormal) -//// #### Websocket -//// - [upgrade_websocket](#upgrade_websocket) -//// - [send_binary_frame](#send_binary_frame) -//// - [send_text_frame](#send_text_frame) -//// - [websocket_continue](#websocket_continue) -//// - [websocket_continue_with_selector](#websocket_continue_with_selector) -//// - [websocket_stop](#websocket_stop) -//// - [websocket_stop_abnormal](#websocket_stop_abnormal) -//// #### Server-Sent Events -//// - [sse](#sse) -//// - [event](#event) -//// - [event_name](#event_name) -//// - [event_id](#event_id) -//// - [event_retry](#event_retry) -//// - [send_event](#send_event) -//// - [sse_continue](#sse_continue) -//// - [sse_stop](#sse_stop) -//// - [sse_stop_abnormal](#sse_stop_abnormal) // ----------------------------------------------------------------------------- // IMPORTS @@ -111,20 +156,19 @@ import gleam/result import gleam/string_tree.{type StringTree} import logging +import websocks + import glisten import glisten/internal/listener import glisten/socket/options as glisten_options import glisten/transport -// TODO: replace this once gramps changes are published -import ewe/internal/gramps/websocket as ws - import ewe/internal/file import ewe/internal/handler import ewe/internal/http1 as ewe_http -import ewe/internal/stream/chunked as ewe_chunked -import ewe/internal/stream/sse as ewe_sse -import ewe/internal/stream/websocket as ewe_ws +import ewe/internal/stream/chunked +import ewe/internal/stream/sse +import ewe/internal/stream/websocket // ----------------------------------------------------------------------------- // CONNECTION @@ -665,7 +709,7 @@ fn consumer_adapter( /// Represents a chunked response body. This type is used to send a chunked response to the client. pub type ChunkedBody = - ewe_chunked.ChunkedBody + chunked.ChunkedBody /// Represents an instruction on how chunked response should be processed. /// @@ -699,11 +743,11 @@ pub fn chunked_stop_abnormal(reason: String) -> ChunkedNext(user_state) { fn to_internal_chunked_next( next: ChunkedNext(user_state), -) -> ewe_chunked.ChunkedNext(user_state) { +) -> chunked.ChunkedNext(user_state) { case next { - ChunkedContinue(user_state) -> ewe_chunked.Continue(user_state) - ChunkedStop -> ewe_chunked.NormalStop - ChunkedAbnormalStop(reason) -> ewe_chunked.AbnormalStop(reason) + ChunkedContinue(user_state) -> chunked.Continue(user_state) + ChunkedStop -> chunked.NormalStop + ChunkedAbnormalStop(reason) -> chunked.AbnormalStop(reason) } } @@ -735,13 +779,13 @@ pub fn chunked_body( let socket = req.body.socket let factory_name = req.body.factory_name - case ewe_chunked.send_response(resp, transport, socket) { + case chunked.send_response(resp, transport, socket) { Ok(Nil) -> { let supervisor = factory.get_by_name(factory_name) let start_result = factory.start_child(supervisor, fn() { - ewe_chunked.start(transport, socket, on_init, handler, on_close) + chunked.start(transport, socket, on_init, handler, on_close) }) case start_result { @@ -762,7 +806,7 @@ pub fn send_chunk( body: ChunkedBody, chunk: BitArray, ) -> Result(Nil, glisten.SocketReason) { - ewe_chunked.send_chunk(body.transport, body.socket, chunk) + chunked.send_chunk(body.transport, body.socket, chunk) } // ----------------------------------------------------------------------------- @@ -772,7 +816,7 @@ pub fn send_chunk( /// Represents a WebSocket connection between a client and a server. /// pub type WebsocketConnection = - ewe_ws.WebsocketConnection + websocket.WebsocketConnection /// Represents an instruction on how WebSocket connection should proceed. /// @@ -822,12 +866,12 @@ pub fn websocket_stop_abnormal( fn to_internal_websocket_next( next: WebsocketNext(user_state, user_message), -) -> ewe_ws.WebsocketNext(user_state, user_message) { +) -> websocket.WebsocketNext(user_state, user_message) { case next { WebsocketContinue(user_state, selector) -> - ewe_ws.Continue(user_state, selector) - WebsocketNormalStop -> ewe_ws.NormalStop - WebsocketAbnormalStop(reason) -> ewe_ws.AbnormalStop(reason) + websocket.Continue(user_state, selector) + WebsocketNormalStop -> websocket.NormalStop + WebsocketAbnormalStop(reason) -> websocket.AbnormalStop(reason) } } @@ -846,20 +890,13 @@ pub type WebsocketMessage(user_message) { } fn transform_websocket_message( - message: ewe_ws.WebsocketMessage(user_message), + message: websocket.WebsocketMessage(user_message), ) -> Result(WebsocketMessage(user_message), Nil) { - // NOTE: see "https://github.com/rawhat/gramps/pull/7" case message { - ewe_ws.WebsocketFrame(ws.Data(frame)) -> { - ws.match_data_frame( - frame, - on_text: fn(payload, _) { - bit_array.to_string(payload) |> result.map(Text) - }, - on_binary: fn(payload, _) { Ok(Binary(payload)) }, - ) - } - ewe_ws.UserMessage(user_message) -> Ok(User(user_message)) + websocket.Frame(websocks.Text(payload)) -> + bit_array.to_string(payload) |> result.map(Text) + websocket.Frame(websocks.Binary(payload)) -> Ok(Binary(payload)) + websocket.UserMessage(user_message) -> Ok(User(user_message)) _ -> Error(Nil) } } @@ -905,7 +942,7 @@ pub fn upgrade_websocket( let supervisor = factory.get_by_name(factory_name) let start_result = factory.start_child(supervisor, fn() { - ewe_ws.start( + websocket.start( transport, socket, on_init, @@ -935,11 +972,11 @@ pub fn send_binary_frame( conn: WebsocketConnection, bits: BitArray, ) -> Result(Nil, glisten.SocketReason) { - ewe_ws.send_frame( - ws.encode_binary_frame, + websocket.send_frame( + websocks.encode_binary_frame, conn.transport, conn.socket, - conn.deflate, + conn.context, bits, ) } @@ -950,12 +987,12 @@ pub fn send_text_frame( conn: WebsocketConnection, text: String, ) -> Result(Nil, glisten.SocketReason) { - ewe_ws.send_frame( - ws.encode_text_frame, + websocket.send_frame( + websocks.encode_text_frame, conn.transport, conn.socket, - conn.deflate, - text, + conn.context, + bit_array.from_string(text), ) } @@ -966,7 +1003,7 @@ pub fn send_text_frame( /// Represents a Server-Sent Events connection between a client and a server. /// pub type SSEConnection = - ewe_sse.SSEConnection + sse.SSEConnection /// Represents an instruction on how Server-Sent Events connection should /// proceed. @@ -999,13 +1036,11 @@ pub fn sse_stop_abnormal(reason: String) -> SSENext(user_state) { SSEAbnormalStop(reason) } -fn to_internal_sse_next( - next: SSENext(user_state), -) -> ewe_sse.SSENext(user_state) { +fn to_internal_sse_next(next: SSENext(user_state)) -> sse.SSENext(user_state) { case next { - SSEContinue(user_state) -> ewe_sse.Continue(user_state) - SSENormalStop -> ewe_sse.NormalStop - SSEAbnormalStop(reason) -> ewe_sse.AbnormalStop(reason) + SSEContinue(user_state) -> sse.Continue(user_state) + SSENormalStop -> sse.NormalStop + SSEAbnormalStop(reason) -> sse.AbnormalStop(reason) } } @@ -1020,31 +1055,31 @@ fn to_internal_sse_next( /// `ewe.event_id`, and `ewe.event_retry`. /// pub type SSEEvent = - ewe_sse.SSEEvent + sse.SSEEvent /// Creates a new SSE event with the given data. Use `ewe.event_name`, /// `ewe.event_id`, and `ewe.event_retry` to modify other fields of the event. /// pub fn event(data: String) -> SSEEvent { - ewe_sse.SSEEvent(event: None, data:, id: None, retry: None) + sse.SSEEvent(event: None, data:, id: None, retry: None) } /// Sets the name of the event. /// pub fn event_name(event: SSEEvent, name: String) -> SSEEvent { - ewe_sse.SSEEvent(..event, event: Some(name)) + sse.SSEEvent(..event, event: Some(name)) } /// Sets the ID of the event. /// pub fn event_id(event: SSEEvent, id: String) -> SSEEvent { - ewe_sse.SSEEvent(..event, id: Some(id)) + sse.SSEEvent(..event, id: Some(id)) } /// Sets the retry time of the event. /// pub fn event_retry(event: SSEEvent, retry: Int) -> SSEEvent { - ewe_sse.SSEEvent(..event, retry: Some(retry)) + sse.SSEEvent(..event, retry: Some(retry)) } /// Sets up the connection for Server-Sent Events. @@ -1074,12 +1109,12 @@ pub fn sse( let socket = req.body.socket let factory_name = req.body.factory_name - case ewe_sse.send_response(transport, socket) { + case sse.send_response(transport, socket) { Ok(Nil) -> { let supervisor = factory.get_by_name(factory_name) let start_result = factory.start_child(supervisor, fn() { - ewe_sse.start(transport, socket, on_init, handler, on_close) + sse.start(transport, socket, on_init, handler, on_close) }) case start_result { @@ -1100,5 +1135,5 @@ pub fn send_event( conn: SSEConnection, event: SSEEvent, ) -> Result(Nil, glisten.SocketReason) { - ewe_sse.send_event(conn.transport, conn.socket, event) + sse.send_event(conn.transport, conn.socket, event) } diff --git a/src/ewe/internal/gramps/websocket.gleam b/src/ewe/internal/gramps/websocket.gleam deleted file mode 100644 index 17d9edc..0000000 --- a/src/ewe/internal/gramps/websocket.gleam +++ /dev/null @@ -1,678 +0,0 @@ -// TODO: remove this once gramps changes are published -// See https://github.com/rawhat/gramps - -import ewe/internal/gramps/websocket/compression.{ - type Context, type ContextTakeover, ContextTakeover, -} -import gleam/bit_array -import gleam/bool -import gleam/bytes_tree.{type BytesTree} -import gleam/crypto -import gleam/list -import gleam/option.{type Option, None, Some} -import gleam/result -import gleam/string - -pub opaque type DataFrame { - TextFrame(payload: BitArray) - BinaryFrame(payload: BitArray) - - CompressedTextFrame(payload: BitArray) - CompressedBinaryFrame(payload: BitArray) -} - -pub fn text_frame(payload: BitArray) -> DataFrame { - TextFrame(payload) -} - -pub fn binary_frame(payload: BitArray) -> DataFrame { - BinaryFrame(payload) -} - -pub fn match_data_frame( - data_frame: DataFrame, - on_text on_text: fn(BitArray, Bool) -> a, - on_binary on_binary: fn(BitArray, Bool) -> a, -) -> a { - case data_frame { - TextFrame(payload) -> on_text(payload, False) - CompressedTextFrame(payload) -> on_text(payload, True) - BinaryFrame(payload) -> on_binary(payload, False) - CompressedBinaryFrame(payload) -> on_binary(payload, True) - } -} - -pub type CloseReason { - NotProvided - Normal(body: BitArray) - GoingAway(body: BitArray) - ProtocolError(body: BitArray) - UnexpectedDataType(body: BitArray) - InconsistentDataType(body: BitArray) - PolicyViolation(body: BitArray) - MessageTooBig(body: BitArray) - MissingExtensions(body: BitArray) - UnexpectedCondition(body: BitArray) - /// Usually used for `4000` codes. - CustomCloseReason( - /// If `code >= 5000`, it will be the same as a `Normal` close reason. - code: Int, - body: BitArray, - ) -} - -pub type ControlFrame { - CloseFrame(reason: CloseReason) - PingFrame(payload: BitArray) - PongFrame(payload: BitArray) -} - -pub type Frame { - Data(DataFrame) - Control(ControlFrame) - Continuation(length: Int, payload: BitArray) -} - -@external(erlang, "crypto", "exor") -fn crypto_exor(a a: BitArray, b b: BitArray) -> BitArray - -fn mask_data(data: BitArray, masks: List(BitArray)) -> BitArray { - let assert [m1, m2, m3, m4] = masks - let mask_key = <> - - let payload_size = bit_array.byte_size(data) - let full_mask = create_repeating_mask(mask_key, payload_size) - crypto_exor(data, full_mask) -} - -fn create_repeating_mask(mask_key: BitArray, size: Int) -> BitArray { - case size { - 1 | 2 | 3 | 4 -> bit_array.slice(mask_key, 0, size) |> result.unwrap(<<>>) - - _ -> { - let repetitions = size / 4 - let remainder = size % 4 - let base = list.repeat(mask_key, repetitions) |> bit_array.concat - - case remainder { - 0 -> base - n -> { - let partial = bit_array.slice(mask_key, 0, n) |> result.unwrap(<<>>) - <> - } - } - } - } -} - -pub type FrameParseError { - NeedMoreData(BitArray) - InvalidFrame -} - -pub type ParsedFrame { - Complete(Frame) - Incomplete(Frame) -} - -pub fn decode_frame( - message: BitArray, - context: Option(Context), -) -> Result(#(ParsedFrame, BitArray), FrameParseError) { - case message { - << - complete:1, - compressed:1, - rsv2:1, - rsv3:1, - opcode:int-size(4), - masked:1, - payload_length:int-size(7), - rest:bits, - >> -> { - let compressed = compressed == 1 - let masked = masked == 1 - - use <- bool.guard( - when: compressed && option.is_none(context), - return: Error(InvalidFrame), - ) - - use <- bool.guard(rsv2 == 1 || rsv3 == 1, return: Error(InvalidFrame)) - - use <- bool.guard( - when: { - let is_control_frame = opcode >= 8 && opcode <= 10 - let is_fragmented = complete == 0 - is_control_frame && is_fragmented - }, - return: Error(InvalidFrame), - ) - - let payload_size = case payload_length { - 126 -> 16 - 127 -> 64 - _ -> 0 - } - - let maybe_pair = case masked, rest { - True, - << - length:int-size(payload_size), - mask1:bytes-size(1), - mask2:bytes-size(1), - mask3:bytes-size(1), - mask4:bytes-size(1), - rest:bits, - >> - -> { - let payload_byte_size = case length { - 0 -> payload_length - n -> n - } - - case bit_array.byte_size(rest) >= payload_byte_size, rest { - True, <> -> { - let data = mask_data(payload, [mask1, mask2, mask3, mask4]) - Ok(#(data, remaining)) - } - _, _ -> Error(NeedMoreData(message)) - } - } - True, _rest -> Error(NeedMoreData(message)) - False, <> -> { - let payload_byte_size = case length { - 0 -> payload_length - n -> n - } - case rest { - <> -> { - Ok(#(payload, rest)) - } - _ -> { - Error(NeedMoreData(message)) - } - } - } - _, _ -> Error(InvalidFrame) - } - - use #(data, rest) <- result.try(maybe_pair) - case opcode { - 0 -> Ok(Continuation(payload_size, data)) - 1 -> { - case compressed { - True -> Ok(Data(CompressedTextFrame(data))) - False -> Ok(Data(TextFrame(data))) - } - } - 2 -> { - case compressed { - True -> Ok(Data(CompressedBinaryFrame(data))) - False -> Ok(Data(BinaryFrame(data))) - } - } - 8 -> { - case data { - <<>> -> Ok(Control(CloseFrame(NotProvided))) - <> -> { - use <- bool.guard( - when: !bit_array.is_utf8(rest), - return: Error(InvalidFrame), - ) - - case code { - 1000 -> Ok(Control(CloseFrame(Normal(rest)))) - 1001 -> Ok(Control(CloseFrame(GoingAway(rest)))) - 1002 -> Ok(Control(CloseFrame(ProtocolError(rest)))) - 1003 -> Ok(Control(CloseFrame(UnexpectedDataType(rest)))) - 1007 -> Ok(Control(CloseFrame(InconsistentDataType(rest)))) - 1008 -> Ok(Control(CloseFrame(PolicyViolation(rest)))) - 1009 -> Ok(Control(CloseFrame(MessageTooBig(rest)))) - 1010 -> Ok(Control(CloseFrame(MissingExtensions(rest)))) - 1011 -> Ok(Control(CloseFrame(UnexpectedCondition(rest)))) - code if code >= 3000 && code <= 4999 -> - Ok(Control(CloseFrame(CustomCloseReason(code, rest)))) - _ -> Error(InvalidFrame) - } - } - _ -> Error(InvalidFrame) - } - } - 9 -> Ok(Control(PingFrame(data))) - 10 -> Ok(Control(PongFrame(data))) - _ -> Error(InvalidFrame) - } - |> result.try(fn(frame) { - case complete { - 1 -> Ok(#(Complete(frame), rest)) - 0 -> Ok(#(Incomplete(frame), rest)) - _ -> Error(InvalidFrame) - } - }) - } - _ -> Error(NeedMoreData(message)) - } -} - -pub fn encode_text_frame( - data: String, - context: Option(Context), - mask: Option(BitArray), -) -> BytesTree { - to_frame(bit_array.from_string(data), context, mask, TextFrame, Data) -} - -pub fn encode_binary_frame( - data: BitArray, - context: Option(Context), - mask: Option(BitArray), -) -> BytesTree { - to_frame(data, context, mask, BinaryFrame, Data) -} - -pub fn encode_close_frame( - reason: CloseReason, - mask: Option(BitArray), -) -> BytesTree { - encode_frame(Control(CloseFrame(reason)), Uncompressed, mask) -} - -pub fn encode_ping_frame(data: BitArray, mask: Option(BitArray)) -> BytesTree { - to_frame(data, None, mask, PingFrame, Control) -} - -pub fn encode_pong_frame(data: BitArray, mask: Option(BitArray)) -> BytesTree { - to_frame(data, None, mask, PongFrame, Control) -} - -pub fn encode_continuation_frame( - data: BitArray, - total_size: Int, - mask: Option(BitArray), -) -> BytesTree { - let payload = apply_mask(data, mask) - encode_frame(Continuation(total_size, payload), Uncompressed, mask) -} - -fn encode_frame( - frame: Frame, - compressed: Compression, - mask: Option(BitArray), -) -> BytesTree { - case frame { - Data(TextFrame(payload)) | Data(CompressedTextFrame(payload)) -> { - let payload_length = bit_array.byte_size(payload) - make_frame(1, payload_length, payload, compressed, mask) - } - - Data(BinaryFrame(payload)) | Data(CompressedBinaryFrame(payload)) -> { - let payload_length = bit_array.byte_size(payload) - make_frame(2, payload_length, payload, compressed, mask) - } - - Control(CloseFrame(reason)) -> { - let #(payload_length, payload) = case reason { - NotProvided -> #(0, <<>>) - GoingAway(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1001:16, body:bits>>) - } - InconsistentDataType(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1007:16, body:bits>>) - } - MessageTooBig(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1009:16, body:bits>>) - } - MissingExtensions(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1010:16, body:bits>>) - } - Normal(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1000:16, body:bits>>) - } - PolicyViolation(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1008:16, body:bits>>) - } - ProtocolError(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1002:16, body:bits>>) - } - UnexpectedCondition(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1011:16, body:bits>>) - } - UnexpectedDataType(body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - #(payload_size, <<1003:16, body:bits>>) - } - CustomCloseReason(code:, body:) -> { - let payload_size = bit_array.byte_size(body) + 2 - // Prevents integer overflow and changes the status code to `Normal` for invalid codes. - let code = case code < 5000 { - True -> code - False -> 1000 - } - #(payload_size, <>) - } - } - make_frame(8, payload_length, apply_mask(payload, mask), compressed, mask) - } - Control(PongFrame(payload)) -> { - let payload_length = bit_array.byte_size(payload) - make_frame(10, payload_length, payload, compressed, mask) - } - Control(PingFrame(payload)) -> { - let payload_length = bit_array.byte_size(payload) - make_frame(9, payload_length, payload, compressed, mask) - } - Continuation(length, payload) -> - make_frame(0, length, payload, compressed, mask) - } -} - -fn make_length(length: Int) -> BitArray { - case length { - length if length > 65_535 -> <<127:7, length:int-size(64)>> - length if length >= 126 -> <<126:7, length:int-size(16)>> - _length -> <> - } -} - -type Compression { - Compressed - Uncompressed -} - -fn make_frame( - opcode: Int, - length: Int, - payload: BitArray, - compressed: Compression, - mask: Option(BitArray), -) -> BytesTree { - let length_section = make_length(length) - - let masked = case option.is_some(mask) { - True -> 1 - False -> 0 - } - - let mask_key = option.unwrap(mask, <<>>) - - let compressed = case compressed { - Compressed -> 1 - Uncompressed -> 0 - } - - << - 1:1, - compressed:1, - 0:2, - opcode:4, - masked:1, - length_section:bits, - mask_key:bits, - payload:bits, - >> - |> bytes_tree.from_bit_array -} - -pub fn apply_mask(data: BitArray, mask: Option(BitArray)) -> BitArray { - case mask { - Some(mask) -> { - let assert << - mask1:bytes-size(1), - mask2:bytes-size(1), - mask3:bytes-size(1), - mask4:bytes-size(1), - >> = mask - mask_data(data, [mask1, mask2, mask3, mask4]) - } - None -> data - } -} - -pub fn apply_deflate(data: BitArray, context: Option(Context)) -> BitArray { - case context { - Some(context) -> compression.deflate(context, data) - _ -> data - } -} - -pub fn apply_inflate(data: BitArray, context: Option(Context)) -> BitArray { - case context { - Some(context) -> compression.inflate(context, data) - _ -> data - } -} - -fn to_frame( - data: BitArray, - context: Option(Context), - mask: Option(BitArray), - create_inner_frame: fn(BitArray) -> a, - create_frame: fn(a) -> Frame, -) -> BytesTree { - let frame = - data - |> apply_deflate(context) - |> apply_mask(mask) - |> create_inner_frame - |> create_frame - let compress = case context { - Some(_context) -> Compressed - _ -> Uncompressed - } - encode_frame(frame, compress, mask) -} - -pub fn decode_many_frames( - data: BitArray, - context: Option(Context), - frames: List(ParsedFrame), -) -> #(List(ParsedFrame), BitArray) { - case decode_frame(data, context) { - Ok(#(frame, <<>>)) -> #(list.reverse([frame, ..frames]), <<>>) - Ok(#(frame, rest)) -> decode_many_frames(rest, context, [frame, ..frames]) - Error(NeedMoreData(rest)) -> #(list.reverse(frames), rest) - Error(InvalidFrame) -> #(list.reverse(frames), data) - } -} - -pub type ManyFramesParseError { - NeedMoreDataAccumulated(parsed: List(ParsedFrame), rest: BitArray) - ContainsInvalidFrame -} - -pub fn decode_many_frames_result( - data: BitArray, - context: Option(Context), - frames: List(ParsedFrame), -) -> Result(#(List(ParsedFrame), BitArray), ManyFramesParseError) { - case decode_frame(data, context) { - Ok(#(frame, <<>>)) -> Ok(#(list.reverse([frame, ..frames]), <<>>)) - Ok(#(frame, rest)) -> - decode_many_frames_result(rest, context, [frame, ..frames]) - Error(NeedMoreData(rest)) -> - Error(NeedMoreDataAccumulated(list.reverse(frames), rest)) - Error(InvalidFrame) -> Error(ContainsInvalidFrame) - } -} - -pub fn aggregate_frames( - frames: List(ParsedFrame), - accumulated: Option(#(Frame, List(BitArray))), - joined: List(Frame), - context: Option(Context), -) -> Result(List(Frame), Nil) { - case frames, accumulated { - // No more frames - we are done - [], _ -> Ok(list.reverse(joined)) - - // Complete standalone frame - [Complete(Data(CompressedTextFrame(data))), ..rest], None -> { - case context { - Some(ctx) -> { - let decompressed = compression.inflate(ctx, data) - case bit_array.is_utf8(decompressed) { - True -> { - let final_frame = Data(TextFrame(decompressed)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - False -> Error(Nil) - } - } - None -> Error(Nil) - } - } - [Complete(Data(CompressedBinaryFrame(data))), ..rest], None -> { - case context { - Some(ctx) -> { - let decompressed = compression.inflate(ctx, data) - let final_frame = Data(BinaryFrame(decompressed)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - None -> Error(Nil) - } - } - [Complete(Data(TextFrame(data))), ..rest], None -> { - case bit_array.is_utf8(data) { - True -> - aggregate_frames( - rest, - None, - [Data(TextFrame(data)), ..joined], - context, - ) - False -> Error(Nil) - } - } - [Complete(Data(BinaryFrame(data))), ..rest], None -> - aggregate_frames(rest, None, [Data(BinaryFrame(data)), ..joined], context) - [Complete(Continuation(..)), ..], None -> Error(Nil) - [Complete(frame), ..rest], None -> - aggregate_frames(rest, None, [frame, ..joined], context) - - // Incomplete frame starting fragmentation - [Incomplete(frame), ..rest], None -> { - let initial_payload = case frame { - Data(TextFrame(payload)) -> payload - Data(BinaryFrame(payload)) -> payload - Data(CompressedTextFrame(payload)) -> payload - Data(CompressedBinaryFrame(payload)) -> payload - Continuation(_, payload) -> payload - Control(_) -> <<>> - } - aggregate_frames(rest, Some(#(frame, [initial_payload])), joined, context) - } - - // Complete continuation; finish fragmented message - [Complete(Continuation(payload: data, ..)), ..rest], - Some(#(initial_frame, payloads)) - -> { - let all_payloads = [data, ..payloads] |> list.reverse() - let final_payload = bit_array.concat(all_payloads) - - case initial_frame { - Data(CompressedTextFrame(_)) -> { - case context { - Some(ctx) -> { - let decompressed = compression.inflate(ctx, final_payload) - case bit_array.is_utf8(decompressed) { - True -> { - let final_frame = Data(TextFrame(decompressed)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - False -> Error(Nil) - } - } - None -> Error(Nil) - } - } - Data(CompressedBinaryFrame(_)) -> { - case context { - Some(ctx) -> { - let decompressed = compression.inflate(ctx, final_payload) - let final_frame = Data(BinaryFrame(decompressed)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - None -> Error(Nil) - } - } - Data(TextFrame(_)) -> { - case bit_array.is_utf8(final_payload) { - True -> { - let final_frame = Data(TextFrame(final_payload)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - False -> Error(Nil) - } - } - Data(BinaryFrame(_)) -> { - let final_frame = Data(BinaryFrame(final_payload)) - aggregate_frames(rest, None, [final_frame, ..joined], context) - } - Control(_) -> Error(Nil) - Continuation(..) -> Error(Nil) - } - } - - // Incomplete continuation; keep building the message - [Incomplete(Continuation(payload: data, ..)), ..rest], - Some(#(initial_frame, payloads)) - -> { - aggregate_frames( - rest, - Some(#(initial_frame, [data, ..payloads])), - joined, - context, - ) - } - - _, _ -> Error(Nil) - } -} - -const websocket_key = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" - -pub fn make_client_key() -> String { - let bytes = crypto.strong_random_bytes(16) - bit_array.base64_encode(bytes, True) -} - -type ShaHash { - Sha -} - -pub fn parse_websocket_key(key: String) -> String { - key - |> string.append(websocket_key) - |> crypto_hash(Sha, _) - |> base64_encode -} - -@external(erlang, "crypto", "hash") -fn crypto_hash(hash hash: ShaHash, data data: String) -> String - -@external(erlang, "base64", "encode") -fn base64_encode(data data: String) -> String - -pub fn has_deflate(extensions: List(String)) -> Bool { - list.any(extensions, fn(str) { str == "permessage-deflate" }) -} - -pub fn get_context_takeovers(extensions: List(String)) -> ContextTakeover { - let no_client_context_takeover = - list.any(extensions, fn(str) { str == "client_no_context_takeover" }) - let no_server_context_takeover = - list.any(extensions, fn(str) { str == "server_no_context_takeover" }) - ContextTakeover( - no_client: no_client_context_takeover, - no_server: no_server_context_takeover, - ) -} diff --git a/src/ewe/internal/gramps/websocket/compression.gleam b/src/ewe/internal/gramps/websocket/compression.gleam deleted file mode 100644 index 73f5478..0000000 --- a/src/ewe/internal/gramps/websocket/compression.gleam +++ /dev/null @@ -1,124 +0,0 @@ -// TODO: remove this once gramps changes are published -// See https://github.com/rawhat/gramps - -import gleam/bit_array -import gleam/bytes_tree.{type BytesTree} -import gleam/erlang/atom.{type Atom} -import gleam/erlang/process.{type Pid} - -pub type CompressionContext - -pub type Context { - Context(context: CompressionContext, no_takeover: Bool) -} - -type Flush { - Sync -} - -type Deflated { - Deflated -} - -type Default { - Default -} - -pub type ContextTakeover { - ContextTakeover(no_client: Bool, no_server: Bool) -} - -pub type Compression { - Compression(inflate: Context, deflate: Context) -} - -pub fn init(takeover: ContextTakeover) -> Compression { - let inflate = open() - let inflate_context = - Context(context: inflate, no_takeover: takeover.no_client) - - inflate_init(inflate, -15) - let deflate = open() - let deflate_context = - Context(context: deflate, no_takeover: takeover.no_server) - deflate_init(deflate, Default, Deflated, -15, 8, Default) - - Compression(inflate: inflate_context, deflate: deflate_context) -} - -@external(erlang, "zlib", "inflateInit") -fn inflate_init(context: CompressionContext, bits: Int) -> Atom - -@external(erlang, "zlib", "deflateInit") -fn deflate_init( - context: CompressionContext, - level: Default, - deflated: Deflated, - bits: Int, - mem_level: Int, - strategy: Default, -) -> Atom - -@external(erlang, "zlib", "open") -fn open() -> CompressionContext - -@external(erlang, "zlib", "inflate") -fn do_inflate(context: CompressionContext, data: BitArray) -> BytesTree - -pub fn inflate(context: Context, data: BitArray) -> BitArray { - let output = - context.context - |> do_inflate(<>) - |> bytes_tree.to_bit_array - - let _ = case context.no_takeover { - True -> inflate_reset(context.context) - False -> Nil - } - - output -} - -@external(erlang, "zlib", "deflate") -fn do_deflate( - context: CompressionContext, - data: BitArray, - flush: Flush, -) -> BytesTree - -pub fn deflate(context: Context, data: BitArray) -> BitArray { - let data = - context.context - |> do_deflate(data, Sync) - |> bytes_tree.to_bit_array - - let size = bit_array.byte_size(data) - 4 - - let return = case data { - <> -> value - _ -> data - } - - let _ = case context.no_takeover { - True -> deflate_reset(context.context) - False -> Nil - } - - return -} - -@external(erlang, "zlib", "set_controlling_process") -pub fn set_controlling_process(context: Context, pid: Pid) -> Atom - -pub fn close(context: Context) -> Nil { - do_close(context.context) -} - -@external(erlang, "zlib", "close") -fn do_close(context: CompressionContext) -> Nil - -@external(erlang, "zlib", "inflateReset") -fn inflate_reset(context: CompressionContext) -> Nil - -@external(erlang, "zlib", "deflateReset") -fn deflate_reset(context: CompressionContext) -> Nil diff --git a/src/ewe/internal/http1.gleam b/src/ewe/internal/http1.gleam index f46f3be..4503a9a 100644 --- a/src/ewe/internal/http1.gleam +++ b/src/ewe/internal/http1.gleam @@ -21,13 +21,12 @@ import gleam/string import gleam/string_tree.{type StringTree} import gleam/uri +import websocks + import glisten import glisten/socket.{type Socket} import glisten/transport.{type Transport} -// TODO: replace this once gramps changes are published -import ewe/internal/gramps/websocket as ws - import ewe/internal/buffer.{type Buffer} import ewe/internal/clock import ewe/internal/decoder.{ @@ -404,14 +403,14 @@ pub fn upgrade_websocket( |> result.replace_error(MissingWebsocketKey), ) - let accept_key = ws.parse_websocket_key(key) + let accept_key = websocks.compute_accept(key) let extensions = request.get_header(req, "sec-websocket-extensions") |> result.map(string.split(_, ";")) |> result.unwrap([]) - let permessage_deflate = ws.has_deflate(extensions) + let permessage_deflate = websocks.has_deflate(extensions) let resp = response.new(101) diff --git a/src/ewe/internal/stream/websocket.gleam b/src/ewe/internal/stream/websocket.gleam index 6dbfa0c..45f8852 100644 --- a/src/ewe/internal/stream/websocket.gleam +++ b/src/ewe/internal/stream/websocket.gleam @@ -2,25 +2,22 @@ // IMPORTS // ----------------------------------------------------------------------------- import gleam/bit_array -import gleam/bytes_tree.{type BytesTree} +import gleam/bytes_tree import gleam/dynamic/decode import gleam/erlang/atom import gleam/erlang/process.{type Selector, type Subject} -import gleam/list import gleam/option.{type Option, None, Some} import gleam/otp/actor import gleam/result import gleam/string import logging +import websocks + import glisten/socket.{type Socket, type SocketReason} import glisten/socket/options.{ActiveMode, Count} import glisten/transport.{type Transport} -// TODO: replace this once gramps changes are published -import ewe/internal/gramps/websocket.{type Frame, CloseFrame, PingFrame} -import ewe/internal/gramps/websocket/compression - import ewe/internal/exception // ----------------------------------------------------------------------------- @@ -32,13 +29,13 @@ pub type WebsocketConnection { WebsocketConnection( transport: Transport, socket: Socket, - deflate: Option(compression.Context), + context: websocks.Context, ) } // Messages that can be sent to or received from the WebSocket pub type WebsocketMessage(user_message) { - WebsocketFrame(Frame) + Frame(websocks.Frame) UserMessage(user_message) } @@ -55,12 +52,7 @@ pub type WebsocketNext(user_state, user_message) { // Internal state maintained by the WebSocket actor type WebsocketState(user_state) { - WebsocketState( - user_state: user_state, - per_message_deflate: Option(compression.Compression), - buffer: BitArray, - awaiting_frames: List(websocket.ParsedFrame), - ) + WebsocketState(user_state: user_state, context: websocks.Context) } // Type alias for actor next steps @@ -110,19 +102,19 @@ const non_owning_process = "Sending WebSocket message from non-owning process" // COMPRESSION UTILITIES // ----------------------------------------------------------------------------- -/// Gets the deflate context from the compression option -fn get_deflate( - compression: Option(compression.Compression), -) -> Option(compression.Context) { - option.map(compression, fn(compression) { compression.deflate }) -} +// /// Gets the deflate context from the compression option +// fn get_deflate( +// compression: Option(compression.Compression), +// ) -> Option(compression.Context) { +// option.map(compression, fn(compression) { compression.deflate }) +// } -/// Gets the inflate context from the compression option -fn get_inflate( - compression: Option(compression.Compression), -) -> Option(compression.Context) { - option.map(compression, fn(compression) { compression.inflate }) -} +// /// Gets the inflate context from the compression option +// fn get_inflate( +// compression: Option(compression.Compression), +// ) -> Option(compression.Context) { +// option.map(compression, fn(compression) { compression.inflate }) +// } // ----------------------------------------------------------------------------- // SELECTOR UTILITIES @@ -167,41 +159,8 @@ fn user_selector( // SOCKET UTILITIES // ----------------------------------------------------------------------------- -/// Sets socket to active mode for one message delivery -// fn set_socket_active_once(transport: Transport, socket: Socket) -> Nil { -// // echo transport.get_socket_opts(transport, socket, [atom.create("active")]) - -// let _ = transport.set_opts(transport, socket, [ActiveMode(Once)]) - -// // echo transport.get_socket_opts(transport, socket, [atom.create("active")]) - -// Nil -// } - const socket_active_count = 100 -// fn set_socket_active_smart( -// transport: Transport, -// socket: Socket, -// count: Int, -// ) -> Int { -// echo #( -// count, -// transport.get_socket_opts(transport, socket, [atom.create("active")]), -// ) - -// case count { -// 0 -> { -// let _ = -// transport.set_opts(transport, socket, [ -// ActiveMode(Count(socket_active_count)), -// ]) -// socket_active_count -// } -// _ -> count - 1 -// } -// } - // ----------------------------------------------------------------------------- // PUBLIC API // ----------------------------------------------------------------------------- @@ -217,47 +176,48 @@ pub fn start( permessage_deflate: Bool, ) -> Result(actor.Started(Nil), actor.StartError) { actor.new_with_initialiser(1000, fn(subject) { - let takeovers = websocket.get_context_takeovers(extensions) - let deflate = case permessage_deflate { - True -> Some(compression.init(takeovers)) + let context_takeovers = websocks.get_context_takeovers(extensions) + let compression = case permessage_deflate { + True -> Some(context_takeovers) False -> None } + let context = websocks.create_context(compression) - let conn = WebsocketConnection(transport, socket, get_deflate(deflate)) - - let #(user_state, user_selector) = on_init(conn, process.new_selector()) + let #(user_state, user_selector) = + WebsocketConnection(transport, socket, context) + |> on_init(process.new_selector()) let selector = process.map_selector(user_selector, User) |> process.merge_selector(create_socket_selector()) - let ws_state = - WebsocketState( - user_state:, - per_message_deflate: deflate, - buffer: <<>>, - awaiting_frames: [], - ) - - actor.initialised(ws_state) + WebsocketState(user_state:, context:) + |> actor.initialised() |> actor.selecting(selector) |> actor.returning(subject) |> Ok }) |> actor.on_message(fn(state, msg) { - let conn = - WebsocketConnection( - transport, - socket, - get_deflate(state.per_message_deflate), - ) - case msg { - Packet(data) -> handle_valid_packet(state, conn, data, handler, on_close) - Close -> handle_close(on_close, state, conn, None) + Packet(data) -> + handle_valid_packet(transport, socket, state, data, handler, on_close) User(user_message) -> - handle_user_message(state, conn, user_message, handler, on_close) - Invalid -> handle_close(on_close, state, conn, Some(malformed)) + handle_user_message( + transport, + socket, + state, + user_message, + handler, + on_close, + ) + Close -> { + let conn = WebsocketConnection(transport, socket, state.context) + handle_close(on_close, state, conn, None) + } + Invalid -> { + let conn = WebsocketConnection(transport, socket, state.context) + handle_close(on_close, state, conn, Some(malformed)) + } TcpPassive -> { let _ = transport.set_opts(transport, socket, [ @@ -273,15 +233,16 @@ pub fn start( /// Sends a frame to the WebSocket pub fn send_frame( - encoder: fn(data, Option(compression.Context), Option(BitArray)) -> BytesTree, + encoder: fn(BitArray, websocks.Context, Option(BitArray)) -> BitArray, transport: Transport, socket: Socket, - deflate: Option(compression.Context), - data: data, + context: websocks.Context, + payload: BitArray, ) -> Result(Nil, SocketReason) { let frame = exception.rescue(fn() { - encoder(data, deflate, option.None) + encoder(payload, context, option.None) + |> bytes_tree.from_bit_array() |> transport.send(transport, socket, _) }) @@ -304,86 +265,37 @@ pub fn send_frame( /// Handles incoming packet data, decoding frames and processing them fn handle_valid_packet( + transport: Transport, + socket: Socket, state: WebsocketState(user_state), - conn: WebsocketConnection, data: BitArray, handler: Handler(user_state, user_message), on_close: OnClose(user_state), ) -> ActorNext(user_state, user_message) { - let buffer = <> - - let decoded = - websocket.decode_many_frames_result( - buffer, - get_inflate(state.per_message_deflate), - [], - ) + let decoded = websocks.decode_many_frames(data, state.context) + + // NOTE: I was doing that before + // let #(data_frames, control_frames) = separate_frames(frames, [], []) + + // let control_result = case control_frames { + // [] -> Continue(state.user_state, None) + // _ -> + // loop_by_frames( + // control_frames, + // conn, + // handler, + // Continue(state.user_state, None), + // ) + // } case decoded { - Ok(#(frames, rest)) -> - handle_frames_processing(state, conn, frames, rest, handler, on_close) - Error(websocket.NeedMoreDataAccumulated(parsed, rest)) -> { - actor.continue( - WebsocketState( - ..state, - buffer: rest, - // NOTE: idk if its correct - awaiting_frames: list.append(state.awaiting_frames, parsed), - ), - ) - } - Error(websocket.ContainsInvalidFrame) -> { - handle_close(on_close, state, conn, Some(malformed)) - } - } -} - -/// Handles frames processing -fn handle_frames_processing( - state: WebsocketState(user_state), - conn: WebsocketConnection, - frames: List(websocket.ParsedFrame), - rest: BitArray, - handler: Handler(user_state, user_message), - on_close: OnClose(user_state), -) { - let frames = list.append(state.awaiting_frames, frames) - - let #(data_frames, control_frames) = separate_frames(frames, [], []) - - let control_result = case control_frames { - [] -> Continue(state.user_state, None) - _ -> - loop_by_frames( - control_frames, - conn, - handler, - Continue(state.user_state, None), - ) - } - - case control_result { - NormalStop -> handle_close(on_close, state, conn, None) - AbnormalStop(reason) -> handle_close(on_close, state, conn, Some(reason)) - Continue(_, _) -> { - let aggregated = - websocket.aggregate_frames( - data_frames, - None, - [], - get_inflate(state.per_message_deflate), - ) - - case aggregated { - Ok([]) -> { - actor.continue( - WebsocketState(..state, buffer: rest, awaiting_frames: data_frames), - ) - } - Ok(data_frames) -> { + Ok(#(decoded_frames, context)) -> { + case websocks.resolve_fragments(decoded_frames, context) { + Ok(#(resolved_frames, context)) -> { + let conn = WebsocketConnection(transport, socket, context) let next = - loop_by_frames( - data_frames, + handle_frames( + resolved_frames, conn, handler, Continue(state.user_state, None), @@ -391,15 +303,7 @@ fn handle_frames_processing( case next { Continue(user_state, selector) -> { - let next = - actor.continue( - WebsocketState( - ..state, - user_state:, - buffer: rest, - awaiting_frames: [], - ), - ) + let next = actor.continue(WebsocketState(user_state:, context:)) case selector { Some(selector) -> actor.with_selector(next, selector) @@ -411,33 +315,41 @@ fn handle_frames_processing( handle_close(on_close, state, conn, Some(reason)) } } - Error(Nil) -> handle_close(on_close, state, conn, Some(malformed)) + Error(violation) -> { + echo violation as "violation during frame resolving" + let conn = WebsocketConnection(transport, socket, context) + handle_close(on_close, state, conn, Some(malformed)) + } } } + Error(Nil) -> { + let conn = WebsocketConnection(transport, socket, state.context) + handle_close(on_close, state, conn, Some(malformed)) + } } } /// Separates frames into data and control frames -fn separate_frames( - frames: List(websocket.ParsedFrame), - data_frames: List(websocket.ParsedFrame), - control_frames: List(websocket.Frame), -) -> #(List(websocket.ParsedFrame), List(websocket.Frame)) { - case frames { - [] -> #(list.reverse(data_frames), list.reverse(control_frames)) - [websocket.Complete(websocket.Control(control_frame)), ..rest] -> - separate_frames(rest, data_frames, [ - websocket.Control(control_frame), - ..control_frames - ]) - [data_frame, ..rest] -> - separate_frames(rest, [data_frame, ..data_frames], control_frames) - } -} +// fn separate_frames( +// frames: List(websocket.ParsedFrame), +// data_frames: List(websocket.ParsedFrame), +// control_frames: List(websocket.Frame), +// ) -> #(List(websocket.ParsedFrame), List(websocket.Frame)) { +// case frames { +// [] -> #(list.reverse(data_frames), list.reverse(control_frames)) +// [websocket.Complete(websocket.Control(control_frame)), ..rest] -> +// separate_frames(rest, data_frames, [ +// websocket.Control(control_frame), +// ..control_frames +// ]) +// [data_frame, ..rest] -> +// separate_frames(rest, [data_frame, ..data_frames], control_frames) +// } +// } /// Processes a list of frames sequentially -fn loop_by_frames( - frames: List(Frame), +fn handle_frames( + frames: List(websocks.Frame), conn: WebsocketConnection, handler: Handler(user_state, user_message), next: WebsocketNext(user_state, InternalMessage(user_message)), @@ -451,7 +363,7 @@ fn loop_by_frames( [], next -> next // Control frames - [websocket.Control(PingFrame(payload)), ..rest], Continue(user_state, _) -> { + [websocks.Ping(payload), ..rest], Continue(user_state, _) -> { case bit_array.byte_size(payload) { size if size > 125 -> AbnormalStop( @@ -462,39 +374,34 @@ fn loop_by_frames( transport.send( conn.transport, conn.socket, - websocket.encode_pong_frame(payload, None), + websocks.encode_pong_frame(payload, None) + |> bytes_tree.from_bit_array(), ) case sent { Ok(Nil) -> - loop_by_frames(rest, conn, handler, Continue(user_state, None)) + handle_frames(rest, conn, handler, Continue(user_state, None)) Error(_) -> AbnormalStop(failed_pong) } } } } - [websocket.Control(CloseFrame(reason)), ..], Continue(_, _) -> { + [websocks.Close(reason), ..], Continue(..) -> { let _ = transport.send( conn.transport, conn.socket, - websocket.encode_close_frame(reason, None), + websocks.encode_close_frame(reason, None) + |> bytes_tree.from_bit_array(), ) NormalStop } - // NOTE: unsure if its should be here - [websocket.Continuation(_, _), ..], Continue(_, _) -> { - AbnormalStop("Unexpected continuation frame") - } - // Data frames [frame, ..rest], Continue(user_state, selector) -> { let call = - exception.rescue(fn() { - handler(conn, user_state, WebsocketFrame(frame)) - }) + exception.rescue(fn() { handler(conn, user_state, Frame(frame)) }) case call { Ok(Continue(user_state, new_selector)) -> { @@ -503,7 +410,7 @@ fn loop_by_frames( |> option.or(selector) |> option.map(process.merge_selector(create_socket_selector(), _)) - loop_by_frames( + handle_frames( rest, conn, handler, @@ -520,12 +427,14 @@ fn loop_by_frames( /// Handles user messages sent to the WebSocket fn handle_user_message( + transport: Transport, + socket: Socket, state: WebsocketState(user_state), - conn: WebsocketConnection, user_message: user_message, handler: Handler(user_state, user_message), on_close: OnClose(user_state), ) -> ActorNext(user_state, user_message) { + let conn = WebsocketConnection(transport, socket, state.context) let call = exception.rescue(fn() { handler(conn, state.user_state, UserMessage(user_message)) @@ -563,11 +472,7 @@ fn handle_close( conn: WebsocketConnection, abnormal_reason: Option(String), ) -> actor.Next(WebsocketState(user_state), InternalMessage(user_message)) { - option.map(state.per_message_deflate, fn(compression) { - compression.close(compression.deflate) - compression.close(compression.inflate) - }) - + websocks.close_context(state.context) on_close(conn, state.user_state) case abnormal_reason { diff --git a/test/autobahn.gleam b/test/autobahn.gleam index 865d3f9..e5a77c7 100644 --- a/test/autobahn.gleam +++ b/test/autobahn.gleam @@ -30,7 +30,7 @@ pub fn main() -> Nil { }) |> ewe.enable_ipv6() |> ewe.bind_all() - |> ewe.listening(port: 8080) + |> ewe.listening(port: 8081) |> ewe.supervised() let assert Ok(_) = -- 2.51.2 From 48d45c361b8dc58595af8722bd5d969fe5418240 Mon Sep 17 00:00:00 2001 From: vshakitskiy Date: Wed, 5 Nov 2025 22:28:13 +0300 Subject: [PATCH 2/3] addapt to updated websocks API --- autobahn/config.json | 7 +- src/ewe/internal/stream/websocket.gleam | 179 ++++++++++-------------- test/autobahn.gleam | 2 +- 3 files changed, 81 insertions(+), 107 deletions(-) diff --git a/autobahn/config.json b/autobahn/config.json index 2603281..d051301 100644 --- a/autobahn/config.json +++ b/autobahn/config.json @@ -2,14 +2,17 @@ "outdir": "./reports/server", "servers": [ { - "url": "ws://localhost:8081", + "url": "ws://localhost:8080", "agent": "unreleased" } ], "cases": [ "*" ], - "exclude-cases": [], + "exclude-cases": [ + "12.*", + "13.*" + ], "exclude-agent-cases": {}, "optioms": { "failByDrop": false diff --git a/src/ewe/internal/stream/websocket.gleam b/src/ewe/internal/stream/websocket.gleam index 45f8852..ca97264 100644 --- a/src/ewe/internal/stream/websocket.gleam +++ b/src/ewe/internal/stream/websocket.gleam @@ -41,7 +41,7 @@ pub type WebsocketMessage(user_message) { // Control flow for WebSocket message handling pub type WebsocketNext(user_state, user_message) { - Continue(user_state, Option(Selector(user_message))) + Continue(user_state: user_state, selector: Option(Selector(user_message))) NormalStop AbnormalStop(reason: String) } @@ -272,136 +272,108 @@ fn handle_valid_packet( handler: Handler(user_state, user_message), on_close: OnClose(user_state), ) -> ActorNext(user_state, user_message) { - let decoded = websocks.decode_many_frames(data, state.context) - - // NOTE: I was doing that before - // let #(data_frames, control_frames) = separate_frames(frames, [], []) - - // let control_result = case control_frames { - // [] -> Continue(state.user_state, None) - // _ -> - // loop_by_frames( - // control_frames, - // conn, - // handler, - // Continue(state.user_state, None), - // ) - // } - - case decoded { - Ok(#(decoded_frames, context)) -> { - case websocks.resolve_fragments(decoded_frames, context) { - Ok(#(resolved_frames, context)) -> { - let conn = WebsocketConnection(transport, socket, context) - let next = - handle_frames( - resolved_frames, - conn, - handler, - Continue(state.user_state, None), - ) - - case next { - Continue(user_state, selector) -> { - let next = actor.continue(WebsocketState(user_state:, context:)) - - case selector { - Some(selector) -> actor.with_selector(next, selector) - None -> next - } - } - NormalStop -> handle_close(on_close, state, conn, None) - AbnormalStop(reason) -> - handle_close(on_close, state, conn, Some(reason)) + let conn = WebsocketConnection(transport, socket, state.context) + let result = + websocks.process_incomming_frames( + data, + state.context, + ResolveState( + socket:, + transport:, + handler:, + next: Continue(state.user_state, None), + ), + handle_frame, + ) + + case result { + Ok(#(resolved_state, context)) -> { + case resolved_state.next { + Continue(user_state, selector) -> { + let next = actor.continue(WebsocketState(user_state:, context:)) + + case selector { + Some(selector) -> actor.with_selector(next, selector) + None -> next } } - Error(violation) -> { - echo violation as "violation during frame resolving" - let conn = WebsocketConnection(transport, socket, context) - handle_close(on_close, state, conn, Some(malformed)) - } + NormalStop -> handle_close(on_close, state, conn, None) + AbnormalStop(reason) -> + handle_close(on_close, state, conn, Some(reason)) } } - Error(Nil) -> { - let conn = WebsocketConnection(transport, socket, state.context) + Error(violation) -> { + echo violation as "violation during frame resolving" handle_close(on_close, state, conn, Some(malformed)) } } } -/// Separates frames into data and control frames -// fn separate_frames( -// frames: List(websocket.ParsedFrame), -// data_frames: List(websocket.ParsedFrame), -// control_frames: List(websocket.Frame), -// ) -> #(List(websocket.ParsedFrame), List(websocket.Frame)) { -// case frames { -// [] -> #(list.reverse(data_frames), list.reverse(control_frames)) -// [websocket.Complete(websocket.Control(control_frame)), ..rest] -> -// separate_frames(rest, data_frames, [ -// websocket.Control(control_frame), -// ..control_frames -// ]) -// [data_frame, ..rest] -> -// separate_frames(rest, [data_frame, ..data_frames], control_frames) -// } -// } +type ResolveState(user_state, user_message) { + ResolveState( + socket: Socket, + transport: Transport, + handler: Handler(user_state, user_message), + next: WebsocketNext(user_state, InternalMessage(user_message)), + ) +} /// Processes a list of frames sequentially -fn handle_frames( - frames: List(websocks.Frame), - conn: WebsocketConnection, - handler: Handler(user_state, user_message), - next: WebsocketNext(user_state, InternalMessage(user_message)), -) -> WebsocketNext(user_state, InternalMessage(user_message)) { - case frames, next { - // Early termination cases - _, NormalStop -> NormalStop - _, AbnormalStop(reason) -> AbnormalStop(reason) - - // No more frames - finish - [], next -> next - - // Control frames - [websocks.Ping(payload), ..rest], Continue(user_state, _) -> { +fn handle_frame( + state: ResolveState(user_state, user_message), + context: websocks.Context, + frame: websocks.Frame, +) -> websocks.ResolveNext(ResolveState(user_state, user_message)) { + case frame { + websocks.Control(websocks.Ping(payload)) -> { case bit_array.byte_size(payload) { size if size > 125 -> - AbnormalStop( - "control frames are only allowed to have payload up to and including 125 octets", + websocks.Stop( + ResolveState( + ..state, + next: AbnormalStop( + "control frames are only allowed to have payload up to and including 125 octets", + ), + ), ) _ -> { let sent = transport.send( - conn.transport, - conn.socket, + state.transport, + state.socket, websocks.encode_pong_frame(payload, None) |> bytes_tree.from_bit_array(), ) case sent { - Ok(Nil) -> - handle_frames(rest, conn, handler, Continue(user_state, None)) - Error(_) -> AbnormalStop(failed_pong) + Ok(Nil) -> websocks.Continue(state) + Error(_) -> + websocks.Stop( + ResolveState(..state, next: AbnormalStop(failed_pong)), + ) } } } } - [websocks.Close(reason), ..], Continue(..) -> { + websocks.Control(websocks.Close(reason)) -> { let _ = transport.send( - conn.transport, - conn.socket, + state.transport, + state.socket, websocks.encode_close_frame(reason, None) |> bytes_tree.from_bit_array(), ) - NormalStop + websocks.Stop(ResolveState(..state, next: NormalStop)) } - // Data frames - [frame, ..rest], Continue(user_state, selector) -> { + frame -> { + let assert Continue(user_state, selector) = state.next + + let conn = WebsocketConnection(state.transport, state.socket, context) + let call = - exception.rescue(fn() { handler(conn, user_state, Frame(frame)) }) + exception.rescue(fn() { state.handler(conn, user_state, Frame(frame)) }) case call { Ok(Continue(user_state, new_selector)) -> { @@ -410,16 +382,15 @@ fn handle_frames( |> option.or(selector) |> option.map(process.merge_selector(create_socket_selector(), _)) - handle_frames( - rest, - conn, - handler, - Continue(user_state, next_selector), + websocks.Continue( + ResolveState(..state, next: Continue(user_state, next_selector)), ) } - Ok(NormalStop) -> NormalStop - Ok(AbnormalStop(reason)) -> AbnormalStop(reason) - Error(_) -> AbnormalStop(crashed) + Ok(NormalStop) -> websocks.Stop(ResolveState(..state, next: NormalStop)) + Ok(AbnormalStop(reason)) -> + websocks.Stop(ResolveState(..state, next: AbnormalStop(reason))) + Error(_) -> + websocks.Stop(ResolveState(..state, next: AbnormalStop(crashed))) } } } diff --git a/test/autobahn.gleam b/test/autobahn.gleam index e5a77c7..865d3f9 100644 --- a/test/autobahn.gleam +++ b/test/autobahn.gleam @@ -30,7 +30,7 @@ pub fn main() -> Nil { }) |> ewe.enable_ipv6() |> ewe.bind_all() - |> ewe.listening(port: 8081) + |> ewe.listening(port: 8080) |> ewe.supervised() let assert Ok(_) = -- 2.51.2 From 35715edbd489c578358a3c85b85d7304a65b71c1 Mon Sep 17 00:00:00 2001 From: vshakitskiy Date: Thu, 6 Nov 2025 15:44:57 +0300 Subject: [PATCH 3/3] adjust websocks package version --- CHANGELOG.md | 1 + autobahn/config.json | 7 ++----- gleam.toml | 5 ++--- manifest.toml | 6 +++--- src/ewe/internal/stream/websocket.gleam | 2 +- 5 files changed, 9 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 914e5e5..1239334 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ - Replace `gramps` with `websocks` package - Remove alias names for internal stream modules - Improve script that change documentation +- Move `gleam_crypto` package as dev dependency # v2.0.2 diff --git a/autobahn/config.json b/autobahn/config.json index d051301..10f44b0 100644 --- a/autobahn/config.json +++ b/autobahn/config.json @@ -3,16 +3,13 @@ "servers": [ { "url": "ws://localhost:8080", - "agent": "unreleased" + "agent": "ewe" } ], "cases": [ "*" ], - "exclude-cases": [ - "12.*", - "13.*" - ], + "exclude-cases": [], "exclude-agent-cases": {}, "optioms": { "failByDrop": false diff --git a/gleam.toml b/gleam.toml index 32a43ea..33ceabd 100644 --- a/gleam.toml +++ b/gleam.toml @@ -17,15 +17,14 @@ gleam_otp = ">= 1.1.0 and < 2.0.0" gleam_http = ">= 4.3.0 and < 5.0.0" logging = ">= 1.3.0 and < 2.0.0" gleam_erlang = ">= 1.3.0 and < 2.0.0" -gleam_crypto = ">= 1.5.1 and < 2.0.0" compresso = "0.1.0" -websocks = ">= 1.0.0 and < 2.0.0" -# gramps = ">= 6.0.0 and < 7.0.0" +websocks = ">= 2.0.0 and < 3.0.0" [dev-dependencies] gleeunit = ">= 1.0.0 and < 2.0.0" gleam_httpc = ">= 5.0.0 and < 6.0.0" gleam_json = ">= 3.0.2 and < 4.0.0" +gleam_crypto = ">= 1.5.1 and < 2.0.0" [erlang] application_start_module = "ewe@internal@clock" diff --git a/manifest.toml b/manifest.toml index 81eacdd..ed3b77a 100644 --- a/manifest.toml +++ b/manifest.toml @@ -16,12 +16,11 @@ packages = [ { name = "glisten", version = "8.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging", "telemetry"], otp_app = "glisten", source = "hex", outer_checksum = "534BB27C71FB9E506345A767C0D76B17A9E9199934340C975DC003C710E3692D" }, { name = "logging", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "1098FBF10B54B44C2C7FDF0B01C1253CAFACDACABEFB4B0D027803246753E06D" }, { name = "telemetry", version = "1.3.0", build_tools = ["rebar3"], requirements = [], otp_app = "telemetry", source = "hex", outer_checksum = "7015FC8919DBE63764F4B4B87A95B7C0996BD539E0D499BE6EC9D7F3875B79E6" }, - { name = "websocks", version = "1.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "2046226F484F61D284879080AD2B968E57F411FDE8701C182E47DE85153C4B6B" }, + { name = "websocks", version = "2.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "A13BF89A8AC63C478C0E9FE502C81A6CBB0513F427FA5C1DFD96383BF46D501D" }, ] [requirements] compresso = { version = "0.1.0" } -gleam_crypto = { version = ">= 1.5.1 and < 2.0.0" } gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } gleam_http = { version = ">= 4.3.0 and < 5.0.0" } gleam_httpc = { version = ">= 5.0.0 and < 6.0.0" } @@ -31,4 +30,5 @@ gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" } gleeunit = { version = ">= 1.0.0 and < 2.0.0" } glisten = { version = ">= 8.0.1 and < 9.0.0" } logging = { version = ">= 1.3.0 and < 2.0.0" } -websocks = { version = ">= 1.0.0 and < 2.0.0" } +websocks = { version = ">= 2.0.0 and < 3.0.0" } +gleam_crypto = { version = ">= 1.5.1 and < 2.0.0" } diff --git a/src/ewe/internal/stream/websocket.gleam b/src/ewe/internal/stream/websocket.gleam index ca97264..5f6a188 100644 --- a/src/ewe/internal/stream/websocket.gleam +++ b/src/ewe/internal/stream/websocket.gleam @@ -274,7 +274,7 @@ fn handle_valid_packet( ) -> ActorNext(user_state, user_message) { let conn = WebsocketConnection(transport, socket, state.context) let result = - websocks.process_incomming_frames( + websocks.process_incoming_frames( data, state.context, ResolveState(