diff --git a/README.md b/README.md index 215515d..a1d919f 100644 --- a/README.md +++ b/README.md @@ -10,457 +10,499 @@ ewe [/juː/] - fluffy package for building web servers. ## Installation ```sh -gleam add ewe@4 gleam_erlang gleam_otp gleam_http logging +gleam add ewe@5 gleam_erlang gleam_otp gleam_http logging ``` ## Getting Started ```gleam +import ewe import gleam/erlang/process -import logging +import gleam/http/request import gleam/http/response - -import ewe.{type Request, type Response} +import logging pub fn main() { logging.configure() logging.set_level(logging.Info) + // The acceptor pool wires the listener and the connection factory together + // through process names. Create them where your program starts and pass them + // in here. + // + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.start process.sleep_forever() } -fn handler(_req: Request) -> Response { +fn handle_request( + _request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + // When sending a body it is important to include a `content-type` header. + // You never set `content-length` or `transfer-encoding` yourself, ewe frames + // the response and writes them for you. + // response.new(200) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Hello, World!")) + |> response.set_body(ewe.Text("Hello, World!")) } ``` +A handler takes a [`request.Request(ewe.Connection)`](https://hexdocs.pm/ewe/ewe.html#Connection) +and returns a [`response.Response(ewe.Body)`](https://hexdocs.pm/ewe/ewe.html#Body). +The connection carried by the request is what [`ewe.read_body`](https://hexdocs.pm/ewe/ewe.html#read_body), +[`ewe.file`](https://hexdocs.pm/ewe/ewe.html#file) and [`ewe.websocket`](https://hexdocs.pm/ewe/ewe.html#websocket) +work on. + +Instead of a port you can bind a unix domain socket with [`ewe.unix`](https://hexdocs.pm/ewe/ewe.html#unix), +or let the OS pick a free port with [`ewe.listening_random`](https://hexdocs.pm/ewe/ewe.html#listening_random) +and ask for the one it picked with [`ewe.get_server_info`](https://hexdocs.pm/ewe/ewe.html#get_server_info). + ## Usage ### [HTTPS](examples/src/https.gleam) -To enable HTTPS support via TLS, use [`ewe.enable_tls`](https://hexdocs.pm/ewe/ewe.html#enable_tls) with paths to your certificate and key files. The server validates the certificate and key files on startup and will crash if they're missing or invalid. +Enable TLS with [`ewe.with_tls`](https://hexdocs.pm/ewe/ewe.html#with_tls), which +takes the certificate source as a [`ewe.Tls`](https://hexdocs.pm/ewe/ewe.html#Tls) +value. The certificate and key are validated on startup and the server crashes if +they are missing or invalid. ```gleam -ewe.new(handler) -|> ewe.bind("0.0.0.0") -|> ewe.listening(port: 8080) -|> ewe.enable_tls( - certificate_file: "priv/localhost.crt", - key_file: "priv/localhost.key", -) +ewe.new(listener_name:, connection_factory_name:, handler: handle_request) +|> ewe.bind(to: "0.0.0.0") +|> ewe.listening(on: 8080) +// Certificate and key files on disk. +|> ewe.with_tls(ewe.Disk("priv/localhost.crt", "priv/localhost.key")) +// Or PEM already in memory: ewe.Pem(cert, key) +// Or DER in memory: ewe.Der(cert, key, ewe.RsaPrivateKey) |> ewe.start ``` -### [Sending Response](examples/src/sending_response.gleam) +To refuse clients that do not present a certificate signed by an authority you +name, add [`ewe.with_client_verification`](https://hexdocs.pm/ewe/ewe.html#with_client_verification). +It needs TLS to be configured. -`ewe` provides several response body types (see [`ewe.ResponseBody`](https://hexdocs.pm/ewe/ewe.html#ResponseBody) type). Request handler must return [`response.Response`](https://hexdocs.pm/gleam_http/gleam/http/response.html#Response) type with [`ewe.ResponseBody`](https://hexdocs.pm/ewe/ewe.html#ResponseBody). You can also use [`ewe.Request`](https://hexdocs.pm/ewe/ewe.html#Request)/[`ewe.Response`](https://hexdocs.pm/ewe/ewe.html#Response) as they are aliases for `request.Request(Connection)`(see [`request.Request`](https://hexdocs.pm/gleam_http/gleam/http/request.html#Request) & [`ewe.Connection`](https://hexdocs.pm/ewe/ewe.html#Connection))/`response.Response(ResponseBody)`. +```gleam +|> ewe.with_tls(ewe.Disk("priv/localhost.crt", "priv/localhost.key")) +|> ewe.with_client_verification(ewe.CaCertFile("priv/ca.crt")) +``` + +### HTTP/2 + +HTTP/2 is always enabled on ewe. Over TLS ewe offers it through ALPN and a plain +connection is served as HTTP/2 when it opens with the HTTP/2 preface which is +what a client with prior knowledge sends. An `Upgrade: h2c` request is not +negotiated, it is answered as HTTP/1.1. +> [!NOTE] +> Extended CONNECT is not negotiated yet, so WebSockets over HTTP/2 are not +> supported. + +### [Sending a Response](examples/src/sending_response.gleam) + +A response body is one of the [`ewe.Body`](https://hexdocs.pm/ewe/ewe.html#Body) +variants. `Text` and `Bytes` are in-memory bodies, `Empty` is for responses that +carry nothing and the rest are built by the functions covered further down: +`File` by [`ewe.file`](https://hexdocs.pm/ewe/ewe.html#file), `Streaming` by +[`ewe.stream_response`](https://hexdocs.pm/ewe/ewe.html#stream_response), `Sse` +by [`ewe.sse`](https://hexdocs.pm/ewe/ewe.html#sse) and `Websocket` by +[`ewe.websocket`](https://hexdocs.pm/ewe/ewe.html#websocket). ```gleam +import ewe +import gleam/bytes_tree import gleam/crypto -import gleam/http/request.{type Request} -import gleam/http/response.{type Response} +import gleam/http/request +import gleam/http/response import gleam/int import gleam/result -import ewe.{type Connection, type ResponseBody} - -fn handler(req: Request(Connection)) -> Response(ResponseBody) { - case request.path_segments(req) { +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + case request.path_segments(request) { ["hello", name] -> { - // Use TextData for text responses. - // + // Text for text responses. response.new(200) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) + |> response.set_body(ewe.Text("Hello, " <> name <> "!")) } ["bytes", amount] -> { - // Use BitsData for binary responses. - // - let random_bytes = + // Bytes for binary responses built from a `BytesTree`. + let body = int.parse(amount) |> result.unwrap(0) - |> crypto.strong_random_bytes() + |> crypto.strong_random_bytes + |> bytes_tree.from_bit_array + |> ewe.Bytes response.new(200) |> response.set_header("content-type", "application/octet-stream") - |> response.set_body(ewe.BitsData(random_bytes)) + |> response.set_body(body) } - _ -> - // Use Empty for responses with no body (like 404, 204, etc). - // + _segments -> + // Empty for responses with no body like 404 or 204. response.new(404) |> response.set_body(ewe.Empty) } } ``` -### [Reading Body](examples/src/reading_body.gleam) +### [Reading the Request Body](examples/src/reading_body.gleam) -To read the body of a request, use [`ewe.read_body`](https://hexdocs.pm/ewe/ewe.html#read_body). This function is intended for cases where the entire body can safely be loaded into memory. +[`ewe.read_body`](https://hexdocs.pm/ewe/ewe.html#read_body) reads the whole body +into memory up to `limit` bytes. Trailer fields of a chunked request are appended +to the returned request's headers. ```gleam -import gleam/http/request -import gleam/http/response -import gleam/result - -import ewe.{type Request, type Response} - -fn handler(req: Request) -> Response { +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { let content_type = - request.get_header(req, "content-type") + request.get_header(request, "content-type") |> result.unwrap("application/octet-stream") - // Read the entire request body into memory with a 10KB limit. This blocks - // until the full body is received. - // - case ewe.read_body(req, 10_240) { + case ewe.read_body(request, limit: 10_240) { Ok(req) -> response.new(200) |> response.set_header("content-type", content_type) - |> response.set_body(ewe.BitsData(req.body)) + |> response.set_body(ewe.Bytes(bytes_tree.from_bit_array(req.body))) Error(ewe.BodyTooLarge) -> response.new(413) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Body too large")) + |> response.set_body(ewe.Text("Body too large")) Error(ewe.InvalidBody) -> response.new(400) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Invalid request")) + |> response.set_body(ewe.Text("Invalid request")) } } ``` -### [Streaming Body](examples/src/streaming_body.gleam) +A body the handler never read is drained by the server so the connection can be +reused. One larger than `auto_drain_limit` closes the connection instead. -For larger request bodies, [`ewe.stream_body`](https://hexdocs.pm/ewe/ewe.html#stream_body) provides a streaming interface. It produces a [`ewe.Consumer`](https://hexdocs.pm/ewe/ewe.html#Consumer) which can be called repeatedly to read fixed-size chunks. This enables efficient handling of large payloads without buffering them fully. +### [Streaming Bodies](examples/src/streaming_bodies.gleam) -As for responses, use [`ewe.chunked_body`](https://hexdocs.pm/ewe/ewe.html#chunked_body) to send a chunked response for streaming data to the client. The response body is managed through [`ewe.ChunkedBody`](https://hexdocs.pm/ewe/ewe.html#ChunkedBody) and chunks are sent by calling [`ewe.send_chunk`](https://hexdocs.pm/ewe/ewe.html#send_chunk). Handlers control the connection lifecycle with [`ewe.ChunkedNext`](https://hexdocs.pm/ewe/ewe.html#ChunkedNext). +[`ewe.read_body_chunk`](https://hexdocs.pm/ewe/ewe.html#read_body_chunk) pulls up +to `max_chunk_bytes` per call rather than buffering everything. Each +[`ewe.Chunk`](https://hexdocs.pm/ewe/ewe.html#ReadEvent) carries the request to +feed into the next call. +Going the other way, [`ewe.stream_response`](https://hexdocs.pm/ewe/ewe.html#stream_response) +turns a response into a streamed one. Its handler owns an +[`ewe.ResponseWriter`](https://hexdocs.pm/ewe/ewe.html#ResponseWriter) and must +end by calling [`ewe.finish_chunk`](https://hexdocs.pm/ewe/ewe.html#finish_chunk) +or [`ewe.finish_response`](https://hexdocs.pm/ewe/ewe.html#finish_response) since +that is what closes the stream. The callback runs in the same connection process. ```gleam -pub type Message { - Chunk(BitArray) - Done - BodyError(ewe.BodyError) -} - -// Recursively consume chunks from the request body and send them to the -// chunked response handler via the subject. -// -fn stream_resource( - consumer: ewe.Consumer, - subject: Subject(Message), - chunk_size: Int, -) -> Nil { - process.sleep(int.random(250)) - // Call the consumer with the chunk size. It returns the next chunk of data - // and a new consumer for the remaining body. - // - case consumer(chunk_size) { - Ok(ewe.Consumed(data, next)) -> { - logging.log(logging.Info, { - "Consumed " <> int.to_string(bit_array.byte_size(data)) <> " bytes." - }) - - process.send(subject, Chunk(data)) - // Recursively process the next chunk. - // - stream_resource(next, subject, chunk_size) - } - Ok(ewe.Done) -> process.send(subject, Done) - Error(body_error) -> process.send(subject, BodyError(body_error)) - } -} - -fn handle_stream(req: Request, chunk_size: Int) -> Response { +fn handle_stream( + req: request.Request(ewe.Connection), + max_chunk_bytes: Int, +) -> response.Response(ewe.Body) { let content_type = request.get_header(req, "content-type") |> result.unwrap("application/octet-stream") - // Get a consumer function for streaming the request body. - // - case ewe.stream_body(req) { - Ok(consumer) -> { - // Set up a chunked response. The response is sent in chunks as we - // consume the request body. - // - ewe.chunked_body( - req, - response.new(200) |> response.set_header("content-type", content_type), - // Spawn a separate process to consume the body and send chunks. - // This prevents blocking the handler while reading data. - // - on_init: fn(subject) { - let _pid = - fn() { stream_resource(consumer, subject, chunk_size) } - |> process.spawn - }, - handler: fn(chunked_body, state, message) { - case message { - Chunk(data) -> - case ewe.send_chunk(chunked_body, data) { - Ok(Nil) -> ewe.chunked_continue(state) - Error(_) -> ewe.chunked_stop_abnormal("Failed to send chunk") - } - Done -> ewe.chunked_stop() - BodyError(_body_error) -> - ewe.chunked_stop_abnormal("failed to read body") - } - }, - on_close: fn(_conn, _state) { - logging.log(logging.Info, "Stream closed") - }, - ) + response.new(200) + |> response.set_header("content-type", content_type) + |> ewe.stream_response(echo_body(req, _, max_chunk_bytes)) +} + +// Read the request body one chunk at a time and write each one back out. +// +fn echo_body( + req: request.Request(ewe.Connection), + writer: ewe.ResponseWriter, + max_chunk_bytes: Int, +) -> Result(Nil, ewe.SendError) { + case ewe.read_body_chunk(req, max_chunk_bytes:, limit: 10_485_760) { + Ok(ewe.Chunk(data:, request:)) -> { + use writer <- result.try(ewe.send_chunk(writer, data)) + echo_body(request, writer, max_chunk_bytes) } - Error(_) -> - response.new(400) - |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Invalid request")) + Ok(ewe.Done(_request)) -> ewe.finish_response(writer) + Error(_body_error) -> ewe.finish_response(writer) } } ``` ### [Serving Files](examples/src/serving_files.gleam) -Static files can be sent using [`ewe.file`](https://hexdocs.pm/ewe/ewe.html#file). It accepts a path and optional `offset`/`limit` parameters. This allows serving HTML pages, assets, or binary files with minimal effort. +[`ewe.file`](https://hexdocs.pm/ewe/ewe.html#file) prepares a file as a response +body, streamed from disk rather than read into memory. `offset` and `limit` serve +a byte range, which is what a range request needs. It takes the connection, so it +is the request's body you pass in first. ```gleam -import gleam/bool -import gleam/http/response -import gleam/list -import gleam/string +case ewe.file(request.body, resolved, offset: None, limit: None) { + Ok(file) -> + response.new(200) + |> response.set_header("content-type", "application/octet-stream") + |> response.set_body(file) + Error(_error) -> not_found() +} +``` -fn serve_file(path: String) -> Response { - // Resolve the URL path against the `public` directory and confirm the result - // stays inside it. - // - let dir = absname("public") - let relative = string.drop_start(path, 1) - let segments = string.split(relative, "/") +On HTTP/1 this opens the file and the body holds it open until the response is +written so put it on a response you go on to return. A body that is built and +then discarded keeps its file open until it is collected. On HTTP/2 a file at or +below `file_read_threshold` is read into memory and framed like any other body. - use <- bool.guard( - when: list.any(segments, fn(seg) { seg == ".." }), - return: not_found(), - ) +### [Client Address](examples/src/client_info.gleam) - let resolved = absname_join(dir, relative) - - case string.starts_with(resolved, dir <> "/") { - True -> { - // Load file from disk using ewe.file(). This efficiently streams the file - // content without loading it entirely into memory. - // - case ewe.file(resolved, offset: None, limit: None) { - Ok(file) -> { - // Using "application/octet-stream" is safe for any file type, but you - // may want to specify content-type based on file extension in - // production. - // - response.new(200) - |> response.set_header("content-type", "application/octet-stream") - |> response.set_body(file) - } - Error(_) -> not_found() +[`ewe.get_client_info`](https://hexdocs.pm/ewe/ewe.html#get_client_info) reads the +address a request came from off its connection as a +[`ewe.SocketAddress`](https://hexdocs.pm/ewe/ewe.html#SocketAddress). It fails +only when the socket is already gone. + +```gleam +fn describe_client(connection: ewe.Connection) -> String { + case ewe.get_client_info(connection) { + Ok(ewe.TcpSocketAddress(ip_address:, port:)) -> { + // An IPv6 address is bracketed so the port stays readable next to the + // colons the address itself is full of. + let host = case ip_address { + ewe.IpV6(..) -> "[" <> ewe.ip_address_to_string(ip_address) <> "]" + ewe.IpV4(..) -> ewe.ip_address_to_string(ip_address) } + + host <> ":" <> int.to_string(port) } - False -> not_found() + Ok(ewe.UnixSocketAddress(path: "")) -> "unix socket" + Ok(ewe.UnixSocketAddress(path:)) -> "unix:" <> path + Error(Nil) -> "unknown" } } - -@external(erlang, "filename", "absname") -fn absname(path: String) -> String - -@external(erlang, "filename", "absname_join") -fn absname_join(dir: String, file: String) -> String ``` +Behind a proxy this is the proxy's address rather than the browser's. The one the +proxy puts in `x-forwarded-for` is the address to use there but only when the +proxy is yours, since any client can send that header itself. + ### [WebSocket](examples/src/websocket.gleam) -Use [`ewe.upgrade_websocket`](https://hexdocs.pm/ewe/ewe.html#upgrade_websocket) to switch an HTTP request into a WebSocket connection. Incoming messages are represented as [`ewe.WebsocketMessage`](https://hexdocs.pm/ewe/ewe.html#WebsocketMessage). Outgoing frames are sent with [`ewe.send_text_frame`](https://hexdocs.pm/ewe/ewe.html#send_text_frame) or [`ewe.send_binary_frame`](https://hexdocs.pm/ewe/ewe.html#send_binary_frame). Handlers control the connection lifecycle with [`ewe.WebsocketNext`](https://hexdocs.pm/ewe/ewe.html#WebsocketNext). +[`ewe.websocket`](https://hexdocs.pm/ewe/ewe.html#websocket) turns a request into +a WebSocket. A request that is not a valid handshake is answered with a 400 and +your handler never runs. Frames from the client and messages from the rest of +your program arrive as [`ewe.WebsocketMessage`](https://hexdocs.pm/ewe/ewe.html#WebsocketMessage) +values. Answer them with [`ewe.send_text_frame`](https://hexdocs.pm/ewe/ewe.html#send_text_frame) +or [`ewe.send_binary_frame`](https://hexdocs.pm/ewe/ewe.html#send_binary_frame) +and say what happens next with +[`ewe.WebsocketNext`](https://hexdocs.pm/ewe/ewe.html#WebsocketNext). ```gleam -import gleam/erlang/charlist.{type Charlist} -import gleam/erlang/process.{type Pid, type Subject} -import gleam/http/request -import gleam/http/response -import logging - -import ewe.{type Request, type Response} - -type PubSubMessage { - Subscribe(topic: String, client: Subject(Broadcast)) - Publish(topic: String, message: Broadcast) - Unsubscribe(topic: String, client: Subject(Broadcast)) -} - -type Broadcast { - Text(String) - Bytes(BitArray) -} - -type WebsocketState { - WebsocketState( - pubsub: Subject(PubSubMessage), - topic: String, - client: Subject(Broadcast), - ) -} - -fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { - case request.path_segments(req) { - ["topic", topic] -> handle_topic(req, pubsub, topic) - _ -> - response.new(404) - |> response.set_body(ewe.Empty) - } -} - -fn handle_topic(req: Request, pubsub: Subject(PubSubMessage), topic: String) { - // Upgrade the HTTP connection to WebSocket. Unlike SSE, WebSocket is - // bidirectional - both client and server can send messages at any time. - // - ewe.upgrade_websocket( - req, - // Initialize the WebSocket connection. The selector allows receiving - // messages from both the WebSocket and the pubsub system. - // +fn handle_topic( + req: request.Request(ewe.Connection), + pubsub: Subject(pubsub.Message(Broadcast)), + topic: String, +) -> response.Response(ewe.Body) { + ewe.websocket( + request: req, + // Called once. The selector is where you add whatever the rest of your + // program sends this connection. on_init: fn(_conn, selector) { let client = process.new_subject() - process.send(pubsub, Subscribe(topic:, client:)) + pubsub.subscribe(pubsub, topic:, client:) let state = WebsocketState(pubsub:, topic:, client:) - // Add the client subject to the selector to receive broadcast messages. - // let selector = process.select(selector, client) #(state, selector) }, handler: handle_websocket_message, + // Called once however the WebSocket ended. on_close: fn(_conn, state) { - process.send(pubsub, Unsubscribe(state.topic, state.client)) + pubsub.unsubscribe(state.pubsub, topic: state.topic, client: state.client) }, ) } -// Handle three types of messages: text from client, binary from client, -// and broadcast messages from the pubsub system. -// fn handle_websocket_message( conn: ewe.WebsocketConnection, state: WebsocketState, - msg: ewe.WebsocketMessage(Broadcast), + message: ewe.WebsocketMessage(Broadcast), ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { - case msg { - // Text message from the client - broadcast to all subscribers. - // - ewe.Text(text) -> { - process.send(state.pubsub, Publish(state.topic, Text(text))) + case message { + ewe.TextFrame(text) -> { + pubsub.publish(state.pubsub, topic: state.topic, message: Text(text)) ewe.websocket_continue(state) } - // Binary message from the client - broadcast to all subscribers. - // - ewe.Binary(binary) -> { - process.send(state.pubsub, Publish(state.topic, Bytes(binary))) + ewe.BinaryFrame(data) -> { + pubsub.publish(state.pubsub, topic: state.topic, message: Bytes(data)) ewe.websocket_continue(state) } - // User message from the pubsub - forward to this client. - // - ewe.User(message) -> { - let assert Ok(_) = case message { + // A message from the rest of the program. + ewe.UserMessage(broadcast) -> { + let sent = case broadcast { Text(text) -> ewe.send_text_frame(conn, text) - Bytes(binary) -> ewe.send_binary_frame(conn, binary) + Bytes(data) -> ewe.send_binary_frame(conn, data) } - ewe.websocket_continue(state) + case sent { + Ok(Nil) -> ewe.websocket_continue(state) + Error(_send_error) -> + ewe.websocket_stop_abnormal("Failed to send a frame") + } } } } ``` +Ping and pong frames are answered by the server and never reach the handler. To +start the closing handshake yourself, return +[`ewe.send_close_frame`](https://hexdocs.pm/ewe/ewe.html#send_close_frame) with a +[`ewe.CloseReason`](https://hexdocs.pm/ewe/ewe.html#CloseReason). No frame can be +sent after it! + ### [Server-Sent Events](examples/src/sse.gleam) +[`ewe.sse`](https://hexdocs.pm/ewe/ewe.html#sse) turns a response into an SSE +stream which runs until the handler stops it or the client goes away. `on_init` +receives the subject the rest of your program pushes messages to, `handler` is +called for each of those messages and `on_close` runs however the stream ended. +The `content-type` and `cache-control` headers the stream needs are set by ewe. + +```gleam +response.new(200) +|> ewe.sse( + on_init: fn(client) { + pubsub.subscribe(pubsub, topic:, client:) + + client + }, + handler: fn(conn, client, message) { + case ewe.send_event(conn, ewe.event(message)) { + Ok(Nil) -> ewe.sse_continue(client) + Error(_send_error) -> ewe.sse_stop() + } + }, + on_close: fn(_conn, client) { + pubsub.unsubscribe(pubsub, topic:, client:) + }, +) +``` -Use [`ewe.sse`](https://hexdocs.pm/ewe/ewe.html#sse) to establish a Server-Sent Events connection for real-time data streaming to clients. The connection is managed through [`ewe.SSEConnection`](https://hexdocs.pm/ewe/ewe.html#SSEConnection) and events are sent with [`ewe.send_event`](https://hexdocs.pm/ewe/ewe.html#send_event). Handlers control the connection lifecycle with [`ewe.SSENext`](https://hexdocs.pm/ewe/ewe.html#SSENext). This enables efficient one-way communication for live updates, notifications, or real-time data feeds. +An event is built with [`ewe.event`](https://hexdocs.pm/ewe/ewe.html#event) and +can carry a name, an id and a reconnection delay through +[`ewe.event_name`](https://hexdocs.pm/ewe/ewe.html#event_name), +[`ewe.event_id`](https://hexdocs.pm/ewe/ewe.html#event_id) and +[`ewe.event_retry`](https://hexdocs.pm/ewe/ewe.html#event_retry). +[`ewe.comment`](https://hexdocs.pm/ewe/ewe.html#comment) sends something clients +ignore which is the usual way to keep an idle stream from being closed by a +proxy. + +### Connection Limits and Timeouts + +Every connection is held to a set of limits and timeouts. Start from +[`ewe.default_http1_options`](https://hexdocs.pm/ewe/ewe.html#default_http1_options) +or [`ewe.default_http2_options`](https://hexdocs.pm/ewe/ewe.html#default_http2_options), +update the fields you care about and hand the result to +[`ewe.with_http1`](https://hexdocs.pm/ewe/ewe.html#with_http1) or +[`ewe.with_http2`](https://hexdocs.pm/ewe/ewe.html#with_http2). Sizes are in bytes +and timeouts in milliseconds. ```gleam -import gleam/bit_array -import gleam/erlang/process.{type Subject} -import gleam/http -import gleam/http/response +let http1 = + ewe.Http1Options( + ..ewe.default_http1_options(), + // Refuse a request carrying more than 50 header fields with a 431. + max_headers: 50, + // Close a connection that sits idle for 30 seconds. + idle_timeout: 30_000, + ) -import ewe +let http2 = + ewe.Http2Options( + ..ewe.default_http2_options(), + // Cap how many streams a client may have open at once. + max_concurrent_streams: Some(100), + // Trip a GOAWAY sooner on a client resetting streams in bulk. + rapid_reset_threshold: 50, + ) -type PubSubMessage { - Subscribe(client: Subject(String)) - Unsubscribe(client: Subject(String)) - Publish(String) -} +ewe.new(listener_name:, connection_factory_name:, handler: handle_request) +|> ewe.with_http1(http1) +|> ewe.with_http2(http2) +|> ewe.start +``` -fn handler(req: ewe.Request, pubsub: Subject(PubSubMessage)) -> ewe.Response { - case req.method, req.path { - http.Get, "/sse" -> - // Establish a Server-Sent Events connection. SSE is a one-way channel - // from server to client. The connection stays open and the server can - // push events at any time. - // - ewe.sse( - req, - // Initialize the connection and subscribe this client to the pubsub. - // - on_init: fn(client) { - process.send(pubsub, Subscribe(client)) - - client - }, - // Handle messages from the pubsub and send them as SSE events. - // - handler: fn(conn, client, message) { - case ewe.send_event(conn, ewe.event(message)) { - Ok(Nil) -> ewe.sse_continue(client) - Error(_) -> ewe.sse_stop() - } - }, - // Clean up when the client disconnects. - // - on_close: fn(_conn, client) { - process.send(pubsub, Unsubscribe(client)) - }, - ) - - // Accept messages via POST and broadcast them to all SSE clients. - // - http.Post, "/post" -> { - case ewe.read_body(req, 128) { - Ok(req) -> { - case bit_array.to_string(req.body) { - Ok(message) -> { - process.send(pubsub, Publish(message)) - - response.new(200) |> response.set_body(ewe.Empty) - } - Error(Nil) -> response.new(400) |> response.set_body(ewe.Empty) - } - } - Error(_) -> response.new(400) |> response.set_body(ewe.Empty) - } - } +[`ewe.Http1Options`](https://hexdocs.pm/ewe/ewe.html#Http1Options) covers the +request line, header line and header count caps, the chunk size line cap, the +idle and body read timeouts and how much of an unread body is drained so the +connection can be reused: + +| Field | Default | What it does | +| --- | --- | --- | +| `max_request_line` | `8192` | Longer request lines are refused with a 414. | +| `max_header_line` | `8192` | Longer header lines are refused with a 431. | +| `max_headers` | `100` | Requests carrying more header fields are refused with a 431. | +| `max_chunk_size_line` | `128` | Longest chunk size line in a chunked body. | +| `idle_timeout` | `10_000` | How long a connection may sit without sending anything. | +| `body_read_timeout` | `10_000` | How long a single body read waits for the client. | +| `auto_drain_limit` | `1_048_576` | An unread body larger than this closes the connection instead of being drained. | +| `auto_drain_chunk_bytes` | `65_536` | How much of that drain is read at a time. | + +[`ewe.Http2Options`](https://hexdocs.pm/ewe/ewe.html#Http2Options) covers the same +ground plus what the protocol adds. A value the protocol does not allow is +replaced with the default rather than reaching a peer: + +| Field | Default | What it does | +| --- | --- | --- | +| `max_concurrent_streams` | `None` | How many streams a client may have open at once. | +| `initial_window_size` | `2_097_152` | How much response body a stream may have in flight. | +| `max_frame_size` | `16_384` | Largest frame accepted, between 16384 and 16777215. | +| `max_header_list_size` | `Some(32_768)` | Largest header list accepted. | +| `header_table_size` | `4096` | HPACK dynamic table kept for decoding. | +| `max_continuation_frames` | `100` | How many CONTINUATION frames one header sequence may span. | +| `max_header_block_bytes` | `65_536` | Bytes one header block may total before decoding. | +| `rapid_reset_window` | `10_000` | Window over which client stream resets are counted. | +| `rapid_reset_threshold` | `100` | Resets within that window that trip a GOAWAY which is what keeps Rapid Reset (CVE-2023-44487) in check. | +| `handshake_timeout` | `10_000` | How long a connection may sit in the preface and SETTINGS handshake. | +| `drain_timeout` | `4000` | How long a draining connection waits for its streams after GOAWAY. | +| `recv_window_low_water_mark` | `262_144` | Once a receive window falls to this it is topped back up. | +| `recv_window_high_water_mark` | `2_097_152` | What it is topped up to; a wider gap costs fewer WINDOW_UPDATE round trips. | +| `file_read_threshold` | `1_048_576` | Files at or below this are read into memory, larger ones are streamed from disk. | +| `body_read_timeout` | `10_000` | How long a single body read waits for the client. | + +### Running Under Supervision + +[`ewe.start`](https://hexdocs.pm/ewe/ewe.html#start) runs the server on its own. +When it belongs to a supervision tree next to the rest of your program use +[`ewe.supervised`](https://hexdocs.pm/ewe/ewe.html#supervised) instead, which +returns a child specification. - _, _ -> response.new(404) |> response.set_body(ewe.Empty) - } -} +```gleam +supervisor.new(supervisor.OneForAll) +|> supervisor.add(pubsub.worker(pubsub_name)) +|> supervisor.add( + ewe.new(listener_name:, connection_factory_name:, handler:) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) + |> ewe.supervised, +) +|> supervisor.start ``` +The line printed on startup comes from [`ewe.on_start`](https://hexdocs.pm/ewe/ewe.html#on_start), +which receives the scheme and the address the server bound to. Replace it to log +it your own way or silence it with [`ewe.quiet`](https://hexdocs.pm/ewe/ewe.html#quiet). + +## Examples + +Most sections above link to a runnable example. They live in +[examples](examples/), see [its README](examples/README.md) for how to run them. + ## API Reference For detailed API documentation, see [hexdocs.pm/ewe](https://hexdocs.pm/ewe/ewe.html). diff --git a/benchmark/results.txt b/benchmark/results.txt index 2e79bba..a66b3f2 100644 --- a/benchmark/results.txt +++ b/benchmark/results.txt @@ -1,5 +1,3 @@ -results/20260812-115143-throughput/throughput.csv - profiles h1 h1, 50 connections x 1 stream(s) h2 h2, 50 connections x 10 stream(s) @@ -94,8 +92,6 @@ repeats disagreed by over 5% roadrunner h2 stream_big: 5% apart across repeats chatterbox h2 file_big: 64% apart across repeats -results/20260812-151117-latency/latency.csv - p99 at a fixed offered rate for http1 hello diff --git a/examples/README.md b/examples/README.md index 158bf7f..79ab0e2 100644 --- a/examples/README.md +++ b/examples/README.md @@ -1,4 +1,4 @@ -# Ewe Examples +# Examples This directory contains practical examples demonstrating various features of ewe. @@ -19,7 +19,12 @@ Here is a list of all the examples: - [getting_started](./src/getting_started.gleam) - Basic HTTP server with "Hello, World!" response - [sending_response](./src/sending_response.gleam) - Different response body types (text, binary, empty) - [reading_body](./src/reading_body.gleam) - Reading and echoing request bodies with size limits -- [streaming_body](./src/streaming_body.gleam) - Streaming large request/response bodies in chunks +- [streaming_bodies](./src/streaming_bodies.gleam) - Streaming large request/response bodies in chunks - [serving_files](./src/serving_files.gleam) - Serving static files from disk +- [client_info](./src/client_info.gleam) - Reading the address a request came from - [websocket](./src/websocket.gleam) - WebSocket connections with topic-based pubsub - [sse](./src/sse.gleam) - Server-Sent Events for real-time server-to-client updates + +The `websocket` and `sse` examples share [pubsub](./src/examples/pubsub.gleam), a small +topic based broadcaster. It is not an example itself, it is kept out of the +handlers so they stay about ewe's API. diff --git a/examples/manifest.toml b/examples/manifest.toml index e195c62..459e557 100644 --- a/examples/manifest.toml +++ b/examples/manifest.toml @@ -7,16 +7,17 @@ # You should check this file into your source control repository. packages = [ - { name = "ewe", version = "5.0.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_http", "gleam_otp", "gleam_stdlib", "glisten", "logging", "websocks"], source = "local", path = ".." }, + { name = "alpacki", version = "3.0.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "alpacki", source = "hex", outer_checksum = "5BCB617A9606E56018790D4F24AB1E06B6D05BC17ACD6686B99E8D4D18B1F5FD" }, + { name = "ewe", version = "5.0.0", build_tools = ["gleam"], requirements = ["alpacki", "gleam_erlang", "gleam_http", "gleam_otp", "gleam_stdlib", "glisten", "logging", "websocks"], source = "local", path = ".." }, { name = "gleam_crypto", version = "1.6.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "2DE9E4EF53CF6FEE049D4F765731F7178F7A11AEFAE00EEE63BF7536B354AD3F" }, { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, { name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" }, - { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" }, - { name = "gleam_stdlib", version = "1.0.2", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "0C5506589DF4C63DF5D6FFBB834562D6865C6C2AEE0019D7B37886BD6D128141" }, - { name = "gleeunit", version = "1.10.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "254B697FE72EEAD7BF82E941723918E421317813AC49923EE76A18C788C61E72" }, - { name = "glisten", version = "9.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging"], source = "git", repo = "https://github.com/vshakitskiy/glisten.git", commit = "63e9a39f7dc35526c2f62128f44f871dab245d15" }, + { name = "gleam_otp", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "DE4CA6850842F0266EE95317A25DD6A0A0F20CDFAB7C0ADC2E63251D7C3C72EC" }, + { name = "gleam_stdlib", version = "1.0.5", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "CEE5B6C076A85B45F60C585F4316C63EC8B7127C119D5738C3958A9C4D50404E" }, + { name = "gleeunit", version = "1.11.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "EC31ABA74256AEA531EDF8169931D775BBB384FED0A8A1BDC4DD9354E3E21826" }, + { name = "glisten", version = "9.0.1", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging"], source = "git", repo = "https://github.com/vshakitskiy/glisten.git", commit = "81a6b0005451b61a14dad2f5892d04cbaad3bc5d" }, { name = "logging", version = "1.5.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "BC5F18CE5DD9686100229FE5409BDC3DD5C46D5A7DF2F804AD2D8F0DD6C5060E" }, - { name = "websocks", version = "3.0.1", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "C70340E5B6C3390383ADA17029DCA6F8903863A7AD8CD8E1520EDCC4FE70D6FD" }, + { name = "websocks", version = "4.0.1", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "89B0C31A032CBE28D4C5FB5CC25A8A690669FEBA501C63F99A4E8F5748750B2A" }, ] [requirements] diff --git a/examples/src/client_info.gleam b/examples/src/client_info.gleam new file mode 100644 index 0000000..87b9872 --- /dev/null +++ b/examples/src/client_info.gleam @@ -0,0 +1,72 @@ +import ewe +import gleam/erlang/process +import gleam/http +import gleam/http/request +import gleam/http/response +import gleam/int +import logging + +pub fn main() { + logging.configure() + logging.set_level(logging.Info) + + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // A server that logs who every request came from and tells the client its own + // address, the way `curl ifconfig.me` does. + // + let assert Ok(_) = + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) + |> ewe.start + + process.sleep_forever() +} + +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + // The connection the request carries is what the client's address is read + // from, so it is `request.body` that goes in here. + let client = describe_client(request.body) + + logging.log( + logging.Info, + http.method_to_string(request.method) + <> " " + <> request.path + <> " " + <> client, + ) + + response.new(200) + |> response.set_header("content-type", "text/plain; charset=utf-8") + |> response.set_body(ewe.Text(client <> "\n")) +} + +// Behind a proxy this is the proxy's address and not the browser's. The address +// the proxy puts in `x-forwarded-for` is the one to use there but only when +// the proxy is yours, since any client can send that header itself. +// +fn describe_client(connection: ewe.Connection) -> String { + case ewe.get_client_info(connection) { + Ok(ewe.TcpSocketAddress(ip_address:, port:)) -> { + // An IPv6 address is bracketed so the port stays readable next to the + // colons the address itself is full of. + let host = case ip_address { + ewe.IpV6(..) -> "[" <> ewe.ip_address_to_string(ip_address) <> "]" + ewe.IpV4(..) -> ewe.ip_address_to_string(ip_address) + } + + host <> ":" <> int.to_string(port) + } + // A unix socket client is unnamed unless it bound a path of its own, which + // clients rarely do, so most of the time there is no path to report. + Ok(ewe.UnixSocketAddress(path: "")) -> "unix socket" + Ok(ewe.UnixSocketAddress(path:)) -> "unix:" <> path + // The socket is already gone, so there is nothing left to report. + Error(Nil) -> "unknown" + } +} diff --git a/examples/src/examples/pubsub.gleam b/examples/src/examples/pubsub.gleam new file mode 100644 index 0000000..9564ff9 --- /dev/null +++ b/examples/src/examples/pubsub.gleam @@ -0,0 +1,135 @@ +//// A small topic based pubsub shared by the examples that need one. Clients +//// subscribe with a subject of their own message type and every message +//// published to a topic is sent to each of them. + +import gleam/dict.{type Dict} +import gleam/erlang/charlist.{type Charlist} +import gleam/erlang/process.{type Name, type Pid, type Subject} +import gleam/int +import gleam/list +import gleam/option.{None, Some} +import gleam/otp/actor +import gleam/otp/supervision.{type ChildSpecification} +import logging + +pub type Message(message) { + Subscribe(topic: String, client: Subject(message)) + Unsubscribe(topic: String, client: Subject(message)) + Publish(topic: String, message: message) +} + +/// Returns the pubsub worker to add to a supervisor. Pass the same name to +/// `process.named_subject` to talk to it. +pub fn worker( + named: Name(Message(message)), +) -> ChildSpecification(Subject(Message(message))) { + supervision.worker(fn() { + logging.log(logging.Info, "Starting pubsub worker") + + dict.new() + |> actor.new + |> actor.on_message(handle_message) + |> actor.named(named) + |> actor.start + }) +} + +pub fn subscribe( + pubsub: Subject(Message(message)), + topic topic: String, + client client: Subject(message), +) -> Nil { + process.send(pubsub, Subscribe(topic:, client:)) +} + +pub fn unsubscribe( + pubsub: Subject(Message(message)), + topic topic: String, + client client: Subject(message), +) -> Nil { + process.send(pubsub, Unsubscribe(topic:, client:)) +} + +pub fn publish( + pubsub: Subject(Message(message)), + topic topic: String, + message message: message, +) -> Nil { + process.send(pubsub, Publish(topic:, message:)) +} + +fn handle_message( + topics: Dict(String, List(Subject(message))), + message: Message(message), +) -> actor.Next(Dict(String, List(Subject(message))), Message(message)) { + case message { + Subscribe(topic:, client:) -> { + let topics = + dict.upsert(in: topics, update: topic, with: fn(clients) { + case clients { + Some(clients) -> [client, ..clients] + None -> { + logging.log(logging.Info, "Creating topic " <> topic) + [client] + } + } + }) + + log_client("Subscribing client ", client, " to topic " <> topic) + + actor.continue(topics) + } + + Unsubscribe(topic:, client:) -> { + log_client("Unsubscribing client ", client, " from topic " <> topic) + + let topics = case dict.get(topics, topic) { + Ok([_client]) | Ok([]) -> { + logging.log(logging.Info, "Dropping topic " <> topic) + dict.drop(topics, [topic]) + } + Ok(clients) -> + list.filter(clients, fn(subscribed) { subscribed != client }) + |> dict.insert(topics, topic, _) + Error(Nil) -> topics + } + + actor.continue(topics) + } + + Publish(topic:, message:) -> { + case dict.get(topics, topic) { + Ok(clients) -> { + list.each(clients, process.send(_, message)) + + { "Published to " <> topic <> ", " <> client_count(clients) } + |> logging.log(logging.Info, _) + } + Error(Nil) -> + logging.log(logging.Info, "Nobody is subscribed to " <> topic) + } + + actor.continue(topics) + } + } +} + +fn client_count(clients: List(Subject(message))) -> String { + case list.length(clients) { + 1 -> "1 client" + count -> int.to_string(count) <> " clients" + } +} + +fn log_client(before: String, client: Subject(message), after: String) -> Nil { + let assert Ok(pid) = process.subject_owner(client) + + logging.log(logging.Info, before <> pid_to_string(pid) <> after) +} + +fn pid_to_string(pid: Pid) -> String { + charlist.to_string(pid_to_list(pid)) +} + +@external(erlang, "erlang", "pid_to_list") +fn pid_to_list(pid: Pid) -> Charlist diff --git a/examples/src/getting_started.gleam b/examples/src/getting_started.gleam index 4a4e518..9c05704 100644 --- a/examples/src/getting_started.gleam +++ b/examples/src/getting_started.gleam @@ -1,5 +1,6 @@ -import ewe.{type Request, type Response} +import ewe import gleam/erlang/process +import gleam/http/request import gleam/http/response import logging @@ -10,27 +11,34 @@ pub fn main() { logging.configure() logging.set_level(logging.Info) - // Start the ewe web server. + // For the web server setup we need to create process names at the place where + // your program starts. These names are required for the acceptor pool working + // correctly. + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // Start the ewe web server binding to all interfaces. // let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.start - // Put process into sleep. + // Put the main process into sleep. // process.sleep_forever() } // This is the HTTP request handler. // -fn handler(_req: Request) -> Response { - // When sending response with body, it is important to include `content-type` +fn handle_request( + _request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + // When sending response with body it is important to include `content-type` // header representing what type your body is. You don't need to specify - // `content-length`, it is calculated automatically by ewe. - // + // `content-length` as it is calculated automatically by ewe. response.new(200) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Hello, World!")) + |> response.set_body(ewe.Text("Hello, World!")) } diff --git a/examples/src/https.gleam b/examples/src/https.gleam index 35f8e29..4ac5ca2 100644 --- a/examples/src/https.gleam +++ b/examples/src/https.gleam @@ -1,5 +1,6 @@ -import ewe.{type Request, type Response} +import ewe import gleam/erlang/process +import gleam/http/request import gleam/http/response import logging @@ -7,22 +8,25 @@ pub fn main() { logging.configure() logging.set_level(logging.Debug) - // Start the server that has TLS enabled. + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // Start the server that has TLS enabled with certificates on disk. You can + // also provide any in-memory certificates. let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) - |> ewe.enable_tls( - certificate_file: "priv/localhost.crt", - key_file: "priv/localhost.key", - ) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) + |> ewe.with_tls(ewe.Disk("priv/localhost.crt", "priv/localhost.key")) |> ewe.start process.sleep_forever() } -fn handler(_req: Request) -> Response { +fn handle_request( + _request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { response.new(200) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Hello, World!")) + |> response.set_body(ewe.Text("Hello, World!")) } diff --git a/examples/src/reading_body.gleam b/examples/src/reading_body.gleam index d480004..4944298 100644 --- a/examples/src/reading_body.gleam +++ b/examples/src/reading_body.gleam @@ -1,4 +1,5 @@ -import ewe.{type Request, type Response} +import ewe +import gleam/bytes_tree import gleam/erlang/process import gleam/http/request import gleam/http/response @@ -9,40 +10,43 @@ pub fn main() { logging.configure() logging.set_level(logging.Info) + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + // An echo server that reads the request body and sends it back. // let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.start process.sleep_forever() } -fn handler(req: Request) -> Response { +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { // Preserve the original content-type from the request to send back. - // let content_type = - request.get_header(req, "content-type") + request.get_header(request, "content-type") |> result.unwrap("application/octet-stream") // Read the entire request body into memory with a 10KB limit. This blocks - // until the full body is received. For large uploads or streaming data, - // use ewe.stream_body() instead. - // - case ewe.read_body(req, 10_240) { + // until the full body is received. For large uploads or streaming data use + // ewe.read_body_chunk instead. + case ewe.read_body(request, limit: 10_240) { Ok(req) -> response.new(200) |> response.set_header("content-type", content_type) - |> response.set_body(ewe.BitsData(req.body)) + |> response.set_body(ewe.Bytes(bytes_tree.from_bit_array(req.body))) Error(ewe.BodyTooLarge) -> response.new(413) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Body too large")) + |> response.set_body(ewe.Text("Body too large")) Error(ewe.InvalidBody) -> response.new(400) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Invalid request")) + |> response.set_body(ewe.Text("Invalid request")) } } diff --git a/examples/src/sending_response.gleam b/examples/src/sending_response.gleam index 69bc95e..a344863 100644 --- a/examples/src/sending_response.gleam +++ b/examples/src/sending_response.gleam @@ -1,8 +1,9 @@ -import ewe.{type Connection, type ResponseBody} +import ewe +import gleam/bytes_tree import gleam/crypto import gleam/erlang/process -import gleam/http/request.{type Request} -import gleam/http/response.{type Response} +import gleam/http/request +import gleam/http/response import gleam/int import gleam/result import logging @@ -11,47 +12,49 @@ pub fn main() { logging.configure() logging.set_level(logging.Info) + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + // A server demonstrating different response body types and path routing. // let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.start process.sleep_forever() } -fn handler(req: Request(Connection)) -> Response(ResponseBody) { +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { // Pattern match on path segments for cleaner routing. // Example: "/hello/alice" becomes ["hello", "alice"] - // - case request.path_segments(req) { + case request.path_segments(request) { ["hello", name] -> { - // Here, we will use TextData for text responses. - // + // Here, we will use Text for text responses. response.new(200) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) + |> response.set_body(ewe.Text("Hello, " <> name <> "!")) } ["bytes", amount] -> { - // Use BitsData for binary responses. We generate random bytes - // to demonstrate sending binary data. - // + // Use Bytes for binary responses. We generate random bytes to + // demonstrate sending binary data. let body = int.parse(amount) |> result.unwrap(0) |> crypto.strong_random_bytes - |> ewe.BitsData + |> bytes_tree.from_bit_array + |> ewe.Bytes response.new(200) |> response.set_header("content-type", "application/octet-stream") |> response.set_body(body) } - _ -> + _segments -> // Use Empty for responses with no body (like 404, 204, etc). // You don't need to set content-type for empty bodies. - // response.new(404) |> response.set_body(ewe.Empty) } diff --git a/examples/src/serving_files.gleam b/examples/src/serving_files.gleam index a0f8d83..6dcfeb7 100644 --- a/examples/src/serving_files.gleam +++ b/examples/src/serving_files.gleam @@ -1,6 +1,7 @@ -import ewe.{type Response} +import ewe import gleam/bool import gleam/erlang/process +import gleam/http/request import gleam/http/response import gleam/list import gleam/option.{None} @@ -11,58 +12,59 @@ pub fn main() { logging.configure() logging.set_level(logging.Info) + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + // Start a simple file server that serves files from the "public" directory. // let assert Ok(_) = - ewe.new(fn(req) { serve_file(req.path) }) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.start process.sleep_forever() } -fn serve_file(path: String) -> Response { - // Resolve the URL path against the `public` directory and confirm the result +fn handle_request( + request: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + // Resolve the URL path against the `public` directory and confirm the result // stays inside it. - // let dir = absname("public") - let relative = string.drop_start(path, 1) + let relative = string.drop_start(request.path, 1) let segments = string.split(relative, "/") use <- bool.guard( - when: list.any(segments, fn(seg) { seg == ".." }), + when: list.any(segments, fn(segment) { segment == ".." }), return: not_found(), ) let resolved = absname_join(dir, relative) - case string.starts_with(resolved, dir <> "/") { True -> { - // Load file from disk using ewe.file(). This efficiently streams the file - // content without loading it entirely into memory. - // - case ewe.file(resolved, offset: None, limit: None) { + // Load the file from disk with ewe.file(). This function will provide the + // most optimized way of serving the file in ewe. + case ewe.file(request.body, resolved, offset: None, limit: None) { Ok(file) -> { - // Using "application/octet-stream" is safe for any file type, but you - // may want to specify content-type based on file extension in + // Using "application/octet-stream" is safe for any file type but you + // may want to specify content-type based on file extension in // production. - // response.new(200) |> response.set_header("content-type", "application/octet-stream") |> response.set_body(file) } - Error(_) -> not_found() + Error(_error) -> not_found() } } False -> not_found() } } -fn not_found() -> Response { +fn not_found() -> response.Response(ewe.Body) { response.new(404) |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("File not found")) + |> response.set_body(ewe.Text("File not found")) } @external(erlang, "filename", "absname") diff --git a/examples/src/sse.gleam b/examples/src/sse.gleam index 2f5d549..f6bfa92 100644 --- a/examples/src/sse.gleam +++ b/examples/src/sse.gleam @@ -1,19 +1,21 @@ import ewe +import examples/pubsub import gleam/bit_array -import gleam/erlang/charlist -import gleam/erlang/process.{type Name, type Pid, type Subject} +import gleam/erlang/process import gleam/http +import gleam/http/request import gleam/http/response -import gleam/list import gleam/option.{None} -import gleam/otp/actor import gleam/otp/static_supervisor as supervisor -import gleam/otp/supervision.{type ChildSpecification} -import gleam/string import logging +// Every SSE client here listens to the same topic, so a message posted to the +// server reaches all of them. +const topic = "messages" + pub fn main() -> Nil { logging.configure() + logging.set_level(logging.Info) // Create a named pubsub process for broadcasting messages to all connected // SSE clients. @@ -21,18 +23,24 @@ pub fn main() -> Nil { let pubsub_name = process.new_name("pubsub") let pubsub = process.named_subject(pubsub_name) + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // Remember, `handle_request(_, pubsub)` is the same as: + // fn(request) { handle_request(request, pubsub) } + let handler = handle_request(_, pubsub) + // Use a supervisor to manage both the pubsub worker and web server. // OneForAll means if either crashes, both will restart together. // let assert Ok(_) = supervisor.new(supervisor.OneForAll) - |> supervisor.add(pubsub_worker(pubsub_name)) + |> supervisor.add(pubsub.worker(pubsub_name)) |> supervisor.add( - ewe.new(handler(_, pubsub)) - |> ewe.listening(port: 8080) - |> ewe.bind("0.0.0.0") + ewe.new(listener_name:, connection_factory_name:, handler:) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) // Use ewe.supervised instead of ewe.start to run under supervision. - // |> ewe.supervised, ) |> supervisor.start @@ -40,149 +48,76 @@ pub fn main() -> Nil { process.sleep_forever() } -// SSE -// ----------------------------------------------------------------------------- - -fn handler(req: ewe.Request, pubsub: Subject(PubSubMessage)) -> ewe.Response { +fn handle_request( + req: request.Request(ewe.Connection), + pubsub: process.Subject(pubsub.Message(String)), +) -> response.Response(ewe.Body) { case req.method, req.path { // Serve the demo HTML page. - // http.Get, "/" -> { - case ewe.file("priv/index.html", offset: None, limit: None) { + case ewe.file(req.body, "priv/index.html", offset: None, limit: None) { Ok(file) -> { response.new(200) |> response.set_body(file) |> response.set_header("content-type", "text/html") } - Error(_) -> empty_response(500) + Error(_error) -> + response.new(500) + |> response.set_body(ewe.Empty) } } // Establish a Server-Sent Events connection. SSE is a one-way channel // from server to client. The connection stays open and the server can - // push events at any time. - // + // push events at any time. The `content-type` and `cache-control` headers + // the stream needs are set by ewe. http.Get, "/sse" -> - ewe.sse( - req, + response.new(200) + |> ewe.sse( // Initialize the connection and subscribe this client to the pubsub. - // on_init: fn(client) { - process.send(pubsub, Subscribe(client)) + pubsub.subscribe(pubsub, topic:, client:) client }, // Handle messages from the pubsub and send them as SSE events. - // handler: fn(conn, client, message) { case ewe.send_event(conn, ewe.event(message)) { Ok(Nil) -> ewe.sse_continue(client) - Error(_) -> ewe.sse_stop() + Error(_send_error) -> ewe.sse_stop() } }, // Clean up when the client disconnects. - // on_close: fn(_conn, client) { - process.send(pubsub, Unsubscribe(client)) + pubsub.unsubscribe(pubsub, topic:, client:) }, ) // Accept messages via POST and broadcast them to all SSE clients. - // http.Post, "/post" -> { - // Limit matches the frontend restriction (see index.html). - // - case ewe.read_body(req, 128) { + // Limit matches the frontend restriction. + case ewe.read_body(req, limit: 128) { Ok(req) -> { case bit_array.to_string(req.body) { Ok(message) -> { - process.send(pubsub, Publish(message)) + pubsub.publish(pubsub, topic:, message:) - empty_response(200) + response.new(200) + |> response.set_body(ewe.Empty) } - Error(Nil) -> empty_response(400) + Error(Nil) -> + response.new(400) + |> response.set_body(ewe.Empty) } } - Error(_) -> empty_response(400) + Error(_body_error) -> + response.new(400) + |> response.set_body(ewe.Empty) } } - _, _ -> empty_response(404) - } -} - -fn empty_response(status: Int) -> ewe.Response { - response.new(status) |> response.set_body(ewe.Empty) -} - -// PubSub -// ----------------------------------------------------------------------------- - -type PubSubMessage { - Subscribe(client: Subject(String)) - Unsubscribe(client: Subject(String)) - Publish(String) -} - -fn pubsub_worker( - named: Name(PubSubMessage), -) -> ChildSpecification(Subject(PubSubMessage)) { - supervision.worker(fn() { - actor.new([]) - |> actor.on_message(handle_pubsub_message) - |> actor.named(named) - |> actor.start() - }) -} - -fn handle_pubsub_message( - clients: List(Subject(String)), - message: PubSubMessage, -) { - case message { - Subscribe(client) -> { - let assert Ok(pid) = process.subject_owner(client) - - logging.log(logging.Info, "Client " <> pid_to_string(pid) <> " connected") - - actor.continue([client, ..clients]) - } - - Unsubscribe(client) -> { - let assert Ok(pid) = process.subject_owner(client) - - { "Client " <> pid_to_string(pid) <> " disconnected" } - |> logging.log(logging.Info, _) - - list.filter(clients, fn(subscribed) { subscribed != client }) - |> actor.continue() - } - - Publish(message) -> { - let pids = - list.fold(over: clients, from: [], with: fn(acc, client) { - let assert Ok(pid) = process.subject_owner(client) - let _ = process.send(client, message) - - [pid_to_string(pid), ..acc] - }) - |> string.join(", ") - - { "Sent message `" <> message <> "` to clients: " <> pids } - |> logging.log(logging.Info, _) - - actor.continue(clients) - } + _method, _path -> + response.new(404) + |> response.set_body(ewe.Empty) } } - -// Utilities -// ----------------------------------------------------------------------------- - -fn pid_to_string(pid: Pid) -> String { - pid_to_list(pid) - |> charlist.to_string() -} - -@external(erlang, "erlang", "pid_to_list") -fn pid_to_list(pid: Pid) -> charlist.Charlist diff --git a/examples/src/streaming_bodies.gleam b/examples/src/streaming_bodies.gleam new file mode 100644 index 0000000..022b26c --- /dev/null +++ b/examples/src/streaming_bodies.gleam @@ -0,0 +1,91 @@ +import ewe +import gleam/bit_array +import gleam/erlang/process +import gleam/http/request +import gleam/http/response +import gleam/int +import gleam/result +import logging + +const body_limit = 10_485_760 + +pub fn main() { + logging.configure() + logging.set_level(logging.Info) + + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // A server that streams the request body back as a chunked response. + // This demonstrates how to handle large uploads. + // + let assert Ok(_) = + ewe.new(listener_name:, connection_factory_name:, handler: handle_request) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) + |> ewe.start + + process.sleep_forever() +} + +fn handle_request( + req: request.Request(ewe.Connection), +) -> response.Response(ewe.Body) { + case request.path_segments(req) { + // /stream/:max_chunk_bytes controls how many bytes to read at a time. + ["stream", max_chunk_bytes] -> + int.parse(max_chunk_bytes) + |> result.unwrap(16) + |> handle_stream(req, _) + _segments -> + response.new(404) + |> response.set_body(ewe.Empty) + } +} + +fn handle_stream( + req: request.Request(ewe.Connection), + max_chunk_bytes: Int, +) -> response.Response(ewe.Body) { + let content_type = + request.get_header(req, "content-type") + |> result.unwrap("application/octet-stream") + + // Stream the response. The body is framed as chunked and every chunk we send + // reaches the client right away. The handler runs in the same connection + // process and owns the stream until it finishes it. + response.new(200) + |> response.set_header("content-type", content_type) + // Remember, `echo_body(req, _, max_chunk_bytes)` is the same as: + // fn(writer) { echo_body(req, writer, max_chunk_bytes) } + |> ewe.stream_response(echo_body(req, _, max_chunk_bytes)) +} + +// Let's read the request body one chunk in a time and write each one back out, +// acting as an echo. The request body returned by `Chunk` case carries the +// rest of the body so it has to be fed into the next read to achieve correct +// reading. +fn echo_body( + req: request.Request(ewe.Connection), + writer: ewe.ResponseWriter, + max_chunk_bytes: Int, +) -> Result(Nil, ewe.SendError) { + // Simulating processing delay here, like some work being done... + process.sleep(int.random(250)) + + case ewe.read_body_chunk(req, max_chunk_bytes:, limit: body_limit) { + Ok(ewe.Chunk(data:, request:)) -> { + logging.log(logging.Info, { + "Consumed " <> int.to_string(bit_array.byte_size(data)) <> " bytes." + }) + + use writer <- result.try(ewe.send_chunk(writer, data)) + echo_body(request, writer, max_chunk_bytes) + } + Ok(ewe.Done(_request)) -> ewe.finish_response(writer) + Error(_body_error) -> { + logging.log(logging.Info, "Failed to read the request body.") + ewe.finish_response(writer) + } + } +} diff --git a/examples/src/streaming_body.gleam b/examples/src/streaming_body.gleam deleted file mode 100644 index 0b8c9c0..0000000 --- a/examples/src/streaming_body.gleam +++ /dev/null @@ -1,126 +0,0 @@ -import ewe.{type Request, type Response} -import gleam/bit_array -import gleam/erlang/process.{type Subject} -import gleam/http/request -import gleam/http/response -import gleam/int -import gleam/result -import logging - -pub fn main() { - logging.configure() - logging.set_level(logging.Info) - - // A server that streams the request body back as chunked response. - // This demonstrates how to handle large uploads. - // - let assert Ok(_) = - ewe.new(handler) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) - |> ewe.start() - - process.sleep_forever() -} - -fn handler(req: Request) -> Response { - // Route: /stream/{chunk_size} - controls how many bytes to read at a time. - // - case request.path_segments(req) { - ["stream", chunk_size] -> - int.parse(chunk_size) - |> result.unwrap(16) - |> handle_stream(req, _) - _ -> - response.new(404) - |> response.set_body(ewe.Empty) - } -} - -pub type Message { - Chunk(BitArray) - Done - BodyError(ewe.BodyError) -} - -fn handle_stream(req: Request, chunk_size: Int) -> Response { - let content_type = - request.get_header(req, "content-type") - |> result.unwrap("application/octet-stream") - - // Get a consumer function for streaming the request body. This allows - // reading data incrementally. - // - case ewe.stream_body(req) { - Ok(consumer) -> { - // For example purporses, let's set up a chunked response. The response is - // sent in chunks as we consume the request body. - // - ewe.chunked_body( - req, - response.new(200) |> response.set_header("content-type", content_type), - // Spawn a separate process to consume the body and send chunks. - // This prevents blocking the handler while reading data. - // - on_init: fn(subject) { - let _pid = - fn() { stream_resource(consumer, subject, chunk_size) } - |> process.spawn - }, - handler: fn(chunked_body, state, message) { - case message { - Chunk(data) -> - case ewe.send_chunk(chunked_body, data) { - Ok(Nil) -> ewe.chunked_continue(state) - Error(_) -> ewe.chunked_stop_abnormal("Failed to send chunk") - } - Done -> ewe.chunked_stop() - BodyError(_body_error) -> - ewe.chunked_stop_abnormal("failed to read body") - } - }, - on_close: fn(_conn, _state) { - logging.log(logging.Info, "Stream closed") - }, - ) - } - Error(_) -> - response.new(400) - |> response.set_header("content-type", "text/plain; charset=utf-8") - |> response.set_body(ewe.TextData("Invalid request")) - } -} - -// Recursively consume chunks from the request body and send them to the -// chunked response handler via the subject. -// -fn stream_resource( - consumer: ewe.Consumer, - subject: Subject(Message), - chunk_size: Int, -) -> Nil { - // Simulating processing delay here... - // - process.sleep(int.random(250)) - // Call the consumer with the chunk size. It returns the next chunk of data - // and a new consumer for the remaining body. - // - case consumer(chunk_size) { - Ok(ewe.Consumed(data, next)) -> { - logging.log(logging.Info, { - "Consumed " <> int.to_string(bit_array.byte_size(data)) <> " bytes." - }) - - process.send(subject, Chunk(data)) - // Recursively process the next chunk. - // - stream_resource(next, subject, chunk_size) - } - Ok(ewe.Done) -> { - process.send(subject, Done) - } - Error(body_error) -> { - process.send(subject, BodyError(body_error)) - } - } -} diff --git a/examples/src/websocket.gleam b/examples/src/websocket.gleam index ac5de07..0c5d27a 100644 --- a/examples/src/websocket.gleam +++ b/examples/src/websocket.gleam @@ -1,14 +1,9 @@ -import ewe.{type Request, type Response} -import gleam/dict -import gleam/erlang/charlist.{type Charlist} -import gleam/erlang/process.{type Name, type Pid, type Subject} +import ewe +import examples/pubsub +import gleam/erlang/process.{type Subject} import gleam/http/request import gleam/http/response -import gleam/list -import gleam/option.{None, Some} -import gleam/otp/actor import gleam/otp/static_supervisor as supervisor -import gleam/otp/supervision.{type ChildSpecification} import logging pub fn main() { @@ -22,15 +17,22 @@ pub fn main() { let pubsub_name = process.new_name("pubsub") let pubsub = process.named_subject(pubsub_name) + let listener_name = process.new_name("listener_name") + let connection_factory_name = process.new_name("connection_factory_name") + + // Remember, `handle_request(_, pubsub)` is the same as: + // fn(request) { handle_request(request, pubsub) } + let handler = handle_request(_, pubsub) + // Set up supervision for both pubsub and the web server. // let assert Ok(_) = supervisor.new(supervisor.OneForAll) - |> supervisor.add(pubsub_worker(pubsub_name)) + |> supervisor.add(pubsub.worker(pubsub_name)) |> supervisor.add( - ewe.new(handler(_, pubsub)) - |> ewe.bind("0.0.0.0") - |> ewe.listening(port: 8080) + ewe.new(listener_name:, connection_factory_name:, handler:) + |> ewe.bind(to: "0.0.0.0") + |> ewe.listening(on: 8080) |> ewe.supervised, ) |> supervisor.start @@ -38,65 +40,63 @@ pub fn main() { process.sleep_forever() } -fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { +fn handle_request( + req: request.Request(ewe.Connection), + pubsub: Subject(pubsub.Message(Broadcast)), +) -> response.Response(ewe.Body) { case request.path_segments(req) { ["topic", topic] -> handle_topic(req, pubsub, topic) - _ -> + _segments -> response.new(404) |> response.set_body(ewe.Empty) } } -// Websocket -// ----------------------------------------------------------------------------- - type WebsocketState { WebsocketState( - pubsub: Subject(PubSubMessage), + pubsub: Subject(pubsub.Message(Broadcast)), topic: String, client: Subject(Broadcast), ) } +// What one client sends to everyone else subscribed to the same topic. type Broadcast { Text(String) Bytes(BitArray) } -fn handle_topic(req: Request, pubsub: Subject(PubSubMessage), topic: String) { +fn handle_topic( + req: request.Request(ewe.Connection), + pubsub: Subject(pubsub.Message(Broadcast)), + topic: String, +) -> response.Response(ewe.Body) { // Upgrade the HTTP connection to WebSocket. Unlike SSE, WebSocket is - // bidirectional - both client and server can send messages at any time. + // bidirectional, so both client and server can send messages at any time. + // The upgrade needs an HTTP/1.1 connection, an HTTP/2 one is answered + // with a 501. // - ewe.upgrade_websocket( - req, + ewe.websocket( + request: req, // Initialize the WebSocket connection. The selector allows receiving // messages from both the WebSocket and the pubsub system. - // on_init: fn(_conn, selector) { - logging.log( - logging.Info, - "WebSocket connection opened: " <> pid_to_string(process.self()), - ) + logging.log(logging.Info, "WebSocket connection opened") let client = process.new_subject() - process.send(pubsub, Subscribe(topic:, client:)) + pubsub.subscribe(pubsub, topic:, client:) let state = WebsocketState(pubsub:, topic:, client:) // Add the client subject to the selector to receive broadcast messages. - // let selector = process.select(selector, client) #(state, selector) }, handler: handle_websocket_message, on_close: fn(_conn, state) { - let assert Ok(pid) = process.subject_owner(state.client) - logging.log( - logging.Info, - "WebSocket connection closed: " <> pid_to_string(pid), - ) + logging.log(logging.Info, "WebSocket connection closed") - process.send(pubsub, Unsubscribe(state.topic, state.client)) + pubsub.unsubscribe(state.pubsub, topic: state.topic, client: state.client) }, ) } @@ -107,133 +107,33 @@ fn handle_topic(req: Request, pubsub: Subject(PubSubMessage), topic: String) { fn handle_websocket_message( conn: ewe.WebsocketConnection, state: WebsocketState, - msg: ewe.WebsocketMessage(Broadcast), + message: ewe.WebsocketMessage(Broadcast), ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { - case msg { - // Text message from the client - broadcast to all subscribers. - // - ewe.Text(text) -> { - process.send(state.pubsub, Publish(state.topic, Text(text))) + case message { + // Text frame from the client; broadcast to all subscribers. + ewe.TextFrame(text) -> { + pubsub.publish(state.pubsub, topic: state.topic, message: Text(text)) ewe.websocket_continue(state) } - // Binary message from the client - broadcast to all subscribers. - // - ewe.Binary(binary) -> { - process.send(state.pubsub, Publish(state.topic, Bytes(binary))) + // Binary frame from the client; broadcast to all subscribers. + ewe.BinaryFrame(data) -> { + pubsub.publish(state.pubsub, topic: state.topic, message: Bytes(data)) ewe.websocket_continue(state) } - // User message from the pubsub - forward to this client. - // - ewe.User(message) -> { - let assert Ok(_) = case message { + // Message from the pubsub; forward to this client. + ewe.UserMessage(broadcast) -> { + let sent = case broadcast { Text(text) -> ewe.send_text_frame(conn, text) - Bytes(binary) -> ewe.send_binary_frame(conn, binary) - } - - ewe.websocket_continue(state) - } - } -} - -// PubSub -// ----------------------------------------------------------------------------- - -type PubSubMessage { - Subscribe(topic: String, client: Subject(Broadcast)) - Publish(topic: String, message: Broadcast) - Unsubscribe(topic: String, client: Subject(Broadcast)) -} - -fn pubsub_worker( - named: Name(PubSubMessage), -) -> ChildSpecification(Subject(PubSubMessage)) { - let pubsub = - dict.new() - |> actor.new - |> actor.on_message(handle_pubsub_message) - |> actor.named(named) - - supervision.worker(fn() { - logging.log(logging.Info, "Starting pubsub worker") - actor.start(pubsub) - }) -} - -fn handle_pubsub_message(state, message) { - case message { - Subscribe(topic:, client:) -> { - let new_state = - dict.upsert(in: state, update: topic, with: fn(clients) { - case clients { - Some(clients) -> [client, ..clients] - None -> { - logging.log(logging.Info, "Creating topic " <> topic) - [client] - } - } - }) - - let assert Ok(pid) = process.subject_owner(client) - logging.log( - logging.Info, - "Subscribing client " <> pid_to_string(pid) <> " to topic " <> topic, - ) - - actor.continue(new_state) - } - Publish(topic:, message:) -> { - case message { - Text(text) -> - logging.log( - logging.Info, - "Publishing text message `" <> text <> "` to topic " <> topic, - ) - Bytes(_binary) -> - logging.log( - logging.Info, - "Publishing binary message to topic " <> topic, - ) + Bytes(data) -> ewe.send_binary_frame(conn, data) } - case dict.get(state, topic) { - Ok(clients) -> list.each(clients, actor.send(_, message)) - Error(_) -> Nil + case sent { + Ok(Nil) -> ewe.websocket_continue(state) + Error(_send_error) -> + ewe.websocket_stop_abnormal("Failed to send a frame") } - - actor.continue(state) - } - Unsubscribe(topic:, client:) -> { - let assert Ok(pid) = process.subject_owner(client) - logging.log( - logging.Info, - "Unsubscribing client " <> pid_to_string(pid) <> " from topic " <> topic, - ) - - let new_state = case dict.get(state, topic) { - Ok([_]) | Ok([]) -> { - logging.log(logging.Info, "Dropping topic " <> topic) - dict.drop(state, [topic]) - } - Ok(clients) -> { - list.filter(clients, fn(c) { c != client }) - |> dict.insert(state, topic, _) - } - Error(_) -> state - } - - actor.continue(new_state) } } } - -// Utilities -// ----------------------------------------------------------------------------- - -fn pid_to_string(pid: Pid) -> String { - charlist.to_string(pid_to_list(pid)) -} - -@external(erlang, "erlang", "pid_to_list") -fn pid_to_list(pid: Pid) -> Charlist