+5

CHANGELOG.md

Reviewed

··· 1 1 # Changelog 2 2 3 3 + # Unreleased 4 4 + 5 5 + - Move examples to appropriate folder 6 6 + - Improve examples with useful documentation lines 7 7 + 3 8 # v2.1.2 - 19.11.2025 4 9 5 10 - Improve internal codebase

+121 -487

README.md

Reviewed

··· 2 2 3 3 # 🐑 ewe 4 4 5 5 - ewe [/juː/] - fluffy package for building web servers. Inspired by [mist](https://github.com/rawhat/mist). 5 5 + ewe [/juː/] - fluffy package for building web servers. 6 6 7 7 [](https://hex.pm/packages/ewe) 8 8 [](https://hexdocs.pm/ewe/) ··· 13 13 gleam add ewe@2 gleam_erlang gleam_otp gleam_http logging 14 14 ``` 15 15 16 16 - ## Quick Start 16 16 + ## Getting Started 17 17 18 18 ```gleam 19 19 import gleam/erlang/process ··· 21 21 import gleam/http/response 22 22 23 23 import ewe.{type Request, type Response} 24 24 - 25 25 - fn handler(_req: Request) -> Response { 26 26 - response.new(200) 27 27 - |> response.set_header("content-type", "text/plain; charset=utf-8") 28 28 - |> response.set_body(ewe.TextData("Hello, World!")) 29 29 - } 30 24 31 25 pub fn main() { 32 26 logging.configure() ··· 34 28 35 29 let assert Ok(_) = 36 30 ewe.new(handler) 37 37 - |> ewe.bind_all() 31 31 + |> ewe.bind_all 38 32 |> ewe.listening(port: 8080) 39 39 - |> ewe.start() 33 33 + |> ewe.start 40 34 41 35 process.sleep_forever() 42 36 } 37 37 + 38 38 + fn handler(_req: Request) -> Response { 39 39 + response.new(200) 40 40 + |> response.set_header("content-type", "text/plain; charset=utf-8") 41 41 + |> response.set_body(ewe.TextData("Hello, World!")) 42 42 + } 43 43 ``` 44 44 45 45 ## Usage 46 46 47 47 - ### [Sending Response](test/preview/sending_response.gleam) 47 47 + ### [Sending Response](examples/src/sending_response.gleam) 48 48 49 49 `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)`. 50 50 ··· 61 61 fn handler(req: Request(Connection)) -> Response(ResponseBody) { 62 62 case request.path_segments(req) { 63 63 ["hello", name] -> { 64 64 + // Use TextData for text responses. 65 65 + // 64 66 response.new(200) 65 67 |> response.set_header("content-type", "text/plain; charset=utf-8") 66 68 |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) 67 69 } 68 70 ["bytes", amount] -> { 71 71 + // Use BitsData for binary responses. 72 72 + // 69 73 let random_bytes = 70 74 int.parse(amount) 71 75 |> result.unwrap(0) ··· 76 80 |> response.set_body(ewe.BitsData(random_bytes)) 77 81 } 78 82 _ -> 83 83 + // Use Empty for responses with no body (like 404, 204, etc). 84 84 + // 79 85 response.new(404) 80 86 |> response.set_body(ewe.Empty) 81 87 } 82 88 } 83 89 ``` 84 90 85 85 - ### [Getting Request Body](test/preview/getting_request_body.gleam) 91 91 + ### [Reading Body](examples/src/reading_body.gleam) 86 92 87 93 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. 88 94 ··· 93 99 94 100 import ewe.{type Request, type Response} 95 101 96 96 - fn handle_echo(req: Request) -> Response { 102 102 + fn handler(req: Request) -> Response { 97 103 let content_type = 98 104 request.get_header(req, "content-type") 99 105 |> result.unwrap("application/octet-stream") 100 106 101 101 - case ewe.read_body(req, 1024) { 107 107 + // Read the entire request body into memory with a 10KB limit. This blocks 108 108 + // until the full body is received. 109 109 + // 110 110 + case ewe.read_body(req, 10_240) { 102 111 Ok(req) -> 103 112 response.new(200) 104 113 |> response.set_header("content-type", content_type) ··· 115 124 } 116 125 ``` 117 126 118 118 - ### [Streaming](test/preview/streaming.gleam) 127 127 + ### [Streaming Body](examples/src/streaming_body.gleam) 119 128 120 129 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. 121 130 ··· 129 138 BodyError(ewe.BodyError) 130 139 } 131 140 141 141 + // Recursively consume chunks from the request body and send them to the 142 142 + // chunked response handler via the subject. 143 143 + // 132 144 fn stream_resource( 133 145 consumer: ewe.Consumer, 134 146 subject: Subject(Message), 135 147 chunk_size: Int, 136 148 ) -> Nil { 137 137 - // Simulating delay for demonstration purpose 138 149 process.sleep(int.random(250)) 150 150 + // Call the consumer with the chunk size. It returns the next chunk of data 151 151 + // and a new consumer for the remaining body. 152 152 + // 139 153 case consumer(chunk_size) { 140 154 Ok(ewe.Consumed(data, next)) -> { 141 155 logging.log(logging.Info, { ··· 146 160 }) 147 161 148 162 process.send(subject, Chunk(data)) 163 163 + // Recursively process the next chunk. 164 164 + // 149 165 stream_resource(next, subject, chunk_size) 150 166 } 151 151 - Ok(ewe.Done) -> { 152 152 - process.send(subject, Done) 153 153 - } 154 154 - Error(body_error) -> { 155 155 - process.send(subject, BodyError(body_error)) 156 156 - } 167 167 + Ok(ewe.Done) -> process.send(subject, Done) 168 168 + Error(body_error) -> process.send(subject, BodyError(body_error)) 157 169 } 158 170 } 159 171 ··· 162 174 request.get_header(req, "content-type") 163 175 |> result.unwrap("application/octet-stream") 164 176 177 177 + // Get a consumer function for streaming the request body. 178 178 + // 165 179 case ewe.stream_body(req) { 166 180 Ok(consumer) -> { 181 181 + // Set up a chunked response. The response is sent in chunks as we 182 182 + // consume the request body. 183 183 + // 167 184 ewe.chunked_body( 168 185 req, 169 186 response.new(200) |> response.set_header("content-type", content_type), 187 187 + // Spawn a separate process to consume the body and send chunks. 188 188 + // This prevents blocking the handler while reading data. 189 189 + // 170 190 on_init: fn(subject) { 171 171 - process.spawn(fn() { stream_resource(consumer, subject, chunk_size) }) 172 172 - 173 173 - Nil 191 191 + let _pid = 192 192 + fn() { stream_resource(consumer, subject, chunk_size) } 193 193 + |> process.spawn 174 194 }, 175 195 handler: fn(chunked_body, state, message) { 176 196 case message { ··· 197 217 } 198 218 ``` 199 219 200 200 - ### [File Serving](test/preview/file_serving.gleam) 220 220 + ### [Serving Files](examples/src/serving_files.gleam) 201 221 202 222 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. 203 223 ··· 208 228 import ewe.{type Response} 209 229 210 230 fn serve_file(path: String) -> Response { 231 231 + // Load file from disk using ewe.file(). This efficiently streams the file 232 232 + // content without loading it entirely into memory. 233 233 + // 234 234 + // In production, make sure to validate paths to prevent directory traversal 235 235 + // attacks! (e.g., requests to "../../../etc/passwd") 236 236 + // 211 237 case ewe.file("public" <> path, offset: None, limit: None) { 212 238 Ok(file) -> { 213 239 response.new(200) ··· 223 249 } 224 250 ``` 225 251 226 226 - ## [WebSocket](test/preview/websocket.gleam) 252 252 + ### [WebSocket](examples/src/websocket.gleam) 227 253 228 254 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). 229 255 ··· 257 283 258 284 fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 259 285 case request.path_segments(req) { 260 260 - ["topic", topic] -> 261 261 - ewe.upgrade_websocket( 262 262 - req, 263 263 - on_init: fn(_conn, selector) { 264 264 - logging.log( 265 265 - logging.Info, 266 266 - "WebSocket connection opened: " <> pid_to_string(process.self()), 267 267 - ) 268 268 - 269 269 - let client = process.new_subject() 270 270 - process.send(pubsub, Subscribe(topic:, client:)) 271 271 - 272 272 - let state = WebsocketState(pubsub:, topic:, client:) 273 273 - let selector = process.select(selector, client) 274 274 - 275 275 - #(state, selector) 276 276 - }, 277 277 - handler: handle_websocket, 278 278 - on_close: fn(_conn, state) { 279 279 - let assert Ok(pid) = process.subject_owner(state.client) 280 280 - logging.log( 281 281 - logging.Info, 282 282 - "WebSocket connection closed: " <> pid_to_string(pid), 283 283 - ) 284 284 - 285 285 - process.send(pubsub, Unsubscribe(state.topic, state.client)) 286 286 - }, 287 287 - ) 286 286 + ["topic", topic] -> handle_topic(req, pubsub, topic) 288 287 _ -> 289 288 response.new(404) 290 289 |> response.set_body(ewe.Empty) 291 290 } 292 291 } 293 292 294 294 - fn handle_websocket( 293 293 + fn handle_topic(req: Request, pubsub: Subject(PubSubMessage), topic: String) { 294 294 + // Upgrade the HTTP connection to WebSocket. Unlike SSE, WebSocket is 295 295 + // bidirectional - both client and server can send messages at any time. 296 296 + // 297 297 + ewe.upgrade_websocket( 298 298 + req, 299 299 + // Initialize the WebSocket connection. The selector allows receiving 300 300 + // messages from both the WebSocket and the pubsub system. 301 301 + // 302 302 + on_init: fn(_conn, selector) { 303 303 + let client = process.new_subject() 304 304 + process.send(pubsub, Subscribe(topic:, client:)) 305 305 + 306 306 + let state = WebsocketState(pubsub:, topic:, client:) 307 307 + // Add the client subject to the selector to receive broadcast messages. 308 308 + // 309 309 + let selector = process.select(selector, client) 310 310 + 311 311 + #(state, selector) 312 312 + }, 313 313 + handler: handle_websocket_message, 314 314 + on_close: fn(_conn, state) { 315 315 + process.send(pubsub, Unsubscribe(state.topic, state.client)) 316 316 + }, 317 317 + ) 318 318 + } 319 319 + 320 320 + // Handle three types of messages: text from client, binary from client, 321 321 + // and broadcast messages from the pubsub system. 322 322 + // 323 323 + fn handle_websocket_message( 295 324 conn: ewe.WebsocketConnection, 296 325 state: WebsocketState, 297 326 msg: ewe.WebsocketMessage(Broadcast), 298 327 ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { 299 328 case msg { 329 329 + // Text message from the client - broadcast to all subscribers. 330 330 + // 300 331 ewe.Text(text) -> { 301 332 process.send(state.pubsub, Publish(state.topic, Text(text))) 302 333 ewe.websocket_continue(state) 303 334 } 304 335 336 336 + // Binary message from the client - broadcast to all subscribers. 337 337 + // 305 338 ewe.Binary(binary) -> { 306 339 process.send(state.pubsub, Publish(state.topic, Bytes(binary))) 307 340 ewe.websocket_continue(state) 308 341 } 309 342 343 343 + // User message from the pubsub - forward to this client. 344 344 + // 310 345 ewe.User(message) -> { 311 346 let assert Ok(_) = case message { 312 347 Text(text) -> ewe.send_text_frame(conn, text) ··· 317 352 } 318 353 } 319 354 } 320 320 - 321 321 - fn pid_to_string(pid: Pid) -> String { 322 322 - charlist.to_string(pid_to_list(pid)) 323 323 - } 324 324 - 325 325 - @external(erlang, "erlang", "pid_to_list") 326 326 - fn pid_to_list(pid: Pid) -> Charlist 327 355 ``` 328 356 329 329 - ### [Server-Sent Events](test/preview/sse.gleam) 357 357 + ### [Server-Sent Events](examples/src/sse.gleam) 330 358 331 359 332 360 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. 333 361 334 362 ```gleam 335 363 import gleam/bit_array 336 336 - import gleam/erlang/charlist.{type Charlist} 337 337 - import gleam/erlang/process.{type Pid, type Subject} 338 338 - import gleam/http/request 364 364 + import gleam/erlang/process.{type Subject} 365 365 + import gleam/http 339 366 import gleam/http/response 340 340 - import logging 341 367 342 342 - import ewe.{type Request, type Response} 368 368 + import ewe 343 369 344 370 type PubSubMessage { 345 371 Subscribe(client: Subject(String)) ··· 347 373 Publish(String) 348 374 } 349 375 350 350 - fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 351 351 - case request.path_segments(req) { 352 352 - ["sse"] -> 376 376 + fn handler(req: ewe.Request, pubsub: Subject(PubSubMessage)) -> ewe.Response { 377 377 + case req.method, req.path { 378 378 + http.Get, "/sse" -> 379 379 + // Establish a Server-Sent Events connection. SSE is a one-way channel 380 380 + // from server to client. The connection stays open and the server can 381 381 + // push events at any time. 382 382 + // 353 383 ewe.sse( 354 384 req, 385 385 + // Initialize the connection and subscribe this client to the pubsub. 386 386 + // 355 387 on_init: fn(client) { 356 388 process.send(pubsub, Subscribe(client)) 357 357 - logging.log( 358 358 - logging.Info, 359 359 - "SSE connection opened: " <> pid_to_string(process.self()), 360 360 - ) 361 389 362 390 client 363 391 }, 392 392 + // Handle messages from the pubsub and send them as SSE events. 393 393 + // 364 394 handler: fn(conn, client, message) { 365 395 case ewe.send_event(conn, ewe.event(message)) { 366 396 Ok(Nil) -> ewe.sse_continue(client) 367 397 Error(_) -> ewe.sse_stop() 368 398 } 369 399 }, 400 400 + // Clean up when the client disconnects. 401 401 + // 370 402 on_close: fn(_conn, client) { 371 403 process.send(pubsub, Unsubscribe(client)) 372 372 - logging.log( 373 373 - logging.Info, 374 374 - "SSE connection closed: " <> pid_to_string(process.self()), 375 375 - ) 376 404 }, 377 405 ) 378 378 - ["publish"] -> { 379 379 - case ewe.read_body(req, 1024) { 380 380 - Ok(req) -> { 381 381 - let assert Ok(text) = bit_array.to_string(req.body) 382 382 - process.send(pubsub, Publish(text)) 383 383 - response.new(200) |> response.set_body(ewe.Empty) 384 384 - } 385 385 - Error(_) -> response.new(400) |> response.set_body(ewe.Empty) 386 386 - } 387 387 - } 388 388 - _ -> response.new(404) |> response.set_body(ewe.Empty) 389 389 - } 390 390 - } 391 391 - 392 392 - fn pid_to_string(pid: Pid) -> String { 393 393 - charlist.to_string(pid_to_list(pid)) 394 394 - } 395 395 - 396 396 - @external(erlang, "erlang", "pid_to_list") 397 397 - fn pid_to_list(pid: Pid) -> Charlist 398 398 - ``` 399 399 - 400 400 - ### [Complete Preview](test/preview.gleam) 401 401 - 402 402 - ```gleam 403 403 - import gleam/bit_array 404 404 - import gleam/crypto 405 405 - import gleam/dict 406 406 - import gleam/erlang/charlist.{type Charlist} 407 407 - import gleam/erlang/process.{type Name, type Pid, type Subject} 408 408 - import gleam/http/request 409 409 - import gleam/http/response 410 410 - import gleam/int 411 411 - import gleam/list 412 412 - import gleam/option.{None, Some} 413 413 - import gleam/otp/actor 414 414 - import gleam/otp/static_supervisor as supervisor 415 415 - import gleam/otp/supervision.{type ChildSpecification} 416 416 - import gleam/result 417 417 - import gleam/string 418 418 - import logging 419 419 - 420 420 - import ewe.{type Request, type Response} 421 421 - 422 422 - pub fn main() { 423 423 - logging.configure() 424 424 - logging.set_level(logging.Info) 425 425 - 426 426 - // Create a named subject for the pubsub worker 427 427 - let pubsub_name = process.new_name("pubsub") 428 428 - let pubsub = process.named_subject(pubsub_name) 429 429 - 430 430 - // Configure and start the supervision tree with pubsub worker and the ewe 431 431 - // server, that listens on port 8080 432 432 - let assert Ok(_) = 433 433 - supervisor.new(supervisor.OneForAll) 434 434 - |> supervisor.add(pubsub_worker(pubsub_name)) 435 435 - |> supervisor.add( 436 436 - ewe.new(handler(_, pubsub)) 437 437 - |> ewe.bind_all() 438 438 - |> ewe.listening(port: 8080) 439 439 - |> ewe.supervised(), 440 440 - ) 441 441 - |> supervisor.start() 442 442 - 443 443 - process.sleep_forever() 444 444 - } 445 445 - 446 446 - // Define the messages that can be sent to the pubsub worker 447 447 - type PubSubMessage { 448 448 - Subscribe(topic: String, client: Subject(Broadcast)) 449 449 - Publish(topic: String, message: Broadcast) 450 450 - Unsubscribe(topic: String, client: Subject(Broadcast)) 451 451 - } 452 452 - 453 453 - // Define the messages that could be received by websocket and SSE clients 454 454 - type Broadcast { 455 455 - Text(String) 456 456 - Bytes(BitArray) 457 457 - } 458 458 - 459 459 - // Define the state of the websocket connection 460 460 - type WebsocketState { 461 461 - WebsocketState( 462 462 - pubsub: Subject(PubSubMessage), 463 463 - topic: String, 464 464 - client: Subject(Broadcast), 465 465 - ) 466 466 - } 467 467 - 468 468 - // Main logic of the pubsub worker, that handles the messages and keeps track of 469 469 - // the clients on topics. Its implementation is not really important 470 470 - fn pubsub_worker( 471 471 - named: Name(PubSubMessage), 472 472 - ) -> ChildSpecification(Subject(PubSubMessage)) { 473 473 - let pubsub = 474 474 - actor.new(dict.new()) 475 475 - |> actor.on_message(fn(state, msg) { 476 476 - case msg { 477 477 - Subscribe(topic:, client:) -> { 478 478 - let new_state = 479 479 - dict.upsert(in: state, update: topic, with: fn(clients) { 480 480 - case clients { 481 481 - Some(clients) -> [client, ..clients] 482 482 - None -> { 483 483 - logging.log(logging.Info, "Creating topic " <> topic) 484 484 - [client] 485 485 - } 486 486 - } 487 487 - }) 488 488 - 489 489 - let assert Ok(pid) = process.subject_owner(client) 490 490 - logging.log( 491 491 - logging.Info, 492 492 - "Subscribing client " <> pid_to_string(pid) <> " to topic " <> topic, 493 493 - ) 494 406 495 495 - actor.continue(new_state) 496 496 - } 497 497 - Publish(topic:, message:) -> { 498 498 - case message { 499 499 - Text(text) -> 500 500 - logging.log( 501 501 - logging.Info, 502 502 - "Publishing text message `" <> text <> "` to topic " <> topic, 503 503 - ) 504 504 - Bytes(binary) -> 505 505 - logging.log( 506 506 - logging.Info, 507 507 - "Publishing binary message `" 508 508 - <> string.inspect(binary) 509 509 - <> "` to topic " 510 510 - <> topic, 511 511 - ) 512 512 - } 513 513 - 514 514 - case dict.get(state, topic) { 515 515 - Ok(clients) -> list.each(clients, actor.send(_, message)) 516 516 - Error(_) -> Nil 517 517 - } 518 518 - 519 519 - actor.continue(state) 520 520 - } 521 521 - Unsubscribe(topic:, client:) -> { 522 522 - let assert Ok(pid) = process.subject_owner(client) 523 523 - logging.log( 524 524 - logging.Info, 525 525 - "Unsubscribing client " 526 526 - <> pid_to_string(pid) 527 527 - <> " from topic " 528 528 - <> topic, 529 529 - ) 407 407 + // Accept messages via POST and broadcast them to all SSE clients. 408 408 + // 409 409 + http.Post, "/post" -> { 410 410 + case ewe.read_body(req, 128) { 411 411 + Ok(req) -> { 412 412 + case bit_array.to_string(req.body) { 413 413 + Ok(message) -> { 414 414 + process.send(pubsub, Publish(message)) 530 415 531 531 - let new_state = case dict.get(state, topic) { 532 532 - Ok([_]) | Ok([]) -> { 533 533 - logging.log(logging.Info, "Dropping topic " <> topic) 534 534 - dict.drop(state, [topic]) 416 416 + response.new(200) |> response.set_body(ewe.Empty) 535 417 } 536 536 - Ok(clients) -> { 537 537 - list.filter(clients, fn(c) { c != client }) 538 538 - |> dict.insert(state, topic, _) 539 539 - } 540 540 - Error(_) -> state 418 418 + Error(Nil) -> response.new(400) |> response.set_body(ewe.Empty) 541 419 } 542 542 - 543 543 - actor.continue(new_state) 544 420 } 421 421 + Error(_) -> response.new(400) |> response.set_body(ewe.Empty) 545 422 } 546 546 - }) 547 547 - |> actor.named(named) 548 548 - 549 549 - supervision.worker(fn() { 550 550 - logging.log(logging.Info, "Starting pubsub worker") 551 551 - actor.start(pubsub) 552 552 - }) 553 553 - } 554 554 - 555 555 - // Main HTTP request handler that routes requests to different endpoints 556 556 - fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 557 557 - case request.path_segments(req) { 558 558 - // GET /hello/:name - Simple greeting endpoint 559 559 - ["hello", name] -> { 560 560 - response.new(200) 561 561 - |> response.set_header("content-type", "text/plain; charset=utf-8") 562 562 - |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) 563 423 } 564 564 - // GET /bytes/:amount - Generate random N bytes 565 565 - ["bytes", amount] -> { 566 566 - let random_bytes = 567 567 - int.parse(amount) 568 568 - |> result.unwrap(0) 569 569 - |> crypto.strong_random_bytes() 570 424 571 571 - response.new(200) 572 572 - |> response.set_header("content-type", "application/octet-stream") 573 573 - |> response.set_body(ewe.BitsData(random_bytes)) 574 574 - } 575 575 - 576 576 - // POST /echo - Echo back the request body 577 577 - ["echo"] -> handle_echo(req) 578 578 - // POST /stream/:chunk_size - Stream and echo back the request body in chunks 579 579 - ["stream", chunk_size] -> 580 580 - handle_stream(req, int.parse(chunk_size) |> result.unwrap(16)) 581 581 - 582 582 - // GET /file/:path - Serve a file from the public directory 583 583 - ["file", path] -> serve_file(path) 584 584 - 585 585 - // POST /topic/:topic/ws - Upgrade to WebSocket connection 586 586 - ["topic", topic, "ws"] -> 587 587 - ewe.upgrade_websocket( 588 588 - req, 589 589 - on_init: fn(_conn, selector) { 590 590 - logging.log( 591 591 - logging.Info, 592 592 - "WebSocket connection opened: " <> pid_to_string(process.self()), 593 593 - ) 594 594 - 595 595 - let client = process.new_subject() 596 596 - process.send(pubsub, Subscribe(topic:, client:)) 597 597 - 598 598 - let state = WebsocketState(pubsub:, topic:, client:) 599 599 - let selector = process.select(selector, client) 600 600 - 601 601 - #(state, selector) 602 602 - }, 603 603 - handler: handle_websocket, 604 604 - on_close: fn(_conn, state) { 605 605 - let assert Ok(pid) = process.subject_owner(state.client) 606 606 - logging.log( 607 607 - logging.Info, 608 608 - "WebSocket connection closed: " <> pid_to_string(pid), 609 609 - ) 610 610 - 611 611 - process.send(pubsub, Unsubscribe(state.topic, state.client)) 612 612 - }, 613 613 - ) 614 614 - 615 615 - // POST /topic/:topic/sse - Switch to Server-Sent Events connection 616 616 - ["topic", topic, "sse"] -> 617 617 - ewe.sse( 618 618 - req, 619 619 - on_init: fn(client) { 620 620 - logging.log( 621 621 - logging.Info, 622 622 - "SSE connection opened: " <> pid_to_string(process.self()), 623 623 - ) 624 624 - 625 625 - process.send(pubsub, Subscribe(topic:, client:)) 626 626 - client 627 627 - }, 628 628 - handler: fn(conn, client, message) { 629 629 - let assert Ok(_) = case message { 630 630 - Text(text) -> ewe.send_event(conn, ewe.event(text)) 631 631 - _ -> Ok(Nil) 632 632 - } 633 633 - 634 634 - ewe.sse_continue(client) 635 635 - }, 636 636 - on_close: fn(_conn, client) { 637 637 - logging.log( 638 638 - logging.Info, 639 639 - "SSE connection closed: " <> pid_to_string(process.self()), 640 640 - ) 641 641 - 642 642 - process.send(pubsub, Unsubscribe(topic:, client:)) 643 643 - }, 644 644 - ) 645 645 - 646 646 - // All other routes return 404 647 647 - _ -> 648 648 - response.new(404) 649 649 - |> response.set_body(ewe.Empty) 650 650 - } 651 651 - } 652 652 - 653 653 - fn handle_echo(req: Request) -> Response { 654 654 - let content_type = 655 655 - request.get_header(req, "content-type") 656 656 - |> result.unwrap("application/octet-stream") 657 657 - 658 658 - case ewe.read_body(req, 1024) { 659 659 - Ok(req) -> 660 660 - response.new(200) 661 661 - |> response.set_header("content-type", content_type) 662 662 - |> response.set_body(ewe.BitsData(req.body)) 663 663 - Error(ewe.BodyTooLarge) -> 664 664 - response.new(413) 665 665 - |> response.set_header("content-type", "text/plain; charset=utf-8") 666 666 - |> response.set_body(ewe.TextData("Body too large")) 667 667 - Error(ewe.InvalidBody) -> 668 668 - response.new(400) 669 669 - |> response.set_header("content-type", "text/plain; charset=utf-8") 670 670 - |> response.set_body(ewe.TextData("Invalid request")) 671 671 - } 672 672 - } 673 673 - 674 674 - pub type StreamMessage { 675 675 - Chunk(BitArray) 676 676 - Done 677 677 - BodyError(ewe.BodyError) 678 678 - } 679 679 - 680 680 - fn stream_resource( 681 681 - consumer: ewe.Consumer, 682 682 - subject: Subject(StreamMessage), 683 683 - chunk_size: Int, 684 684 - ) -> Nil { 685 685 - process.sleep(int.random(250)) 686 686 - case consumer(chunk_size) { 687 687 - Ok(ewe.Consumed(data, next)) -> { 688 688 - logging.log(logging.Info, { 689 689 - "Consumed " 690 690 - <> int.to_string(bit_array.byte_size(data)) 691 691 - <> " bytes: " 692 692 - <> string.inspect(data) 693 693 - }) 694 694 - 695 695 - process.send(subject, Chunk(data)) 696 696 - stream_resource(next, subject, chunk_size) 697 697 - } 698 698 - Ok(ewe.Done) -> { 699 699 - process.send(subject, Done) 700 700 - } 701 701 - Error(body_error) -> { 702 702 - process.send(subject, BodyError(body_error)) 703 703 - } 704 704 - } 705 705 - } 706 706 - 707 707 - fn handle_stream(req: Request, chunk_size: Int) -> Response { 708 708 - let content_type = 709 709 - request.get_header(req, "content-type") 710 710 - |> result.unwrap("application/octet-stream") 711 711 - 712 712 - case ewe.stream_body(req) { 713 713 - Ok(consumer) -> { 714 714 - ewe.chunked_body( 715 715 - req, 716 716 - response.new(200) |> response.set_header("content-type", content_type), 717 717 - on_init: fn(subject) { 718 718 - process.spawn(fn() { stream_resource(consumer, subject, chunk_size) }) 719 719 - 720 720 - Nil 721 721 - }, 722 722 - handler: fn(chunked_body, state, message) { 723 723 - case message { 724 724 - Chunk(data) -> 725 725 - case ewe.send_chunk(chunked_body, data) { 726 726 - Ok(Nil) -> ewe.chunked_continue(state) 727 727 - Error(_) -> ewe.chunked_stop_abnormal("Failed to send chunk") 728 728 - } 729 729 - Done -> ewe.chunked_stop() 730 730 - BodyError(body_error) -> 731 731 - ewe.chunked_stop_abnormal(string.inspect(body_error)) 732 732 - } 733 733 - }, 734 734 - on_close: fn(_conn, _state) { 735 735 - logging.log(logging.Info, "Stream closed") 736 736 - }, 737 737 - ) 738 738 - } 739 739 - Error(_) -> 740 740 - response.new(400) 741 741 - |> response.set_header("content-type", "text/plain; charset=utf-8") 742 742 - |> response.set_body(ewe.TextData("Invalid request")) 743 743 - } 744 744 - } 745 745 - 746 746 - fn serve_file(path: String) -> Response { 747 747 - case ewe.file("public" <> path, offset: None, limit: None) { 748 748 - Ok(file) -> { 749 749 - response.new(200) 750 750 - |> response.set_header("content-type", "application/octet-stream") 751 751 - |> response.set_body(file) 752 752 - } 753 753 - Error(_) -> { 754 754 - response.new(404) 755 755 - |> response.set_header("content-type", "text/plain; charset=utf-8") 756 756 - |> response.set_body(ewe.TextData("File not found")) 757 757 - } 758 758 - } 759 759 - } 760 760 - 761 761 - fn handle_websocket( 762 762 - conn: ewe.WebsocketConnection, 763 763 - state: WebsocketState, 764 764 - msg: ewe.WebsocketMessage(Broadcast), 765 765 - ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { 766 766 - case msg { 767 767 - ewe.Text(text) -> { 768 768 - process.send(state.pubsub, Publish(state.topic, Text(text))) 769 769 - ewe.websocket_continue(state) 770 770 - } 771 771 - 772 772 - ewe.Binary(binary) -> { 773 773 - process.send(state.pubsub, Publish(state.topic, Bytes(binary))) 774 774 - ewe.websocket_continue(state) 775 775 - } 776 776 - 777 777 - ewe.User(message) -> { 778 778 - let assert Ok(_) = case message { 779 779 - Text(text) -> ewe.send_text_frame(conn, text) 780 780 - Bytes(binary) -> ewe.send_binary_frame(conn, binary) 781 781 - } 782 782 - 783 783 - ewe.websocket_continue(state) 784 784 - } 425 425 + _, _ -> response.new(404) |> response.set_body(ewe.Empty) 785 426 } 786 427 } 787 787 - 788 788 - fn pid_to_string(pid: Pid) -> String { 789 789 - charlist.to_string(pid_to_list(pid)) 790 790 - } 791 791 - 792 792 - @external(erlang, "erlang", "pid_to_list") 793 793 - fn pid_to_list(pid: Pid) -> Charlist 794 428 ``` 795 429 796 430 ## API Reference 797 431 798 798 - For detailed API documentation, see [hexdocs.pm/ewe](https://hexdocs.pm/ewe/ewe.html). 432 432 + For detailed API documentation, see [hexdocs.pm/ewe](https://hexdocs.pm/ewe/ewe.html).

+25

examples/README.md

Reviewed

··· 1 1 + # Ewe Examples 2 2 + 3 3 + This directory contains practical examples demonstrating various features of ewe. 4 4 + 5 5 + To run an example, use the following command: 6 6 + ```sh 7 7 + gleam run -m <example name> 8 8 + ``` 9 9 + 10 10 + For example, to run the getting started example: 11 11 + ```sh 12 12 + gleam run -m getting_started 13 13 + ``` 14 14 + 15 15 + ## Examples 16 16 + 17 17 + Here is a list of all the examples: 18 18 + 19 19 + - [getting_started](./src/getting_started.gleam) - Basic HTTP server with "Hello, World!" response 20 20 + - [sending_response](./src/sending_response.gleam) - Different response body types (text, binary, empty) 21 21 + - [reading_body](./src/reading_body.gleam) - Reading and echoing request bodies with size limits 22 22 + - [streaming_body](./src/streaming_body.gleam) - Streaming large request/response bodies in chunks 23 23 + - [serving_files](./src/serving_files.gleam) - Serving static files from disk 24 24 + - [websocket](./src/websocket.gleam) - WebSocket connections with topic-based pubsub 25 25 + - [sse](./src/sse.gleam) - Server-Sent Events for real-time server-to-client updates

+15

examples/gleam.toml

Reviewed

··· 1 1 + name = "examples" 2 2 + version = "1.0.0" 3 3 + description = "A collection of ewe examples" 4 4 + 5 5 + [dependencies] 6 6 + gleam_stdlib = ">= 0.44.0 and < 2.0.0" 7 7 + gleam_erlang = ">= 1.3.0 and < 2.0.0" 8 8 + gleam_otp = ">= 1.2.0 and < 2.0.0" 9 9 + logging = ">= 1.3.0 and < 2.0.0" 10 10 + ewe = { path = "../" } 11 11 + gleam_http = ">= 4.3.0 and < 5.0.0" 12 12 + gleam_crypto = ">= 1.5.1 and < 2.0.0" 13 13 + 14 14 + [dev-dependencies] 15 15 + gleeunit = ">= 1.0.0 and < 2.0.0"

+36

examples/src/getting_started.gleam

Reviewed

··· 1 1 + import ewe.{type Request, type Response} 2 2 + import gleam/erlang/process 3 3 + import gleam/http/response 4 4 + import logging 5 5 + 6 6 + pub fn main() { 7 7 + // This sets the logger to print Info level logs. I recommend using `logging` 8 8 + // package, unless you prefer other tools. 9 9 + // 10 10 + logging.configure() 11 11 + logging.set_level(logging.Info) 12 12 + 13 13 + // Start the ewe web server. 14 14 + // 15 15 + let assert Ok(_) = 16 16 + ewe.new(handler) 17 17 + |> ewe.bind_all 18 18 + |> ewe.listening(port: 8080) 19 19 + |> ewe.start 20 20 + 21 21 + // Put process into sleep. 22 22 + // 23 23 + process.sleep_forever() 24 24 + } 25 25 + 26 26 + // This is the HTTP request handler. 27 27 + // 28 28 + fn handler(_req: Request) -> Response { 29 29 + // When sending response with body, it is important to include `content-type` 30 30 + // header representing what type your body is. You don't need to specify 31 31 + // `content-length`, it is calculated automatically by ewe. 32 32 + // 33 33 + response.new(200) 34 34 + |> response.set_header("content-type", "text/plain; charset=utf-8") 35 35 + |> response.set_body(ewe.TextData("Hello, World!")) 36 36 + }

+58

examples/src/sending_response.gleam

Reviewed

··· 1 1 + import ewe.{type Connection, type ResponseBody} 2 2 + import gleam/crypto 3 3 + import gleam/erlang/process 4 4 + import gleam/http/request.{type Request} 5 5 + import gleam/http/response.{type Response} 6 6 + import gleam/int 7 7 + import gleam/result 8 8 + import logging 9 9 + 10 10 + pub fn main() { 11 11 + logging.configure() 12 12 + logging.set_level(logging.Info) 13 13 + 14 14 + // A server demonstrating different response body types and path routing. 15 15 + // 16 16 + let assert Ok(_) = 17 17 + ewe.new(handler) 18 18 + |> ewe.bind_all 19 19 + |> ewe.listening(port: 8080) 20 20 + |> ewe.start 21 21 + 22 22 + process.sleep_forever() 23 23 + } 24 24 + 25 25 + fn handler(req: Request(Connection)) -> Response(ResponseBody) { 26 26 + // Pattern match on path segments for cleaner routing. 27 27 + // Example: "/hello/alice" becomes ["hello", "alice"] 28 28 + // 29 29 + case request.path_segments(req) { 30 30 + ["hello", name] -> { 31 31 + // Here, we will use TextData for text responses. 32 32 + // 33 33 + response.new(200) 34 34 + |> response.set_header("content-type", "text/plain; charset=utf-8") 35 35 + |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) 36 36 + } 37 37 + ["bytes", amount] -> { 38 38 + // Use BitsData for binary responses. We generate random bytes 39 39 + // to demonstrate sending binary data. 40 40 + // 41 41 + let body = 42 42 + int.parse(amount) 43 43 + |> result.unwrap(0) 44 44 + |> crypto.strong_random_bytes 45 45 + |> ewe.BitsData 46 46 + 47 47 + response.new(200) 48 48 + |> response.set_header("content-type", "application/octet-stream") 49 49 + |> response.set_body(body) 50 50 + } 51 51 + _ -> 52 52 + // Use Empty for responses with no body (like 404, 204, etc). 53 53 + // You don't need to set content-type for empty bodies. 54 54 + // 55 55 + response.new(404) 56 56 + |> response.set_body(ewe.Empty) 57 57 + } 58 58 + }

+44

examples/src/serving_files.gleam

Reviewed

··· 1 1 + import ewe.{type Response} 2 2 + import gleam/erlang/process 3 3 + import gleam/http/response 4 4 + import gleam/option.{None} 5 5 + import logging 6 6 + 7 7 + pub fn main() { 8 8 + logging.configure() 9 9 + logging.set_level(logging.Info) 10 10 + 11 11 + // Start a simple file server that serves files from the "public" directory. 12 12 + // 13 13 + let assert Ok(_) = 14 14 + ewe.new(fn(req) { serve_file(req.path) }) 15 15 + |> ewe.bind_all 16 16 + |> ewe.listening(port: 8080) 17 17 + |> ewe.start 18 18 + 19 19 + process.sleep_forever() 20 20 + } 21 21 + 22 22 + fn serve_file(path: String) -> Response { 23 23 + // Load file from disk using ewe.file(). This efficiently streams the file 24 24 + // content without loading it entirely into memory. 25 25 + // 26 26 + // In production, make sure to validate paths to prevent directory traversal 27 27 + // attacks! (e.g., requests to "../../../etc/passwd") 28 28 + // 29 29 + case ewe.file("public" <> path, offset: None, limit: None) { 30 30 + Ok(file) -> { 31 31 + // Using "application/octet-stream" is safe for any file type, but you 32 32 + // may want to specify content-type based on file extension in production. 33 33 + // 34 34 + response.new(200) 35 35 + |> response.set_header("content-type", "application/octet-stream") 36 36 + |> response.set_body(file) 37 37 + } 38 38 + Error(_) -> { 39 39 + response.new(404) 40 40 + |> response.set_header("content-type", "text/plain; charset=utf-8") 41 41 + |> response.set_body(ewe.TextData("File not found")) 42 42 + } 43 43 + } 44 44 + }

+243

examples/src/websocket.gleam

Reviewed

··· 1 1 + import ewe.{type Request, type Response} 2 2 + import gleam/dict 3 3 + import gleam/erlang/charlist.{type Charlist} 4 4 + import gleam/erlang/process.{type Name, type Pid, type Subject} 5 5 + import gleam/http/request 6 6 + import gleam/http/response 7 7 + import gleam/list 8 8 + import gleam/option.{None, Some} 9 9 + import gleam/otp/actor 10 10 + import gleam/otp/static_supervisor as supervisor 11 11 + import gleam/otp/supervision.{type ChildSpecification} 12 12 + import gleam/string 13 13 + import logging 14 14 + 15 15 + pub fn main() { 16 16 + logging.configure() 17 17 + logging.set_level(logging.Info) 18 18 + 19 19 + // Create a named pubsub process for topic-based message broadcasting. 20 20 + // Multiple clients can subscribe to different topics and receive messages 21 21 + // sent to those topics. 22 22 + // 23 23 + let pubsub_name = process.new_name("pubsub") 24 24 + let pubsub = process.named_subject(pubsub_name) 25 25 + 26 26 + // Set up supervision for both pubsub and the web server. 27 27 + // 28 28 + let assert Ok(_) = 29 29 + supervisor.new(supervisor.OneForAll) 30 30 + |> supervisor.add(pubsub_worker(pubsub_name)) 31 31 + |> supervisor.add( 32 32 + ewe.new(handler(_, pubsub)) 33 33 + |> ewe.bind_all 34 34 + |> ewe.listening(port: 8080) 35 35 + |> ewe.supervised, 36 36 + ) 37 37 + |> supervisor.start 38 38 + 39 39 + process.sleep_forever() 40 40 + } 41 41 + 42 42 + fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 43 43 + case request.path_segments(req) { 44 44 + ["topic", topic] -> handle_topic(req, pubsub, topic) 45 45 + _ -> 46 46 + response.new(404) 47 47 + |> response.set_body(ewe.Empty) 48 48 + } 49 49 + } 50 50 + 51 51 + // Websocket 52 52 + // ----------------------------------------------------------------------------- 53 53 + 54 54 + type WebsocketState { 55 55 + WebsocketState( 56 56 + pubsub: Subject(PubSubMessage), 57 57 + topic: String, 58 58 + client: Subject(Broadcast), 59 59 + ) 60 60 + } 61 61 + 62 62 + type Broadcast { 63 63 + Text(String) 64 64 + Bytes(BitArray) 65 65 + } 66 66 + 67 67 + fn handle_topic(req: Request, pubsub: Subject(PubSubMessage), topic: String) { 68 68 + // Upgrade the HTTP connection to WebSocket. Unlike SSE, WebSocket is 69 69 + // bidirectional - both client and server can send messages at any time. 70 70 + // 71 71 + ewe.upgrade_websocket( 72 72 + req, 73 73 + // Initialize the WebSocket connection. The selector allows receiving 74 74 + // messages from both the WebSocket and the pubsub system. 75 75 + // 76 76 + on_init: fn(_conn, selector) { 77 77 + logging.log( 78 78 + logging.Info, 79 79 + "WebSocket connection opened: " <> pid_to_string(process.self()), 80 80 + ) 81 81 + 82 82 + let client = process.new_subject() 83 83 + process.send(pubsub, Subscribe(topic:, client:)) 84 84 + 85 85 + let state = WebsocketState(pubsub:, topic:, client:) 86 86 + // Add the client subject to the selector to receive broadcast messages. 87 87 + // 88 88 + let selector = process.select(selector, client) 89 89 + 90 90 + #(state, selector) 91 91 + }, 92 92 + handler: handle_websocket_message, 93 93 + on_close: fn(_conn, state) { 94 94 + let assert Ok(pid) = process.subject_owner(state.client) 95 95 + logging.log( 96 96 + logging.Info, 97 97 + "WebSocket connection closed: " <> pid_to_string(pid), 98 98 + ) 99 99 + 100 100 + process.send(pubsub, Unsubscribe(state.topic, state.client)) 101 101 + }, 102 102 + ) 103 103 + } 104 104 + 105 105 + // Handle three types of messages: text from client, binary from client, 106 106 + // and broadcast messages from the pubsub system. 107 107 + // 108 108 + fn handle_websocket_message( 109 109 + conn: ewe.WebsocketConnection, 110 110 + state: WebsocketState, 111 111 + msg: ewe.WebsocketMessage(Broadcast), 112 112 + ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { 113 113 + case msg { 114 114 + // Text message from the client - broadcast to all subscribers. 115 115 + // 116 116 + ewe.Text(text) -> { 117 117 + process.send(state.pubsub, Publish(state.topic, Text(text))) 118 118 + ewe.websocket_continue(state) 119 119 + } 120 120 + 121 121 + // Binary message from the client - broadcast to all subscribers. 122 122 + // 123 123 + ewe.Binary(binary) -> { 124 124 + process.send(state.pubsub, Publish(state.topic, Bytes(binary))) 125 125 + ewe.websocket_continue(state) 126 126 + } 127 127 + 128 128 + // User message from the pubsub - forward to this client. 129 129 + // 130 130 + ewe.User(message) -> { 131 131 + let assert Ok(_) = case message { 132 132 + Text(text) -> ewe.send_text_frame(conn, text) 133 133 + Bytes(binary) -> ewe.send_binary_frame(conn, binary) 134 134 + } 135 135 + 136 136 + ewe.websocket_continue(state) 137 137 + } 138 138 + } 139 139 + } 140 140 + 141 141 + // PubSub 142 142 + // ----------------------------------------------------------------------------- 143 143 + 144 144 + type PubSubMessage { 145 145 + Subscribe(topic: String, client: Subject(Broadcast)) 146 146 + Publish(topic: String, message: Broadcast) 147 147 + Unsubscribe(topic: String, client: Subject(Broadcast)) 148 148 + } 149 149 + 150 150 + fn pubsub_worker( 151 151 + named: Name(PubSubMessage), 152 152 + ) -> ChildSpecification(Subject(PubSubMessage)) { 153 153 + let pubsub = 154 154 + dict.new() 155 155 + |> actor.new 156 156 + |> actor.on_message(handle_pubsub_message) 157 157 + |> actor.named(named) 158 158 + 159 159 + supervision.worker(fn() { 160 160 + logging.log(logging.Info, "Starting pubsub worker") 161 161 + actor.start(pubsub) 162 162 + }) 163 163 + } 164 164 + 165 165 + fn handle_pubsub_message(state, message) { 166 166 + case message { 167 167 + Subscribe(topic:, client:) -> { 168 168 + let new_state = 169 169 + dict.upsert(in: state, update: topic, with: fn(clients) { 170 170 + case clients { 171 171 + Some(clients) -> [client, ..clients] 172 172 + None -> { 173 173 + logging.log(logging.Info, "Creating topic " <> topic) 174 174 + [client] 175 175 + } 176 176 + } 177 177 + }) 178 178 + 179 179 + let assert Ok(pid) = process.subject_owner(client) 180 180 + logging.log( 181 181 + logging.Info, 182 182 + "Subscribing client " <> pid_to_string(pid) <> " to topic " <> topic, 183 183 + ) 184 184 + 185 185 + actor.continue(new_state) 186 186 + } 187 187 + Publish(topic:, message:) -> { 188 188 + case message { 189 189 + Text(text) -> 190 190 + logging.log( 191 191 + logging.Info, 192 192 + "Publishing text message `" <> text <> "` to topic " <> topic, 193 193 + ) 194 194 + Bytes(binary) -> 195 195 + logging.log( 196 196 + logging.Info, 197 197 + "Publishing binary message `" 198 198 + <> string.inspect(binary) 199 199 + <> "` to topic " 200 200 + <> topic, 201 201 + ) 202 202 + } 203 203 + 204 204 + case dict.get(state, topic) { 205 205 + Ok(clients) -> list.each(clients, actor.send(_, message)) 206 206 + Error(_) -> Nil 207 207 + } 208 208 + 209 209 + actor.continue(state) 210 210 + } 211 211 + Unsubscribe(topic:, client:) -> { 212 212 + let assert Ok(pid) = process.subject_owner(client) 213 213 + logging.log( 214 214 + logging.Info, 215 215 + "Unsubscribing client " <> pid_to_string(pid) <> " from topic " <> topic, 216 216 + ) 217 217 + 218 218 + let new_state = case dict.get(state, topic) { 219 219 + Ok([_]) | Ok([]) -> { 220 220 + logging.log(logging.Info, "Dropping topic " <> topic) 221 221 + dict.drop(state, [topic]) 222 222 + } 223 223 + Ok(clients) -> { 224 224 + list.filter(clients, fn(c) { c != client }) 225 225 + |> dict.insert(state, topic, _) 226 226 + } 227 227 + Error(_) -> state 228 228 + } 229 229 + 230 230 + actor.continue(new_state) 231 231 + } 232 232 + } 233 233 + } 234 234 + 235 235 + // Utilities 236 236 + // ----------------------------------------------------------------------------- 237 237 + 238 238 + fn pid_to_string(pid: Pid) -> String { 239 239 + charlist.to_string(pid_to_list(pid)) 240 240 + } 241 241 + 242 242 + @external(erlang, "erlang", "pid_to_list") 243 243 + fn pid_to_list(pid: Pid) -> Charlist

examples/sse/.gitignore examples/.gitignore

Reviewed

-12

examples/sse/README.md

Reviewed

··· 1 1 - # Example: Real-Time Gleam Chat 2 2 - 3 3 - > [!NOTE] 4 4 - > This example is taken from article [Building a real-time chat in Gleam](https://gautier.dev/articles/real-time-gleam-chat). 5 5 - 6 6 - 1. Start the server with the following command: 7 7 - 8 8 - ```bash 9 9 - gleam run 10 10 - ``` 11 11 - 12 12 - 2. Open `http://127.0.0.1:8080/` in the browser

-13

examples/sse/gleam.toml

Reviewed

··· 1 1 - name = "app" 2 2 - version = "1.0.0" 3 3 - 4 4 - [dependencies] 5 5 - gleam_stdlib = ">= 0.44.0 and < 2.0.0" 6 6 - gleam_erlang = ">= 1.3.0 and < 2.0.0" 7 7 - gleam_http = ">= 4.2.0 and < 5.0.0" 8 8 - gleam_otp = ">= 1.1.0 and < 2.0.0" 9 9 - logging = ">= 1.3.0 and < 2.0.0" 10 10 - ewe = { path = "../../" } 11 11 - 12 12 - [dev-dependencies] 13 13 - gleeunit = ">= 1.0.0 and < 2.0.0"

+7 -6

examples/sse/manifest.toml examples/manifest.toml

Reviewed

··· 3 3 4 4 packages = [ 5 5 { name = "compresso", version = "0.1.0", build_tools = ["gleam"], requirements = ["exception", "gleam_erlang", "gleam_stdlib", "gleam_yielder", "logging"], otp_app = "compresso", source = "hex", outer_checksum = "8BE29A1EDA42F70826ED148EAE40C46BB3FC18E78FE472663DB01DD4A38172D4" }, 6 6 - { name = "ewe", version = "2.1.0", build_tools = ["gleam"], requirements = ["compresso", "gleam_erlang", "gleam_http", "gleam_otp", "gleam_stdlib", "glisten", "logging", "websocks"], source = "local", path = "../.." }, 6 6 + { name = "ewe", version = "2.1.2", build_tools = ["gleam"], requirements = ["compresso", "exception", "gleam_erlang", "gleam_http", "gleam_otp", "gleam_stdlib", "glisten", "logging", "websocks"], source = "local", path = ".." }, 7 7 { name = "exception", version = "2.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "exception", source = "hex", outer_checksum = "329D269D5C2A314F7364BD2711372B6F2C58FA6F39981572E5CA68624D291F8C" }, 8 8 { name = "gleam_crypto", version = "1.5.1", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_crypto", source = "hex", outer_checksum = "50774BAFFF1144E7872814C566C5D653D83A3EBF23ACC3156B757A1B6819086E" }, 9 9 { name = "gleam_erlang", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_erlang", source = "hex", outer_checksum = "1124AD3AA21143E5AF0FC5CF3D9529F6DB8CA03E43A55711B60B6B7B3874375C" }, 10 10 { name = "gleam_http", version = "4.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_http", source = "hex", outer_checksum = "82EA6A717C842456188C190AFB372665EA56CE13D8559BF3B1DD9E40F619EE0C" }, 11 11 { name = "gleam_otp", version = "1.2.0", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_stdlib"], otp_app = "gleam_otp", source = "hex", outer_checksum = "BA6A294E295E428EC1562DC1C11EA7530DCB981E8359134BEABC8493B7B2258E" }, 12 12 - { name = "gleam_stdlib", version = "0.65.0", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "7C69C71D8C493AE11A5184828A77110EB05A7786EBF8B25B36A72F879C3EE107" }, 12 12 + { name = "gleam_stdlib", version = "0.68.1", build_tools = ["gleam"], requirements = [], otp_app = "gleam_stdlib", source = "hex", outer_checksum = "F7FAEBD8EF260664E86A46C8DBA23508D1D11BB3BCC6EE1B89B3BC3E5C83FF1E" }, 13 13 { name = "gleam_yielder", version = "1.1.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleam_yielder", source = "hex", outer_checksum = "8E4E4ECFA7982859F430C57F549200C7749823C106759F4A19A78AEA6687717A" }, 14 14 { name = "gleeunit", version = "1.9.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "gleeunit", source = "hex", outer_checksum = "DA9553CE58B67924B3C631F96FE3370C49EB6D6DC6B384EC4862CC4AAA718F3C" }, 15 15 - { 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" }, 15 15 + { name = "glisten", version = "8.0.3", build_tools = ["gleam"], requirements = ["gleam_erlang", "gleam_otp", "gleam_stdlib", "logging", "telemetry"], otp_app = "glisten", source = "hex", outer_checksum = "86B838196592D9EBDE7A1D2369AE3A51E568F7DD2D168706C463C42D17B95312" }, 16 16 { name = "logging", version = "1.3.0", build_tools = ["gleam"], requirements = ["gleam_stdlib"], otp_app = "logging", source = "hex", outer_checksum = "1098FBF10B54B44C2C7FDF0B01C1253CAFACDACABEFB4B0D027803246753E06D" }, 17 17 { name = "telemetry", version = "1.3.0", build_tools = ["rebar3"], requirements = [], otp_app = "telemetry", source = "hex", outer_checksum = "7015FC8919DBE63764F4B4B87A95B7C0996BD539E0D499BE6EC9D7F3875B79E6" }, 18 18 { name = "websocks", version = "2.0.0", build_tools = ["gleam"], requirements = ["gleam_crypto", "gleam_erlang", "gleam_stdlib"], otp_app = "websocks", source = "hex", outer_checksum = "A13BF89A8AC63C478C0E9FE502C81A6CBB0513F427FA5C1DFD96383BF46D501D" }, 19 19 ] 20 20 21 21 [requirements] 22 22 - ewe = { path = "../../" } 22 22 + ewe = { path = "../" } 23 23 + gleam_crypto = { version = ">= 1.5.1 and < 2.0.0" } 23 24 gleam_erlang = { version = ">= 1.3.0 and < 2.0.0" } 24 24 - gleam_http = { version = ">= 4.2.0 and < 5.0.0" } 25 25 - gleam_otp = { version = ">= 1.1.0 and < 2.0.0" } 25 25 + gleam_http = { version = ">= 4.3.0 and < 5.0.0" } 26 26 + gleam_otp = { version = ">= 1.2.0 and < 2.0.0" } 26 27 gleam_stdlib = { version = ">= 0.44.0 and < 2.0.0" } 27 28 gleeunit = { version = ">= 1.0.0 and < 2.0.0" } 28 29 logging = { version = ">= 1.3.0 and < 2.0.0" }

examples/sse/priv/index.html examples/priv/index.html

Reviewed

+104 -84

examples/sse/src/app.gleam examples/src/sse.gleam

Reviewed

··· 1 1 + import ewe 1 2 import gleam/bit_array 2 3 import gleam/erlang/charlist 3 4 import gleam/erlang/process.{type Name, type Pid, type Subject} ··· 6 7 import gleam/list 7 8 import gleam/option.{None} 8 9 import gleam/otp/actor 9 9 - import gleam/otp/static_supervisor.{type Supervisor} as supervisor 10 10 + import gleam/otp/static_supervisor as supervisor 10 11 import gleam/otp/supervision.{type ChildSpecification} 11 12 import gleam/string 12 13 import logging 13 14 14 14 - import ewe 15 15 - 16 16 - type PubSubMessage { 17 17 - Subscribe(client: Subject(String)) 18 18 - Unsubscribe(client: Subject(String)) 19 19 - Publish(String) 20 20 - } 21 21 - 22 22 - // If 8080 is already in use, you can change the port here 23 23 - const port = 8080 24 24 - 25 25 - fn handle_pubsub_message(clients: List(Subject(String)), message: PubSubMessage) { 26 26 - case message { 27 27 - Subscribe(client) -> { 28 28 - let assert Ok(pid) = process.subject_owner(client) 29 29 - 30 30 - logging.log(logging.Info, "Client " <> pid_to_string(pid) <> " connected") 31 31 - 32 32 - actor.continue([client, ..clients]) 33 33 - } 34 34 - 35 35 - Unsubscribe(client) -> { 36 36 - let assert Ok(pid) = process.subject_owner(client) 37 37 - 38 38 - { "Client " <> pid_to_string(pid) <> " disconnected" } 39 39 - |> logging.log(logging.Info, _) 40 40 - 41 41 - list.filter(clients, fn(subscribed) { subscribed != client }) 42 42 - |> actor.continue() 43 43 - } 44 44 - 45 45 - Publish(message) -> { 46 46 - let pids = 47 47 - list.fold(over: clients, from: [], with: fn(acc, client) { 48 48 - let assert Ok(pid) = process.subject_owner(client) 49 49 - let _ = process.send(client, message) 50 50 - 51 51 - [pid_to_string(pid), ..acc] 52 52 - }) 53 53 - |> string.join(", ") 54 54 - 55 55 - { "Sent message `" <> message <> "` to clients: " <> pids } 56 56 - |> logging.log(logging.Info, _) 57 57 - 58 58 - actor.continue(clients) 59 59 - } 60 60 - } 61 61 - } 62 62 - 63 15 pub fn main() -> Nil { 64 16 logging.configure() 65 17 66 66 - let pubsub = process.new_name("pubsub") 18 18 + // Create a named pubsub process for broadcasting messages to all connected 19 19 + // SSE clients. 20 20 + // 21 21 + let pubsub_name = process.new_name("pubsub") 22 22 + let pubsub = process.named_subject(pubsub_name) 67 23 24 24 + // Use a supervisor to manage both the pubsub worker and web server. 25 25 + // OneForAll means if either crashes, both will restart together. 26 26 + // 68 27 let assert Ok(_) = 69 28 supervisor.new(supervisor.OneForAll) 70 70 - |> supervisor.add(pubsub_worker(pubsub)) 71 71 - |> supervisor.add(web_server(pubsub)) 72 72 - |> supervisor.start() 29 29 + |> supervisor.add(pubsub_worker(pubsub_name)) 30 30 + |> supervisor.add( 31 31 + ewe.new(handler(_, pubsub)) 32 32 + |> ewe.listening(port: 8080) 33 33 + |> ewe.bind_all 34 34 + // Use ewe.supervised instead of ewe.start to run under supervision. 35 35 + // 36 36 + |> ewe.supervised, 37 37 + ) 38 38 + |> supervisor.start 73 39 74 40 process.sleep_forever() 75 41 } 76 42 77 77 - fn pubsub_worker( 78 78 - named: Name(PubSubMessage), 79 79 - ) -> ChildSpecification(Subject(PubSubMessage)) { 80 80 - supervision.worker(fn() { 81 81 - actor.new([]) 82 82 - |> actor.on_message(handle_pubsub_message) 83 83 - |> actor.named(named) 84 84 - |> actor.start() 85 85 - }) 86 86 - } 87 87 - 88 88 - fn web_server(pubsub: Name(PubSubMessage)) -> ChildSpecification(Supervisor) { 89 89 - ewe.new(handle_request(_, process.named_subject(pubsub))) 90 90 - |> ewe.listening(port: port) 91 91 - |> ewe.bind_all() 92 92 - |> ewe.supervised() 93 93 - } 94 94 - 95 95 - fn empty_response(status: Int) -> ewe.Response { 96 96 - response.new(status) |> response.set_body(ewe.Empty) 97 97 - } 43 43 + // SSE 44 44 + // ----------------------------------------------------------------------------- 98 45 99 99 - fn handle_request( 100 100 - req: ewe.Request, 101 101 - pubsub: Subject(PubSubMessage), 102 102 - ) -> ewe.Response { 46 46 + fn handler(req: ewe.Request, pubsub: Subject(PubSubMessage)) -> ewe.Response { 103 47 case req.method, req.path { 104 104 - // On `GET /`, serve the index.html file 48 48 + // Serve the demo HTML page. 49 49 + // 105 50 http.Get, "/" -> { 106 51 case ewe.file("priv/index.html", offset: None, limit: None) { 107 52 Ok(file) -> { ··· 113 58 } 114 59 } 115 60 116 116 - // On `GET /sse`, start a Server-Sent Events connection 61 61 + // Establish a Server-Sent Events connection. SSE is a one-way channel 62 62 + // from server to client. The connection stays open and the server can 63 63 + // push events at any time. 64 64 + // 117 65 http.Get, "/sse" -> 118 66 ewe.sse( 119 67 req, 68 68 + // Initialize the connection and subscribe this client to the pubsub. 69 69 + // 120 70 on_init: fn(client) { 121 71 process.send(pubsub, Subscribe(client)) 122 72 123 73 client 124 74 }, 75 75 + // Handle messages from the pubsub and send them as SSE events. 76 76 + // 125 77 handler: fn(conn, client, message) { 126 78 case ewe.send_event(conn, ewe.event(message)) { 127 79 Ok(Nil) -> ewe.sse_continue(client) 128 80 Error(_) -> ewe.sse_stop() 129 81 } 130 82 }, 83 83 + // Clean up when the client disconnects. 84 84 + // 131 85 on_close: fn(_conn, client) { 132 86 process.send(pubsub, Unsubscribe(client)) 133 87 }, 134 88 ) 135 89 136 136 - // On `POST /post`, publish the body to the pubsub 90 90 + // Accept messages via POST and broadcast them to all SSE clients. 91 91 + // 137 92 http.Post, "/post" -> { 138 138 - // `128` is the limit of the body size in the frontend (see L66 of 139 139 - // `index.html`) 93 93 + // Limit matches the frontend restriction (see index.html). 94 94 + // 140 95 case ewe.read_body(req, 128) { 141 96 Ok(req) -> { 142 97 case bit_array.to_string(req.body) { ··· 155 110 _, _ -> empty_response(404) 156 111 } 157 112 } 113 113 + 114 114 + fn empty_response(status: Int) -> ewe.Response { 115 115 + response.new(status) |> response.set_body(ewe.Empty) 116 116 + } 117 117 + 118 118 + // PubSub 119 119 + // ----------------------------------------------------------------------------- 120 120 + 121 121 + type PubSubMessage { 122 122 + Subscribe(client: Subject(String)) 123 123 + Unsubscribe(client: Subject(String)) 124 124 + Publish(String) 125 125 + } 126 126 + 127 127 + fn pubsub_worker( 128 128 + named: Name(PubSubMessage), 129 129 + ) -> ChildSpecification(Subject(PubSubMessage)) { 130 130 + supervision.worker(fn() { 131 131 + actor.new([]) 132 132 + |> actor.on_message(handle_pubsub_message) 133 133 + |> actor.named(named) 134 134 + |> actor.start() 135 135 + }) 136 136 + } 137 137 + 138 138 + fn handle_pubsub_message(clients: List(Subject(String)), message: PubSubMessage) { 139 139 + case message { 140 140 + Subscribe(client) -> { 141 141 + let assert Ok(pid) = process.subject_owner(client) 142 142 + 143 143 + logging.log(logging.Info, "Client " <> pid_to_string(pid) <> " connected") 144 144 + 145 145 + actor.continue([client, ..clients]) 146 146 + } 147 147 + 148 148 + Unsubscribe(client) -> { 149 149 + let assert Ok(pid) = process.subject_owner(client) 150 150 + 151 151 + { "Client " <> pid_to_string(pid) <> " disconnected" } 152 152 + |> logging.log(logging.Info, _) 153 153 + 154 154 + list.filter(clients, fn(subscribed) { subscribed != client }) 155 155 + |> actor.continue() 156 156 + } 157 157 + 158 158 + Publish(message) -> { 159 159 + let pids = 160 160 + list.fold(over: clients, from: [], with: fn(acc, client) { 161 161 + let assert Ok(pid) = process.subject_owner(client) 162 162 + let _ = process.send(client, message) 163 163 + 164 164 + [pid_to_string(pid), ..acc] 165 165 + }) 166 166 + |> string.join(", ") 167 167 + 168 168 + { "Sent message `" <> message <> "` to clients: " <> pids } 169 169 + |> logging.log(logging.Info, _) 170 170 + 171 171 + actor.continue(clients) 172 172 + } 173 173 + } 174 174 + } 175 175 + 176 176 + // Utilities 177 177 + // ----------------------------------------------------------------------------- 158 178 159 179 fn pid_to_string(pid: Pid) -> String { 160 180 pid_to_list(pid)

+1 -1

examples/sse/test/app_test.gleam examples/test/examples_test.gleam

Reviewed

··· 1 1 import gleeunit 2 2 3 3 - pub fn main() -> Nil { 3 3 + pub fn main() { 4 4 gleeunit.main() 5 5 }

-35

test/preview/file_serving.gleam

Reviewed

··· 1 1 - import gleam/erlang/process 2 2 - import logging 3 3 - 4 4 - import gleam/http/response 5 5 - import gleam/option.{None} 6 6 - 7 7 - import ewe.{type Response} 8 8 - 9 9 - fn serve_file(path: String) -> Response { 10 10 - case ewe.file("public" <> path, offset: None, limit: None) { 11 11 - Ok(file) -> { 12 12 - response.new(200) 13 13 - |> response.set_header("content-type", "application/octet-stream") 14 14 - |> response.set_body(file) 15 15 - } 16 16 - Error(_) -> { 17 17 - response.new(404) 18 18 - |> response.set_header("content-type", "text/plain; charset=utf-8") 19 19 - |> response.set_body(ewe.TextData("File not found")) 20 20 - } 21 21 - } 22 22 - } 23 23 - 24 24 - pub fn main() { 25 25 - logging.configure() 26 26 - logging.set_level(logging.Info) 27 27 - 28 28 - let assert Ok(_) = 29 29 - ewe.new(fn(req) { serve_file(req.path) }) 30 30 - |> ewe.bind_all() 31 31 - |> ewe.listening(port: 8080) 32 32 - |> ewe.start() 33 33 - 34 34 - process.sleep_forever() 35 35 - }

+23 -17

test/preview/getting_request_body.gleam examples/src/reading_body.gleam

Reviewed

··· 1 1 + import ewe.{type Request, type Response} 1 2 import gleam/erlang/process 2 2 - import logging 3 3 - 4 3 import gleam/http/request 5 4 import gleam/http/response 6 5 import gleam/result 6 6 + import logging 7 7 8 8 - import ewe.{type Request, type Response} 8 8 + pub fn main() { 9 9 + logging.configure() 10 10 + logging.set_level(logging.Info) 9 11 10 10 - fn handle_echo(req: Request) -> Response { 12 12 + // An echo server that reads the request body and sends it back. 13 13 + // 14 14 + let assert Ok(_) = 15 15 + ewe.new(handler) 16 16 + |> ewe.bind_all 17 17 + |> ewe.listening(port: 8080) 18 18 + |> ewe.start 19 19 + 20 20 + process.sleep_forever() 21 21 + } 22 22 + 23 23 + fn handler(req: Request) -> Response { 24 24 + // Preserve the original content-type from the request to send back. 25 25 + // 11 26 let content_type = 12 27 request.get_header(req, "content-type") 13 28 |> result.unwrap("application/octet-stream") 14 29 30 30 + // Read the entire request body into memory with a 10KB limit. This blocks 31 31 + // until the full body is received. For large uploads or streaming data, 32 32 + // use ewe.stream_body() instead. 33 33 + // 15 34 case ewe.read_body(req, 10_240) { 16 35 Ok(req) -> 17 36 response.new(200) ··· 27 46 |> response.set_body(ewe.TextData("Invalid request")) 28 47 } 29 48 } 30 30 - 31 31 - pub fn main() { 32 32 - logging.configure() 33 33 - logging.set_level(logging.Info) 34 34 - 35 35 - let assert Ok(_) = 36 36 - ewe.new(handle_echo) 37 37 - |> ewe.bind_all() 38 38 - |> ewe.listening(port: 8080) 39 39 - |> ewe.start() 40 40 - 41 41 - process.sleep_forever() 42 42 - }

-24

test/preview/quick_start.gleam

Reviewed

··· 1 1 - import gleam/erlang/process 2 2 - import gleam/http/response 3 3 - import logging 4 4 - 5 5 - import ewe.{type Request, type Response} 6 6 - 7 7 - fn handler(_req: Request) -> Response { 8 8 - response.new(200) 9 9 - |> response.set_header("content-type", "text/plain; charset=utf-8") 10 10 - |> response.set_body(ewe.TextData("Hello, World!")) 11 11 - } 12 12 - 13 13 - pub fn main() { 14 14 - logging.configure() 15 15 - logging.set_level(logging.Info) 16 16 - 17 17 - let assert Ok(_) = 18 18 - ewe.new(handler) 19 19 - |> ewe.bind_all() 20 20 - |> ewe.listening(port: 8080) 21 21 - |> ewe.start() 22 22 - 23 23 - process.sleep_forever() 24 24 - }

-46

test/preview/sending_response.gleam

Reviewed

··· 1 1 - import gleam/erlang/process 2 2 - import logging 3 3 - 4 4 - import gleam/crypto 5 5 - import gleam/http/request.{type Request} 6 6 - import gleam/http/response.{type Response} 7 7 - import gleam/int 8 8 - import gleam/result 9 9 - 10 10 - import ewe.{type Connection, type ResponseBody} 11 11 - 12 12 - fn handler(req: Request(Connection)) -> Response(ResponseBody) { 13 13 - case request.path_segments(req) { 14 14 - ["hello", name] -> { 15 15 - response.new(200) 16 16 - |> response.set_header("content-type", "text/plain; charset=utf-8") 17 17 - |> response.set_body(ewe.TextData("Hello, " <> name <> "!")) 18 18 - } 19 19 - ["bytes", amount] -> { 20 20 - let random_bytes = 21 21 - int.parse(amount) 22 22 - |> result.unwrap(0) 23 23 - |> crypto.strong_random_bytes() 24 24 - 25 25 - response.new(200) 26 26 - |> response.set_header("content-type", "application/octet-stream") 27 27 - |> response.set_body(ewe.BitsData(random_bytes)) 28 28 - } 29 29 - _ -> 30 30 - response.new(404) 31 31 - |> response.set_body(ewe.Empty) 32 32 - } 33 33 - } 34 34 - 35 35 - pub fn main() { 36 36 - logging.configure() 37 37 - logging.set_level(logging.Info) 38 38 - 39 39 - let assert Ok(_) = 40 40 - ewe.new(handler) 41 41 - |> ewe.bind_all() 42 42 - |> ewe.listening(port: 8080) 43 43 - |> ewe.start() 44 44 - 45 45 - process.sleep_forever() 46 46 - }

-140

test/preview/sse.gleam

Reviewed

··· 1 1 - import gleam/otp/actor 2 2 - import gleam/otp/static_supervisor as supervisor 3 3 - import gleam/otp/supervision.{type ChildSpecification} 4 4 - 5 5 - import gleam/bit_array 6 6 - import gleam/erlang/charlist.{type Charlist} 7 7 - import gleam/erlang/process.{type Name, type Pid, type Subject} 8 8 - import gleam/http/request 9 9 - import gleam/http/response 10 10 - import gleam/list 11 11 - import gleam/string 12 12 - import logging 13 13 - 14 14 - import ewe.{type Request, type Response} 15 15 - 16 16 - type PubSubMessage { 17 17 - Subscribe(client: Subject(String)) 18 18 - Unsubscribe(client: Subject(String)) 19 19 - Publish(String) 20 20 - } 21 21 - 22 22 - fn pubsub_worker( 23 23 - named: Name(PubSubMessage), 24 24 - ) -> ChildSpecification(Subject(PubSubMessage)) { 25 25 - let pubsub = 26 26 - actor.new([]) 27 27 - |> actor.on_message(fn(clients, message) { 28 28 - case message { 29 29 - Subscribe(client) -> { 30 30 - let assert Ok(pid) = process.subject_owner(client) 31 31 - 32 32 - logging.log( 33 33 - logging.Info, 34 34 - "Client " <> pid_to_string(pid) <> " subscribed", 35 35 - ) 36 36 - 37 37 - actor.continue([client, ..clients]) 38 38 - } 39 39 - 40 40 - Unsubscribe(client) -> { 41 41 - let assert Ok(pid) = process.subject_owner(client) 42 42 - 43 43 - { "Client " <> pid_to_string(pid) <> " unsubscribed" } 44 44 - |> logging.log(logging.Info, _) 45 45 - 46 46 - list.filter(clients, fn(subscribed) { subscribed != client }) 47 47 - |> actor.continue() 48 48 - } 49 49 - 50 50 - Publish(message) -> { 51 51 - let pids = 52 52 - list.fold(over: clients, from: [], with: fn(acc, client) { 53 53 - let assert Ok(pid) = process.subject_owner(client) 54 54 - let _ = process.send(client, message) 55 55 - 56 56 - [pid_to_string(pid), ..acc] 57 57 - }) 58 58 - |> string.join(", ") 59 59 - 60 60 - { "Publishing message `" <> message <> "` to clients: " <> pids } 61 61 - |> logging.log(logging.Info, _) 62 62 - 63 63 - actor.continue(clients) 64 64 - } 65 65 - } 66 66 - }) 67 67 - |> actor.named(named) 68 68 - 69 69 - supervision.worker(fn() { actor.start(pubsub) }) 70 70 - } 71 71 - 72 72 - fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 73 73 - case request.path_segments(req) { 74 74 - ["sse"] -> 75 75 - ewe.sse( 76 76 - req, 77 77 - on_init: fn(client) { 78 78 - process.send(pubsub, Subscribe(client)) 79 79 - logging.log( 80 80 - logging.Info, 81 81 - "SSE connection opened: " <> pid_to_string(process.self()), 82 82 - ) 83 83 - 84 84 - client 85 85 - }, 86 86 - handler: fn(conn, client, message) { 87 87 - case ewe.send_event(conn, ewe.event(message)) { 88 88 - Ok(Nil) -> ewe.sse_continue(client) 89 89 - Error(_) -> ewe.sse_stop() 90 90 - } 91 91 - }, 92 92 - on_close: fn(_conn, client) { 93 93 - process.send(pubsub, Unsubscribe(client)) 94 94 - logging.log( 95 95 - logging.Info, 96 96 - "SSE connection closed: " <> pid_to_string(process.self()), 97 97 - ) 98 98 - }, 99 99 - ) 100 100 - ["publish"] -> { 101 101 - case ewe.read_body(req, 1024) { 102 102 - Ok(req) -> { 103 103 - let assert Ok(text) = bit_array.to_string(req.body) 104 104 - process.send(pubsub, Publish(text)) 105 105 - response.new(200) |> response.set_body(ewe.Empty) 106 106 - } 107 107 - Error(_) -> response.new(400) |> response.set_body(ewe.Empty) 108 108 - } 109 109 - } 110 110 - _ -> response.new(404) |> response.set_body(ewe.Empty) 111 111 - } 112 112 - } 113 113 - 114 114 - pub fn main() { 115 115 - logging.configure() 116 116 - logging.set_level(logging.Info) 117 117 - 118 118 - let pubsub_name = process.new_name("pubsub") 119 119 - let pubsub = process.named_subject(pubsub_name) 120 120 - 121 121 - let assert Ok(_) = 122 122 - supervisor.new(supervisor.OneForAll) 123 123 - |> supervisor.add(pubsub_worker(pubsub_name)) 124 124 - |> supervisor.add( 125 125 - ewe.new(handler(_, pubsub)) 126 126 - |> ewe.bind_all() 127 127 - |> ewe.listening(port: 8080) 128 128 - |> ewe.supervised(), 129 129 - ) 130 130 - |> supervisor.start() 131 131 - 132 132 - process.sleep_forever() 133 133 - } 134 134 - 135 135 - fn pid_to_string(pid: Pid) -> String { 136 136 - charlist.to_string(pid_to_list(pid)) 137 137 - } 138 138 - 139 139 - @external(erlang, "erlang", "pid_to_list") 140 140 - fn pid_to_list(pid: Pid) -> Charlist

+77 -55

test/preview/streaming.gleam examples/src/streaming_body.gleam

Reviewed

··· 1 1 - import gleam/string 2 2 - 1 1 + import ewe.{type Request, type Response} 3 2 import gleam/bit_array 4 3 import gleam/erlang/process.{type Subject} 5 4 import gleam/http/request 6 5 import gleam/http/response 7 6 import gleam/int 8 7 import gleam/result 8 8 + import gleam/string 9 9 import logging 10 10 11 11 - import ewe.{type Request, type Response} 11 11 + pub fn main() { 12 12 + logging.configure() 13 13 + logging.set_level(logging.Info) 14 14 + 15 15 + // A server that streams the request body back as chunked response. 16 16 + // This demonstrates how to handle large uploads. 17 17 + // 18 18 + let assert Ok(_) = 19 19 + ewe.new(handler) 20 20 + |> ewe.bind_all() 21 21 + |> ewe.listening(port: 8080) 22 22 + |> ewe.start() 23 23 + 24 24 + process.sleep_forever() 25 25 + } 26 26 + 27 27 + fn handler(req: Request) -> Response { 28 28 + // Route: /stream/{chunk_size} - controls how many bytes to read at a time. 29 29 + // 30 30 + case request.path_segments(req) { 31 31 + ["stream", chunk_size] -> 32 32 + int.parse(chunk_size) 33 33 + |> result.unwrap(16) 34 34 + |> handle_stream(req, _) 35 35 + _ -> 36 36 + response.new(404) 37 37 + |> response.set_body(ewe.Empty) 38 38 + } 39 39 + } 12 40 13 41 pub type Message { 14 42 Chunk(BitArray) ··· 16 44 BodyError(ewe.BodyError) 17 45 } 18 46 19 19 - fn stream_resource( 20 20 - consumer: ewe.Consumer, 21 21 - subject: Subject(Message), 22 22 - chunk_size: Int, 23 23 - ) -> Nil { 24 24 - process.sleep(int.random(250)) 25 25 - case consumer(chunk_size) { 26 26 - Ok(ewe.Consumed(data, next)) -> { 27 27 - logging.log(logging.Info, { 28 28 - "Consumed " 29 29 - <> int.to_string(bit_array.byte_size(data)) 30 30 - <> " bytes: " 31 31 - <> string.inspect(data) 32 32 - }) 33 33 - 34 34 - process.send(subject, Chunk(data)) 35 35 - stream_resource(next, subject, chunk_size) 36 36 - } 37 37 - Ok(ewe.Done) -> { 38 38 - process.send(subject, Done) 39 39 - } 40 40 - Error(body_error) -> { 41 41 - process.send(subject, BodyError(body_error)) 42 42 - } 43 43 - } 44 44 - } 45 45 - 46 47 fn handle_stream(req: Request, chunk_size: Int) -> Response { 47 48 let content_type = 48 49 request.get_header(req, "content-type") 49 50 |> result.unwrap("application/octet-stream") 50 51 52 52 + // Get a consumer function for streaming the request body. This allows 53 53 + // reading data incrementally. 54 54 + // 51 55 case ewe.stream_body(req) { 52 56 Ok(consumer) -> { 57 57 + // For example purporses, let's set up a chunked response. The response is 58 58 + // sent in chunks as we consume the request body. 59 59 + // 53 60 ewe.chunked_body( 54 61 req, 55 62 response.new(200) |> response.set_header("content-type", content_type), 63 63 + // Spawn a separate process to consume the body and send chunks. 64 64 + // This prevents blocking the handler while reading data. 65 65 + // 56 66 on_init: fn(subject) { 57 57 - process.spawn(fn() { stream_resource(consumer, subject, chunk_size) }) 58 58 - 59 59 - Nil 67 67 + let _pid = 68 68 + fn() { stream_resource(consumer, subject, chunk_size) } 69 69 + |> process.spawn 60 70 }, 61 71 handler: fn(chunked_body, state, message) { 62 72 case message { ··· 82 92 } 83 93 } 84 94 85 85 - fn handler(req: Request) -> Response { 86 86 - case request.path_segments(req) { 87 87 - ["stream", chunk_size] -> 88 88 - int.parse(chunk_size) 89 89 - |> result.unwrap(16) 90 90 - |> handle_stream(req, _) 91 91 - _ -> 92 92 - response.new(404) 93 93 - |> response.set_body(ewe.Empty) 94 94 - } 95 95 - } 96 96 - 97 97 - pub fn main() { 98 98 - logging.configure() 99 99 - logging.set_level(logging.Info) 100 100 - 101 101 - let assert Ok(_) = 102 102 - ewe.new(handler) 103 103 - |> ewe.bind_all() 104 104 - |> ewe.listening(port: 8080) 105 105 - |> ewe.start() 95 95 + // Recursively consume chunks from the request body and send them to the 96 96 + // chunked response handler via the subject. 97 97 + // 98 98 + fn stream_resource( 99 99 + consumer: ewe.Consumer, 100 100 + subject: Subject(Message), 101 101 + chunk_size: Int, 102 102 + ) -> Nil { 103 103 + // Simulating processing delay here... 104 104 + // 105 105 + process.sleep(int.random(250)) 106 106 + // Call the consumer with the chunk size. It returns the next chunk of data 107 107 + // and a new consumer for the remaining body. 108 108 + // 109 109 + case consumer(chunk_size) { 110 110 + Ok(ewe.Consumed(data, next)) -> { 111 111 + logging.log(logging.Info, { 112 112 + "Consumed " 113 113 + <> int.to_string(bit_array.byte_size(data)) 114 114 + <> " bytes: " 115 115 + <> string.inspect(data) 116 116 + }) 106 117 107 107 - process.sleep_forever() 118 118 + process.send(subject, Chunk(data)) 119 119 + // Recursively process the next chunk. 120 120 + // 121 121 + stream_resource(next, subject, chunk_size) 122 122 + } 123 123 + Ok(ewe.Done) -> { 124 124 + process.send(subject, Done) 125 125 + } 126 126 + Error(body_error) -> { 127 127 + process.send(subject, BodyError(body_error)) 128 128 + } 129 129 + } 108 130 }

-210

test/preview/websocket.gleam

Reviewed

··· 1 1 - import gleam/dict 2 2 - import gleam/list 3 3 - import gleam/option.{None, Some} 4 4 - import gleam/otp/actor 5 5 - import gleam/otp/static_supervisor as supervisor 6 6 - import gleam/otp/supervision.{type ChildSpecification} 7 7 - import gleam/string 8 8 - 9 9 - import gleam/erlang/charlist.{type Charlist} 10 10 - import gleam/erlang/process.{type Name, type Pid, type Subject} 11 11 - import gleam/http/request 12 12 - import gleam/http/response 13 13 - import logging 14 14 - 15 15 - import ewe.{type Request, type Response} 16 16 - 17 17 - type PubSubMessage { 18 18 - Subscribe(topic: String, client: Subject(Broadcast)) 19 19 - Publish(topic: String, message: Broadcast) 20 20 - Unsubscribe(topic: String, client: Subject(Broadcast)) 21 21 - } 22 22 - 23 23 - type Broadcast { 24 24 - Text(String) 25 25 - Bytes(BitArray) 26 26 - } 27 27 - 28 28 - fn pubsub_worker( 29 29 - named: Name(PubSubMessage), 30 30 - ) -> ChildSpecification(Subject(PubSubMessage)) { 31 31 - let pubsub = 32 32 - actor.new(dict.new()) 33 33 - |> actor.on_message(fn(state, msg) { 34 34 - case msg { 35 35 - Subscribe(topic:, client:) -> { 36 36 - let new_state = 37 37 - dict.upsert(in: state, update: topic, with: fn(clients) { 38 38 - case clients { 39 39 - Some(clients) -> [client, ..clients] 40 40 - None -> { 41 41 - logging.log(logging.Info, "Creating topic " <> topic) 42 42 - [client] 43 43 - } 44 44 - } 45 45 - }) 46 46 - 47 47 - let assert Ok(pid) = process.subject_owner(client) 48 48 - logging.log( 49 49 - logging.Info, 50 50 - "Subscribing client " <> pid_to_string(pid) <> " to topic " <> topic, 51 51 - ) 52 52 - 53 53 - actor.continue(new_state) 54 54 - } 55 55 - Publish(topic:, message:) -> { 56 56 - case message { 57 57 - Text(text) -> 58 58 - logging.log( 59 59 - logging.Info, 60 60 - "Publishing text message `" <> text <> "` to topic " <> topic, 61 61 - ) 62 62 - Bytes(binary) -> 63 63 - logging.log( 64 64 - logging.Info, 65 65 - "Publishing binary message `" 66 66 - <> string.inspect(binary) 67 67 - <> "` to topic " 68 68 - <> topic, 69 69 - ) 70 70 - } 71 71 - 72 72 - case dict.get(state, topic) { 73 73 - Ok(clients) -> list.each(clients, actor.send(_, message)) 74 74 - Error(_) -> Nil 75 75 - } 76 76 - 77 77 - actor.continue(state) 78 78 - } 79 79 - Unsubscribe(topic:, client:) -> { 80 80 - let assert Ok(pid) = process.subject_owner(client) 81 81 - logging.log( 82 82 - logging.Info, 83 83 - "Unsubscribing client " 84 84 - <> pid_to_string(pid) 85 85 - <> " from topic " 86 86 - <> topic, 87 87 - ) 88 88 - 89 89 - let new_state = case dict.get(state, topic) { 90 90 - Ok([_]) | Ok([]) -> { 91 91 - logging.log(logging.Info, "Dropping topic " <> topic) 92 92 - dict.drop(state, [topic]) 93 93 - } 94 94 - Ok(clients) -> { 95 95 - list.filter(clients, fn(c) { c != client }) 96 96 - |> dict.insert(state, topic, _) 97 97 - } 98 98 - Error(_) -> state 99 99 - } 100 100 - 101 101 - actor.continue(new_state) 102 102 - } 103 103 - } 104 104 - }) 105 105 - |> actor.named(named) 106 106 - 107 107 - supervision.worker(fn() { 108 108 - logging.log(logging.Info, "Starting pubsub worker") 109 109 - actor.start(pubsub) 110 110 - }) 111 111 - } 112 112 - 113 113 - type WebsocketState { 114 114 - WebsocketState( 115 115 - pubsub: Subject(PubSubMessage), 116 116 - topic: String, 117 117 - client: Subject(Broadcast), 118 118 - ) 119 119 - } 120 120 - 121 121 - fn handler(req: Request, pubsub: Subject(PubSubMessage)) -> Response { 122 122 - case request.path_segments(req) { 123 123 - ["topic", topic] -> 124 124 - ewe.upgrade_websocket( 125 125 - req, 126 126 - on_init: fn(_conn, selector) { 127 127 - logging.log( 128 128 - logging.Info, 129 129 - "WebSocket connection opened: " <> pid_to_string(process.self()), 130 130 - ) 131 131 - 132 132 - let client = process.new_subject() 133 133 - process.send(pubsub, Subscribe(topic:, client:)) 134 134 - 135 135 - let state = WebsocketState(pubsub:, topic:, client:) 136 136 - let selector = process.select(selector, client) 137 137 - 138 138 - #(state, selector) 139 139 - }, 140 140 - handler: handle_websocket, 141 141 - on_close: fn(_conn, state) { 142 142 - let assert Ok(pid) = process.subject_owner(state.client) 143 143 - logging.log( 144 144 - logging.Info, 145 145 - "WebSocket connection closed: " <> pid_to_string(pid), 146 146 - ) 147 147 - 148 148 - process.send(pubsub, Unsubscribe(state.topic, state.client)) 149 149 - }, 150 150 - ) 151 151 - _ -> 152 152 - response.new(404) 153 153 - |> response.set_body(ewe.Empty) 154 154 - } 155 155 - } 156 156 - 157 157 - fn handle_websocket( 158 158 - conn: ewe.WebsocketConnection, 159 159 - state: WebsocketState, 160 160 - msg: ewe.WebsocketMessage(Broadcast), 161 161 - ) -> ewe.WebsocketNext(WebsocketState, Broadcast) { 162 162 - case msg { 163 163 - ewe.Text(text) -> { 164 164 - process.send(state.pubsub, Publish(state.topic, Text(text))) 165 165 - ewe.websocket_continue(state) 166 166 - } 167 167 - 168 168 - ewe.Binary(binary) -> { 169 169 - process.send(state.pubsub, Publish(state.topic, Bytes(binary))) 170 170 - ewe.websocket_continue(state) 171 171 - } 172 172 - 173 173 - ewe.User(message) -> { 174 174 - let assert Ok(_) = case message { 175 175 - Text(text) -> ewe.send_text_frame(conn, text) 176 176 - Bytes(binary) -> ewe.send_binary_frame(conn, binary) 177 177 - } 178 178 - 179 179 - ewe.websocket_continue(state) 180 180 - } 181 181 - } 182 182 - } 183 183 - 184 184 - pub fn main() { 185 185 - logging.configure() 186 186 - logging.set_level(logging.Info) 187 187 - 188 188 - let pubsub_name = process.new_name("pubsub") 189 189 - let pubsub = process.named_subject(pubsub_name) 190 190 - 191 191 - let assert Ok(_) = 192 192 - supervisor.new(supervisor.OneForAll) 193 193 - |> supervisor.add(pubsub_worker(pubsub_name)) 194 194 - |> supervisor.add( 195 195 - ewe.new(handler(_, pubsub)) 196 196 - |> ewe.bind_all() 197 197 - |> ewe.listening(port: 8080) 198 198 - |> ewe.supervised(), 199 199 - ) 200 200 - |> supervisor.start() 201 201 - 202 202 - process.sleep_forever() 203 203 - } 204 204 - 205 205 - fn pid_to_string(pid: Pid) -> String { 206 206 - charlist.to_string(pid_to_list(pid)) 207 207 - } 208 208 - 209 209 - @external(erlang, "erlang", "pid_to_list") 210 210 - fn pid_to_list(pid: Pid) -> Charlist