-module(dream_http_client@client). -compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]). -define(FILEPATH, "src/dream_http_client/client.gleam"). -export([method/2, scheme/2, host/2, port/2, path/2, 'query'/2, headers/2, body/2, recorder/2, timeout/2, add_header/3, get_method/1, get_scheme/1, get_host/1, get_port/1, get_path/1, get_query/1, get_headers/1, get_body/1, get_timeout/1, get_recorder/1, stream_yielder/1, send/1, stream_messages/1, select_stream_messages/2, cancel_stream/1]). -export_type([client_request/0, request_id/0, stream_message/0, yielder_state/0]). -if(?OTP_RELEASE >= 27). -define(MODULEDOC(Str), -moduledoc(Str)). -define(DOC(Str), -doc(Str)). -else. -define(MODULEDOC(Str), -compile([])). -define(DOC(Str), -compile([])). -endif. ?MODULEDOC( " Type-safe HTTP client with streaming support\n" "\n" " Gleam doesn't have a built-in HTTPS client, so this module wraps Erlang's battle-hardened\n" " `httpc`. Use this for calling external APIs, downloading files, streaming AI responses,\n" " or building OTP-compatible services with concurrent HTTP streams.\n" "\n" " ## Quick Example - Blocking Request\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, path, add_header, send}\n" "\n" " pub fn call_api() {\n" " let result = client.new\n" " |> host(\"api.example.com\")\n" " |> path(\"/users/123\")\n" " |> add_header(\"Authorization\", \"Bearer \" <> token)\n" " |> send()\n" "\n" " case result {\n" " Ok(body) -> decode_json(body)\n" " Error(msg) -> handle_error(msg)\n" " }\n" " }\n" " ```\n" "\n" " ## Execution Modes\n" "\n" " This module provides three ways to execute HTTP requests:\n" "\n" " ### 1. Blocking - `client.send()`\n" "\n" " Get the complete response at once. Perfect for:\n" " - JSON API calls\n" " - Small files or documents\n" " - Any case where you need the full response before processing\n" "\n" " ### 2. Yielder Streaming - `client.stream_yielder()`\n" "\n" " Get a `yielder.Yielder` that produces chunks sequentially. Perfect for:\n" " - AI/LLM inference endpoints (streaming tokens)\n" " - Simple file downloads\n" " - Scripts or one-off operations\n" "\n" " **Note:** This is a pull-based synchronous API. It blocks the calling process\n" " while waiting for chunks, making it unsuitable for OTP actors that need to\n" " handle multiple concurrent operations.\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, path, stream_yielder}\n" " import gleam/yielder.{each}\n" " import gleam/bytes_tree.{to_string}\n" " import gleam/io.{print, println_error}\n" "\n" " client.new\n" " |> host(\"api.openai.com\")\n" " |> path(\"/v1/chat/completions\")\n" " |> stream_yielder()\n" " |> each(fn(result) {\n" " case result {\n" " Ok(chunk) -> print(to_string(chunk))\n" " Error(reason) -> println_error(\"Stream error: \" <> reason)\n" " }\n" " })\n" " ```\n" "\n" " ### 3. Message-Based Streaming - `client.stream_messages()`\n" "\n" " Get messages sent to your process mailbox. Perfect for:\n" " - OTP actors handling multiple concurrent streams\n" " - Long-lived connections that need cancellation\n" " - Integration with OTP supervisors and selectors\n" "\n" " This is a push-based asynchronous API fully compatible with OTP patterns.\n" "\n" " ```gleam\n" " import dream_http_client/client.{\n" " type StreamMessage, Chunk, StreamEnd, StreamError, StreamStart,\n" " select_stream_messages\n" " }\n" " import gleam/otp/actor.{continue}\n" " import gleam/erlang/process.{new_selector}\n" "\n" " pub type Message {\n" " HttpStream(StreamMessage)\n" " }\n" "\n" " fn init_selector() {\n" " new_selector()\n" " |> select_stream_messages(HttpStream)\n" " }\n" "\n" " fn handle_message(msg: Message, state: State) {\n" " case msg {\n" " HttpStream(Chunk(req_id, data)) -> process_chunk(data, state)\n" " HttpStream(StreamEnd(req_id, _)) -> cleanup(req_id, state)\n" " HttpStream(StreamError(req_id, reason)) -> handle_error(req_id, reason, state)\n" " HttpStream(StreamStart(_, _)) -> continue(state)\n" " HttpStream(DecodeError(reason)) -> {\n" " // FFI corruption - report as bug\n" " log_critical_error(\"DecodeError: \" <> reason)\n" " continue(state)\n" " }\n" " }\n" " }\n" " ```\n" "\n" " ## Configuration\n" "\n" " All execution modes support the same builder pattern for configuration:\n" " - **Timeouts**: Use `timeout()` to set request timeout (default: 30 seconds)\n" " - **Headers**: Use `add_header()` for incremental or `headers()` for batch\n" " - **Method/Path/Query**: Standard HTTP request components\n" "\n" " Example with timeout:\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, timeout, send}\n" "\n" " client.new\n" " |> host(\"slow-api.example.com\")\n" " |> timeout(60_000) // 60 second timeout\n" " |> send()\n" " ```\n" "\n" " ## Inspecting Requests\n" "\n" " The `ClientRequest` type is opaque to ensure API stability. Use getter functions\n" " to inspect request properties for logging, testing, or middleware:\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/io\n" "\n" " let req = client.new\n" " |> client.host(\"api.example.com\")\n" " |> client.path(\"/users/123\")\n" "\n" " // Inspect the request before sending\n" " io.println(\"Calling: \" <> client.get_host(req) <> client.get_path(req))\n" " // Prints: \"Calling: api.example.com/users/123\"\n" "\n" " let result = client.send(req)\n" " ```\n" "\n" " Available getters: `get_method`, `get_scheme`, `get_host`, `get_port`, `get_path`,\n" " `get_query`, `get_headers`, `get_body`, `get_timeout`, `get_recorder`\n" ). -opaque client_request() :: {client_request, gleam@http:method(), gleam@http:scheme(), binary(), gleam@option:option(integer()), binary(), gleam@option:option(binary()), list({binary(), binary()}), binary(), gleam@option:option(integer()), gleam@option:option(dream_http_client@recorder:recorder())}. -opaque request_id() :: {request_id, gleam@dynamic:dynamic_()}. -type stream_message() :: {stream_start, request_id(), list({binary(), binary()})} | {chunk, request_id(), bitstring()} | {stream_end, request_id(), list({binary(), binary()})} | {stream_error, request_id(), binary()} | {decode_error, binary()}. -type yielder_state() :: {yielder_state, gleam@option:option(gleam@dynamic:dynamic_()), gleam@http@request:request(binary()), integer()}. -file("src/dream_http_client/client.gleam", 261). ?DOC( " Set the HTTP method for the request\n" "\n" " Configures the HTTP method (GET, POST, PUT, DELETE, etc.) for the request.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `method_value`: The HTTP method to use\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the method updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/http\n" "\n" " client.new\n" " |> client.method(http.Post)\n" " ```\n" ). -spec method(client_request(), gleam@http:method()) -> client_request(). method(Client_request, Method_value) -> {client_request, Method_value, erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 290). ?DOC( " Set the scheme (protocol) for the request\n" "\n" " Configures whether to use HTTP or HTTPS. Defaults to HTTPS for security.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `scheme_value`: The protocol scheme (`http.Http` or `http.Https`)\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the scheme updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/http\n" "\n" " client.new\n" " |> client.scheme(http.Http) // Use HTTP instead of HTTPS\n" " ```\n" ). -spec scheme(client_request(), gleam@http:scheme()) -> client_request(). scheme(Client_request, Scheme_value) -> {client_request, erlang:element(2, Client_request), Scheme_value, erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 318). ?DOC( " Set the host for the request\n" "\n" " Sets the server hostname or IP address. This is required for all requests.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `host_value`: The hostname (e.g., \"api.example.com\" or \"192.168.1.1\")\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the host updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.host(\"api.example.com\")\n" " ```\n" ). -spec host(client_request(), binary()) -> client_request(). host(Client_request, Host_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), Host_value, erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 345). ?DOC( " Set the port for the request\n" "\n" " Sets a custom port number. If not set, defaults to 80 for HTTP and 443\n" " for HTTPS. Only set this if you're using a non-standard port.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `port_value`: The port number (e.g., 8080, 3000)\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the port updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.host(\"localhost\")\n" " |> client.port(3000) // Use port 3000 instead of default\n" " ```\n" ). -spec port(client_request(), integer()) -> client_request(). port(Client_request, Port_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), {some, Port_value}, erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 370). ?DOC( " Set the path for the request\n" "\n" " Sets the request path. Should start with \"/\" for absolute paths.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `path_value`: The path (e.g., \"/api/users\" or \"/api/users/123\")\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the path updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.path(\"/api/users/123\")\n" " ```\n" ). -spec path(client_request(), binary()) -> client_request(). path(Client_request, Path_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), Path_value, erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 396). ?DOC( " Set the query string for the request\n" "\n" " Sets the query string portion of the URL. Do not include the leading \"?\".\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `query_value`: The query string (e.g., \"page=1&limit=10\")\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the query string updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.path(\"/api/users\")\n" " |> client.query(\"page=1&limit=10\")\n" " ```\n" ). -spec 'query'(client_request(), binary()) -> client_request(). 'query'(Client_request, Query_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), {some, Query_value}, erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 428). ?DOC( " Set the headers for the request\n" "\n" " Replaces all existing headers with the provided list. Use `add_header()`\n" " to add a single header without replacing existing ones.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `headers_value`: List of header tuples `#(name, value)`\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with headers replaced.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.headers([\n" " #(\"Authorization\", \"Bearer \" <> token),\n" " #(\"Content-Type\", \"application/json\"),\n" " ])\n" " ```\n" ). -spec headers(client_request(), list({binary(), binary()})) -> client_request(). headers(Client_request, Headers_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), Headers_value, erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 464). ?DOC( " Set the body for the request\n" "\n" " Sets the request body as a string. Typically used for POST, PUT, and PATCH\n" " requests. For JSON, serialize your data first.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `body_value`: The request body as a string\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the body updated.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/json\n" "\n" " let json_body = json.object([\n" " #(\"name\", json.string(\"Alice\")),\n" " #(\"email\", json.string(\"alice@example.com\")),\n" " ])\n" "\n" " client.new\n" " |> client.method(http.Post)\n" " |> client.body(json.to_string(json_body))\n" " ```\n" ). -spec body(client_request(), binary()) -> client_request(). body(Client_request, Body_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), Body_value, erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 497). ?DOC( " Set the recorder for the request\n" "\n" " Attaches a recorder to the request for recording or playback.\n" " The recorder must be started with `recorder.start()` before use.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `recorder_value`: The recorder to attach\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the recorder attached.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import dream_http_client/recorder\n" "\n" " let assert Ok(rec) = recorder.start(\n" " mode: recorder.Record(directory: \"mocks\"),\n" " matching: recorder.match_url_only(),\n" " )\n" "\n" " client.new\n" " |> client.host(\"api.example.com\")\n" " |> client.recorder(rec)\n" " ```\n" ). -spec recorder(client_request(), dream_http_client@recorder:recorder()) -> client_request(). recorder(Client_request, Recorder_value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(10, Client_request), {some, Recorder_value}}. -file("src/dream_http_client/client.gleam", 522). ?DOC( " Set the timeout for the request in milliseconds\n" "\n" " Sets how long to wait for a response before timing out. If not set,\n" " defaults to 30000ms (30 seconds).\n" "\n" " ## Parameters\n" "\n" " - `timeout_ms`: Timeout duration in milliseconds\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, timeout}\n" "\n" " client.new\n" " |> host(\"slow-api.example.com\")\n" " |> timeout(60_000) // 60 second timeout\n" " ```\n" ). -spec timeout(client_request(), integer()) -> client_request(). timeout(Client_request, Timeout_ms) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), {some, Timeout_ms}, erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 551). ?DOC( " Add a header to the request\n" "\n" " Adds a single header to the existing headers list without replacing them.\n" " The new header is prepended to the list, so it will take precedence if\n" " there's a duplicate header name.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The request to modify\n" " - `name`: The header name (e.g., \"Authorization\", \"Content-Type\")\n" " - `value`: The header value\n" "\n" " ## Returns\n" "\n" " A new `ClientRequest` with the header added.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " client.new\n" " |> client.add_header(\"Authorization\", \"Bearer \" <> token)\n" " |> client.add_header(\"Content-Type\", \"application/json\")\n" " ```\n" ). -spec add_header(client_request(), binary(), binary()) -> client_request(). add_header(Client_request, Name, Value) -> {client_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), [{Name, Value} | erlang:element(8, Client_request)], erlang:element(9, Client_request), erlang:element(10, Client_request), erlang:element(11, Client_request)}. -file("src/dream_http_client/client.gleam", 580). ?DOC( " Get the HTTP method from a request\n" "\n" " Returns the HTTP method (GET, POST, etc.) configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/http.{Post}\n" "\n" " let req = client.new |> client.method(Post)\n" " let method = client.get_method(req)\n" " // method == Post\n" " ```\n" ). -spec get_method(client_request()) -> gleam@http:method(). get_method(Client_request) -> erlang:element(2, Client_request). -file("src/dream_http_client/client.gleam", 598). ?DOC( " Get the URI scheme from a request\n" "\n" " Returns the scheme (HTTP or HTTPS) configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import gleam/http.{Http}\n" "\n" " let req = client.new |> client.scheme(Http)\n" " let scheme = client.get_scheme(req)\n" " // scheme == Http\n" " ```\n" ). -spec get_scheme(client_request()) -> gleam@http:scheme(). get_scheme(Client_request) -> erlang:element(3, Client_request). -file("src/dream_http_client/client.gleam", 615). ?DOC( " Get the host from a request\n" "\n" " Returns the hostname configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.host(\"api.example.com\")\n" " let host = client.get_host(req)\n" " // host == \"api.example.com\"\n" " ```\n" ). -spec get_host(client_request()) -> binary(). get_host(Client_request) -> erlang:element(4, Client_request). -file("src/dream_http_client/client.gleam", 633). ?DOC( " Get the port from a request\n" "\n" " Returns the optional port number configured for the request.\n" " If None, the default port for the scheme will be used (80 for HTTP, 443 for HTTPS).\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.port(8080)\n" " let port = client.get_port(req)\n" " // port == Some(8080)\n" " ```\n" ). -spec get_port(client_request()) -> gleam@option:option(integer()). get_port(Client_request) -> erlang:element(5, Client_request). -file("src/dream_http_client/client.gleam", 650). ?DOC( " Get the path from a request\n" "\n" " Returns the request path configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.path(\"/api/users\")\n" " let path = client.get_path(req)\n" " // path == \"/api/users\"\n" " ```\n" ). -spec get_path(client_request()) -> binary(). get_path(Client_request) -> erlang:element(6, Client_request). -file("src/dream_http_client/client.gleam", 667). ?DOC( " Get the query string from a request\n" "\n" " Returns the optional query string configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.query(\"page=1&limit=10\")\n" " let query = client.get_query(req)\n" " // query == Some(\"page=1&limit=10\")\n" " ```\n" ). -spec get_query(client_request()) -> gleam@option:option(binary()). get_query(Client_request) -> erlang:element(7, Client_request). -file("src/dream_http_client/client.gleam", 686). ?DOC( " Get the headers from a request\n" "\n" " Returns the list of headers configured for the request.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new\n" " |> client.add_header(\"Authorization\", \"Bearer token\")\n" " |> client.add_header(\"Content-Type\", \"application/json\")\n" " let headers = client.get_headers(req)\n" " // headers == [#(\"Content-Type\", \"application/json\"), #(\"Authorization\", \"Bearer token\")]\n" " ```\n" ). -spec get_headers(client_request()) -> list({binary(), binary()}). get_headers(Client_request) -> erlang:element(8, Client_request). -file("src/dream_http_client/client.gleam", 703). ?DOC( " Get the body from a request\n" "\n" " Returns the request body as a string.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.body(\"{\\\"name\\\": \\\"Alice\\\"}\")\n" " let body = client.get_body(req)\n" " // body == \"{\\\"name\\\": \\\"Alice\\\"}\"\n" " ```\n" ). -spec get_body(client_request()) -> binary(). get_body(Client_request) -> erlang:element(9, Client_request). -file("src/dream_http_client/client.gleam", 721). ?DOC( " Get the timeout from a request\n" "\n" " Returns the optional timeout in milliseconds configured for the request.\n" " If None, the default timeout (30000ms) will be used.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" "\n" " let req = client.new |> client.timeout(5000)\n" " let timeout = client.get_timeout(req)\n" " // timeout == Some(5000)\n" " ```\n" ). -spec get_timeout(client_request()) -> gleam@option:option(integer()). get_timeout(Client_request) -> erlang:element(10, Client_request). -file("src/dream_http_client/client.gleam", 744). ?DOC( " Get the recorder from a request\n" "\n" " Returns the optional recorder attached to the request for recording or playback.\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client\n" " import dream_http_client/recorder\n" " import dream_http_client/matching\n" "\n" " let assert Ok(rec) = recorder.start(\n" " mode: recorder.Record(directory: \"mocks\"),\n" " matching: matching.match_url_only(),\n" " )\n" " let req = client.new |> client.recorder(rec)\n" " let recorder_opt = client.get_recorder(req)\n" " // recorder_opt == Some(rec)\n" " ```\n" ). -spec get_recorder(client_request()) -> gleam@option:option(dream_http_client@recorder:recorder()). get_recorder(Client_request) -> erlang:element(11, Client_request). -file("src/dream_http_client/client.gleam", 925). -spec client_request_to_recorded_request(client_request()) -> dream_http_client@recording:recorded_request(). client_request_to_recorded_request(Client_request) -> {recorded_request, erlang:element(2, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request)}. -file("src/dream_http_client/client.gleam", 940). -spec convert_string_error(nil) -> binary(). convert_string_error(_) -> <<"Failed to convert response to string"/utf8>>. -file("src/dream_http_client/client.gleam", 944). -spec resolve_timeout(client_request()) -> integer(). resolve_timeout(Client_request) -> case erlang:element(10, Client_request) of {some, Timeout_value} -> Timeout_value; none -> 30000 end. -file("src/dream_http_client/client.gleam", 1108). -spec convert_chunk_to_result(dream_http_client@recording:chunk()) -> {ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}. convert_chunk_to_result(Chunk) -> Data = gleam@bytes_tree:from_bit_array(erlang:element(2, Chunk)), {ok, Data}. -file("src/dream_http_client/client.gleam", 1100). -spec create_yielder_from_chunks(list(dream_http_client@recording:chunk())) -> gleam@yielder:yielder({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}). create_yielder_from_chunks(Chunks) -> _pipe = Chunks, _pipe@1 = gleam@yielder:from_list(_pipe), gleam@yielder:map(_pipe@1, fun convert_chunk_to_result/1). -file("src/dream_http_client/client.gleam", 1133). -spec to_http_request(client_request()) -> gleam@http@request:request(binary()). to_http_request(Client_request) -> {request, erlang:element(2, Client_request), erlang:element(8, Client_request), erlang:element(9, Client_request), erlang:element(3, Client_request), erlang:element(4, Client_request), erlang:element(5, Client_request), erlang:element(6, Client_request), erlang:element(7, Client_request)}. -file("src/dream_http_client/client.gleam", 1146). -spec handle_yielder_start_with_state(yielder_state()) -> gleam@yielder:step({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}, yielder_state()). handle_yielder_start_with_state(State) -> Request_result = dream_http_client@internal:start_httpc_stream( erlang:element(3, State), erlang:element(4, State) ), Owner = dream_http_client@internal:extract_owner_pid(Request_result), case dream_http_client@internal:receive_next( Owner, erlang:element(4, State) ) of {ok, {some, Bin}} -> {next, {ok, gleam@bytes_tree:from_bit_array(Bin)}, {yielder_state, {some, Owner}, erlang:element(3, State), erlang:element(4, State)}}; {ok, none} -> done; {error, Error_reason} -> {next, {error, Error_reason}, State} end. -file("src/dream_http_client/client.gleam", 1163). -spec handle_yielder_next_with_state(gleam@dynamic:dynamic_(), yielder_state()) -> gleam@yielder:step({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}, yielder_state()). handle_yielder_next_with_state(Owner, State) -> case dream_http_client@internal:receive_next( Owner, erlang:element(4, State) ) of {ok, {some, Bin}} -> {next, {ok, gleam@bytes_tree:from_bit_array(Bin)}, State}; {ok, none} -> done; {error, Error_reason} -> {next, {error, Error_reason}, State} end. -file("src/dream_http_client/client.gleam", 1124). -spec handle_yielder_unfold_with_deps(yielder_state()) -> gleam@yielder:step({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}, yielder_state()). handle_yielder_unfold_with_deps(State) -> case erlang:element(2, State) of none -> handle_yielder_start_with_state(State); {some, Owner} -> handle_yielder_next_with_state(Owner, State) end. -file("src/dream_http_client/client.gleam", 1089). -spec stream_yielder_real(client_request()) -> gleam@yielder:yielder({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}). stream_yielder_real(Req) -> Http_req = to_http_request(Req), Timeout_value = resolve_timeout(Req), Initial_state = {yielder_state, none, Http_req, Timeout_value}, gleam@yielder:unfold(Initial_state, fun handle_yielder_unfold_with_deps/1). -file("src/dream_http_client/client.gleam", 1055). ?DOC( " Stream HTTP response chunks using a yielder\n" "\n" " Sends an HTTP request and returns a yielder that produces chunks of the\n" " response body as they arrive from the server. This allows you to process\n" " large responses incrementally without loading the entire response into memory.\n" "\n" " **Use this for simple sequential streaming:**\n" " - AI/LLM inference endpoints (stream tokens)\n" " - Simple file downloads\n" " - Scripts or one-off operations\n" "\n" " **For OTP actors with concurrency, use `stream_messages()` instead.**\n" "\n" " ## Error Semantics\n" "\n" " The yielder produces `Result(BytesTree, String)` for each chunk:\n" " - `Ok(chunk)` - Successful chunk, more may follow\n" " - `Error(reason)` - **Terminal error**, stream is done\n" "\n" " After an `Error`, the yielder immediately returns `Done` on the next call.\n" " This design reflects that HTTP stream errors (timeouts, connection drops,\n" " etc.) are **not recoverable** - you cannot continue reading from a broken stream.\n" "\n" " **Normal stream completion**: When the stream finishes successfully, the yielder\n" " returns `Done` (no more items). The stream does NOT yield an error for normal completion.\n" "\n" " Possible error reasons (actual errors only):\n" " - `\"timeout\"` - Request timed out\n" " - Connection errors from `httpc`\n" "\n" " ## Parameters\n" "\n" " - `req`: The configured HTTP request\n" "\n" " ## Returns\n" "\n" " A `Yielder` that produces `Result(BytesTree, String)`. Always check each\n" " result - errors are terminal and mean the stream has ended.\n" "\n" " ## Examples\n" "\n" " **Streaming and processing chunks as they arrive:**\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, path, stream_yielder}\n" " import gleam/yielder.{each}\n" " import gleam/bytes_tree.{to_string}\n" " import gleam/io.{print}\n" "\n" " client.new\n" " |> host(\"api.openai.com\")\n" " |> path(\"/v1/chat/completions\")\n" " |> stream_yielder()\n" " |> each(fn(result) {\n" " case result {\n" " Ok(chunk) -> print(to_string(chunk))\n" " Error(reason) -> {\n" " io.println_error(\"Stream error: \" <> reason)\n" " // Stream is now done, no more chunks will arrive\n" " }\n" " }\n" " })\n" " ```\n" "\n" " **Collecting all chunks into a list:**\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, path, stream_yielder}\n" " import gleam/yielder\n" " import gleam/list\n" " import gleam/bytes_tree\n" " import gleam/string\n" "\n" " // The stream automatically completes when done - no need to use take()!\n" " let chunks = \n" " client.new\n" " |> host(\"example.com\")\n" " |> path(\"/data\")\n" " |> stream_yielder()\n" " |> yielder.to_list()\n" "\n" " // Handle results\n" " case list.try_map(chunks, fn(r) { r }) {\n" " Ok(chunk_list) -> {\n" " // Concatenate all chunks\n" " let body = \n" " chunk_list\n" " |> list.map(bytes_tree.to_string)\n" " |> list.map(fn(r) { result.unwrap(r, \"\") })\n" " |> string.join(\"\")\n" " Ok(body)\n" " }\n" " Error(reason) -> Error(\"Stream failed: \" <> reason)\n" " }\n" " ```\n" ). -spec stream_yielder(client_request()) -> gleam@yielder:yielder({ok, gleam@bytes_tree:bytes_tree()} | {error, binary()}). stream_yielder(Req) -> case erlang:element(11, Req) of {some, Rec} -> Recorded_req = client_request_to_recorded_request(Req), case dream_http_client@recorder:find_recording(Rec, Recorded_req) of {some, {recording, _, Response}} -> case Response of {streaming_response, _, _, Chunks} -> create_yielder_from_chunks(Chunks); {blocking_response, _, _, Body} -> Chunk = gleam@bytes_tree:from_bit_array( <> ), gleam@yielder:single({ok, Chunk}) end; none -> stream_yielder_real(Req) end; none -> stream_yielder_real(Req) end. -file("src/dream_http_client/client.gleam", 1255). -spec build_url(gleam@http@request:request(binary())) -> binary(). build_url(Req) -> Port_string = case erlang:element(7, Req) of {some, P} -> <<":"/utf8, (erlang:integer_to_binary(P))/binary>>; none -> <<""/utf8>> end, <<<<<<<<(gleam@http:scheme_to_string(erlang:element(5, Req)))/binary, "://"/utf8>>/binary, (erlang:element(6, Req))/binary>>/binary, Port_string/binary>>/binary, (erlang:element(8, Req))/binary>>. -file("src/dream_http_client/client.gleam", 907). -spec send_real_request(client_request()) -> {ok, binary()} | {error, binary()}. send_real_request(Client_request) -> Http_req = to_http_request(Client_request), Url = build_url(Http_req), Method_atom = dream_http_client@internal:atomize_method( erlang:element(2, Http_req) ), Method_dynamic = gleam_erlang_ffi:identity(Method_atom), Body = <<(erlang:element(4, Http_req))/binary>>, Timeout_value = resolve_timeout(Client_request), case dream_httpc_shim:request_sync( Method_dynamic, Url, erlang:element(3, Http_req), Body, Timeout_value ) of {ok, Response_body} -> _pipe = Response_body, _pipe@1 = gleam@bit_array:to_string(_pipe), gleam@result:map_error(_pipe@1, fun convert_string_error/1); {error, Error_msg} -> {error, Error_msg} end. -file("src/dream_http_client/client.gleam", 847). ?DOC( " Make a blocking HTTP request and get the complete response\n" "\n" " Sends an HTTP request and collects all response chunks, returning the\n" " complete response body as a string. This is ideal for:\n" "\n" " - JSON API responses\n" " - Small files or documents\n" " - Any case where you need the full response before processing\n" "\n" " For large responses or when you need OTP compatibility, use\n" " `stream_yielder()` or `stream_messages()` instead.\n" "\n" " ## Parameters\n" "\n" " - `client_request`: The configured HTTP request\n" "\n" " ## Returns\n" "\n" " - `Ok(String)`: The complete response body as a string\n" " - `Error(String)`: An error message if the request failed\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, path, add_header, send}\n" " import gleam/json.{decode}\n" "\n" " let result = client.new\n" " |> host(\"api.example.com\")\n" " |> path(\"/users/123\")\n" " |> add_header(\"Authorization\", \"Bearer \" <> token)\n" " |> send()\n" "\n" " case result {\n" " Ok(body) -> {\n" " case decode(body, user_decoder) {\n" " Ok(user) -> Ok(user)\n" " Error(_) -> Error(\"Invalid JSON response\")\n" " }\n" " }\n" " Error(msg) -> Error(\"Request failed: \" <> msg)\n" " }\n" " ```\n" ). -spec send(client_request()) -> {ok, binary()} | {error, binary()}. send(Client_request) -> case erlang:element(11, Client_request) of {some, Rec} -> Recorded_req = client_request_to_recorded_request(Client_request), case dream_http_client@recorder:find_recording(Rec, Recorded_req) of {some, {recording, _, Response}} -> case Response of {blocking_response, _, _, Body} -> {ok, Body}; {streaming_response, _, _, _} -> {error, <<"Recording contains streaming response, use stream_yielder() instead"/utf8>>} end; none -> Result = send_real_request(Client_request), case Result of {ok, Body@1} -> case dream_http_client@recorder:is_record_mode(Rec) of true -> Recorded_resp = {blocking_response, 200, [], Body@1}, Rec_entry = {recording, Recorded_req, Recorded_resp}, dream_http_client@recorder:add_recording( Rec, Rec_entry ); false -> nil end, {ok, Body@1}; {error, _} -> Result end end; none -> send_real_request(Client_request) end. -file("src/dream_http_client/client.gleam", 1288). -spec extract_request_id(gleam@dynamic:dynamic_()) -> {ok, request_id()} | {error, binary()}. extract_request_id(Result) -> Id_result = gleam@dynamic@decode:run( Result, gleam@dynamic@decode:at( [1], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), case Id_result of {ok, Id_dyn} -> {ok, {request_id, Id_dyn}}; {error, Decode_errors} -> {error, <<"Failed to extract request ID: "/utf8, (gleam@string:inspect(Decode_errors))/binary>>} end. -file("src/dream_http_client/client.gleam", 1297). -spec extract_error_reason(gleam@dynamic:dynamic_()) -> {ok, request_id()} | {error, binary()}. extract_error_reason(Result) -> Reason_result = gleam@dynamic@decode:run( Result, gleam@dynamic@decode:at( [1], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), case Reason_result of {ok, Reason_dyn} -> Reason = gleam@string:inspect(Reason_dyn), {error, <<"Failed to start stream: "/utf8, Reason/binary>>}; {error, Decode_error} -> {error, <<<<"Failed to start stream (decode error: "/utf8, (gleam@string:inspect(Decode_error))/binary>>/binary, ")"/utf8>>} end. -file("src/dream_http_client/client.gleam", 1276). -spec parse_stream_start_tag(gleam@dynamic:dynamic_(), gleam@dynamic:dynamic_()) -> {ok, request_id()} | {error, binary()}. parse_stream_start_tag(Tag_dyn, Result) -> Tag = begin _pipe = gleam_erlang_ffi:identity(Tag_dyn), erlang:atom_to_binary(_pipe) end, case Tag of <<"ok"/utf8>> -> extract_request_id(Result); <<"error"/utf8>> -> extract_error_reason(Result); _ -> {error, <<"Unknown response from httpc"/utf8>>} end. -file("src/dream_http_client/client.gleam", 1267). -spec parse_stream_start_result(gleam@dynamic:dynamic_()) -> {ok, request_id()} | {error, binary()}. parse_stream_start_result(Result) -> Tag_result = gleam@dynamic@decode:run( Result, gleam@dynamic@decode:at( [0], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), case Tag_result of {ok, Tag_dyn} -> parse_stream_start_tag(Tag_dyn, Result); {error, Decode_errors} -> {error, <<"Failed to parse httpc response: "/utf8, (gleam@string:inspect(Decode_errors))/binary>>} end. -file("src/dream_http_client/client.gleam", 1234). ?DOC( " Start a message-based streaming HTTP request (OTP compatible)\n" "\n" " Sends an HTTP request and returns a request ID immediately. httpc sends\n" " stream messages directly to your process mailbox. Use this for:\n" "\n" " - **OTP actors handling multiple concurrent streams**\n" " - **Long-lived connections that need cancellation**\n" " - **Integration with OTP supervisors and selectors**\n" "\n" " For simple sequential streaming, use `stream_yielder()` instead.\n" "\n" " ## Message Flow\n" "\n" " Messages are sent to your process mailbox automatically:\n" " 1. `StreamStart(request_id, headers)` - Headers received\n" " 2. `Chunk(request_id, data)` - Zero or more data chunks\n" " 3. `StreamEnd(request_id, headers)` or `StreamError(request_id, reason)` - Done\n" "\n" " ## Parameters\n" "\n" " - `req`: The configured HTTP request\n" "\n" " ## Returns\n" "\n" " - `Ok(RequestId)`: Stream started, messages will arrive in your mailbox\n" " - `Error(String)`: Failed to start the stream\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client.{\n" " type StreamMessage, Chunk, StreamEnd, StreamError, StreamStart,\n" " select_stream_messages\n" " }\n" " import gleam/otp/actor.{continue}\n" " import gleam/erlang/process.{new_selector}\n" "\n" " pub type Message {\n" " HttpStream(StreamMessage)\n" " }\n" "\n" " fn handle_message(msg: Message, state: State) {\n" " case msg {\n" " HttpStream(stream_msg) -> {\n" " case stream_msg {\n" " Chunk(req_id, data) -> process_chunk(data, state)\n" " StreamEnd(req_id, _) -> cleanup(req_id, state)\n" " StreamError(req_id, reason) -> handle_error(req_id, reason, state)\n" " StreamStart(_, _) -> continue(state)\n" " }\n" " }\n" " }\n" " }\n" "\n" " fn init_selector() {\n" " new_selector()\n" " |> select_stream_messages(HttpStream)\n" " }\n" " ```\n" ). -spec stream_messages(client_request()) -> {ok, request_id()} | {error, binary()}. stream_messages(Req) -> Http_req = to_http_request(Req), Url = build_url(Http_req), Method_atom = dream_http_client@internal:atomize_method( erlang:element(2, Http_req) ), Body = <<(erlang:element(4, Http_req))/binary>>, Me = erlang:self(), Timeout_value = resolve_timeout(Req), Result = dream_httpc_shim:request_stream_messages( Method_atom, Url, erlang:element(3, Http_req), Body, Me, Timeout_value ), parse_stream_start_result(Result). -file("src/dream_http_client/client.gleam", 1388). -spec handle_tag_decode_error( list(gleam@dynamic@decode:decode_error()), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). handle_tag_decode_error(Decode_error, Req_id_result) -> Error_msg = <<"Internal error: Failed to decode stream message tag: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, case Req_id_result of {ok, Req_id_dyn} -> Req_id = {request_id, Req_id_dyn}, {stream_error, Req_id, Error_msg}; {error, Req_id_error} -> Full_error_msg = <<<<<>/binary, (gleam@string:inspect(Req_id_error))/binary>>/binary, ")"/utf8>>, {decode_error, Full_error_msg} end. -file("src/dream_http_client/client.gleam", 1492). -spec decode_chunk_data(request_id(), gleam@dynamic:dynamic_()) -> stream_message(). decode_chunk_data(Req_id, Data_dyn) -> case gleam@dynamic@decode:run( Data_dyn, {decoder, fun gleam@dynamic@decode:decode_bit_array/1} ) of {ok, Data} -> {chunk, Req_id, Data}; {error, Decode_error} -> Error_msg = <<"Internal error: Failed to decode chunk data: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1477). -spec decode_chunk( request_id(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_chunk(Req_id, Data_result) -> case Data_result of {ok, Data_dyn} -> decode_chunk_data(Req_id, Data_dyn); {error, Decode_error} -> Error_msg = <<"Internal error: Failed to get chunk data: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1550). -spec decode_error_reason(request_id(), gleam@dynamic:dynamic_()) -> stream_message(). decode_error_reason(Req_id, Reason_dyn) -> case gleam@dynamic@decode:run( Reason_dyn, {decoder, fun gleam@dynamic@decode:decode_string/1} ) of {ok, Reason} -> {stream_error, Req_id, Reason}; {error, Decode_error} -> Error_msg = <<<<"Stream error (failed to decode error string: "/utf8, (gleam@string:inspect(Decode_error))/binary>>/binary, ")"/utf8>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1534). -spec decode_stream_error( request_id(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_stream_error(Req_id, Data_result) -> case Data_result of {ok, Reason_dyn} -> decode_error_reason(Req_id, Reason_dyn); {error, Decode_error} -> Error_msg = <<<<"Stream error (failed to decode error reason: "/utf8, (gleam@string:inspect(Decode_error))/binary>>/binary, ")"/utf8>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1582). -spec pair_with_name(binary(), binary()) -> {binary(), binary()}. pair_with_name(Value, Name) -> {Name, Value}. -file("src/dream_http_client/client.gleam", 1577). -spec decode_header_value(binary()) -> gleam@dynamic@decode:decoder({binary(), binary()}). decode_header_value(Name) -> _pipe = gleam@dynamic@decode:at( [1], {decoder, fun gleam@dynamic@decode:decode_string/1} ), gleam@dynamic@decode:map( _pipe, fun(_capture) -> pair_with_name(_capture, Name) end ). -file("src/dream_http_client/client.gleam", 1566). -spec decode_headers(gleam@dynamic:dynamic_()) -> {ok, list({binary(), binary()})} | {error, list(gleam@dynamic@decode:decode_error())}. decode_headers(Dyn) -> Header_decoder = begin _pipe = gleam@dynamic@decode:at( [0], {decoder, fun gleam@dynamic@decode:decode_string/1} ), gleam@dynamic@decode:then(_pipe, fun decode_header_value/1) end, gleam@dynamic@decode:run(Dyn, gleam@dynamic@decode:list(Header_decoder)). -file("src/dream_http_client/client.gleam", 1462). -spec decode_stream_start_headers(request_id(), gleam@dynamic:dynamic_()) -> stream_message(). decode_stream_start_headers(Req_id, Headers_dyn) -> case decode_headers(Headers_dyn) of {ok, Headers} -> {stream_start, Req_id, Headers}; {error, Header_decode_error} -> Error_msg = <<"Failed to decode headers in StreamStart: "/utf8, (gleam@string:inspect(Header_decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1447). -spec decode_stream_start( request_id(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_stream_start(Req_id, Data_result) -> case Data_result of {ok, Headers_dyn} -> decode_stream_start_headers(Req_id, Headers_dyn); {error, Decode_error} -> Error_msg = <<"Failed to get headers data in StreamStart: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1519). -spec decode_stream_end_headers(request_id(), gleam@dynamic:dynamic_()) -> stream_message(). decode_stream_end_headers(Req_id, Headers_dyn) -> case decode_headers(Headers_dyn) of {ok, Headers} -> {stream_end, Req_id, Headers}; {error, Header_decode_error} -> Error_msg = <<"Failed to decode trailing headers in StreamEnd: "/utf8, (gleam@string:inspect(Header_decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1504). -spec decode_stream_end( request_id(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_stream_end(Req_id, Data_result) -> case Data_result of {ok, Headers_dyn} -> decode_stream_end_headers(Req_id, Headers_dyn); {error, Decode_error} -> Error_msg = <<"Failed to get trailing headers data in StreamEnd: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, {stream_error, Req_id, Error_msg} end. -file("src/dream_http_client/client.gleam", 1432). -spec decode_by_tag( binary(), request_id(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_by_tag(Tag, Req_id, Data_result) -> case Tag of <<"stream_start"/utf8>> -> decode_stream_start(Req_id, Data_result); <<"chunk"/utf8>> -> decode_chunk(Req_id, Data_result); <<"stream_end"/utf8>> -> decode_stream_end(Req_id, Data_result); <<"stream_error"/utf8>> -> decode_stream_error(Req_id, Data_result); _ -> {stream_error, Req_id, <<"Internal error: Unknown stream message tag: "/utf8, Tag/binary>>} end. -file("src/dream_http_client/client.gleam", 1411). -spec decode_with_tag( gleam@dynamic:dynamic_(), {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())}, {ok, gleam@dynamic:dynamic_()} | {error, list(gleam@dynamic@decode:decode_error())} ) -> stream_message(). decode_with_tag(Tag_dyn, Req_id_result, Data_result) -> Tag = begin _pipe = gleam_erlang_ffi:identity(Tag_dyn), erlang:atom_to_binary(_pipe) end, case Req_id_result of {ok, Req_id_dyn} -> Req_id = {request_id, Req_id_dyn}, decode_by_tag(Tag, Req_id, Data_result); {error, Decode_error} -> Error_msg = <<"Internal error: Failed to decode request ID from stream message: "/utf8, (gleam@string:inspect(Decode_error))/binary>>, {decode_error, Error_msg} end. -file("src/dream_http_client/client.gleam", 1376). -spec decode_simplified_message(gleam@dynamic:dynamic_()) -> stream_message(). decode_simplified_message(Dyn) -> Tag_result = gleam@dynamic@decode:run( Dyn, gleam@dynamic@decode:at( [0], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), Req_id_result = gleam@dynamic@decode:run( Dyn, gleam@dynamic@decode:at( [1], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), Data_result = gleam@dynamic@decode:run( Dyn, gleam@dynamic@decode:at( [2], {decoder, fun gleam@dynamic@decode:decode_dynamic/1} ) ), case Tag_result of {ok, Tag_dyn} -> decode_with_tag(Tag_dyn, Req_id_result, Data_result); {error, Decode_error} -> handle_tag_decode_error(Decode_error, Req_id_result) end. -file("src/dream_http_client/client.gleam", 1365). -spec apply_mapper_to_dynamic( gleam@dynamic:dynamic_(), fun((stream_message()) -> UPQ) ) -> UPQ. apply_mapper_to_dynamic(Dyn, Mapper) -> Simplified = dream_httpc_shim:decode_stream_message_for_selector(Dyn), Stream_msg = decode_simplified_message(Simplified), Mapper(Stream_msg). -file("src/dream_http_client/client.gleam", 1359). -spec create_selector_mapper(fun((stream_message()) -> UPP)) -> fun((gleam@dynamic:dynamic_()) -> UPP). create_selector_mapper(Mapper) -> fun(_capture) -> apply_mapper_to_dynamic(_capture, Mapper) end. -file("src/dream_http_client/client.gleam", 1347). ?DOC( " Add stream message handling to an OTP selector\n" "\n" " Integrates HTTP stream messages into your OTP actor's selector. This allows\n" " you to handle HTTP streams alongside other messages in your actor.\n" "\n" " The mapper function converts `StreamMessage` to your actor's message type.\n" "\n" " ## Parameters\n" "\n" " - `selector`: Your existing selector\n" " - `mapper`: Function to wrap `StreamMessage` in your message type\n" "\n" " ## Returns\n" "\n" " Updated selector that handles stream messages\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client.{type StreamMessage, select_stream_messages}\n" " import gleam/erlang/process.{type Selector, new_selector, selecting}\n" "\n" " pub type Message {\n" " HttpStream(StreamMessage)\n" " OtherMessage(String)\n" " }\n" "\n" " fn build_selector() -> Selector(Message) {\n" " new_selector()\n" " |> select_stream_messages(HttpStream)\n" " |> selecting(some_subject, OtherMessage)\n" " }\n" " ```\n" ). -spec select_stream_messages( gleam@erlang@process:selector(UPM), fun((stream_message()) -> UPM) ) -> gleam@erlang@process:selector(UPM). select_stream_messages(Selector, Mapper) -> _pipe = Selector, gleam@erlang@process:select_record( _pipe, erlang:binary_to_atom(<<"http"/utf8>>), 1, create_selector_mapper(Mapper) ). -file("src/dream_http_client/client.gleam", 1607). ?DOC( " Cancel an active streaming request\n" "\n" " Cancels an HTTP stream that was started with `stream_messages()`.\n" " After cancellation, no more messages will be sent to your process.\n" "\n" " ## Parameters\n" "\n" " - `request_id`: The request ID returned from `stream_messages()`\n" "\n" " ## Example\n" "\n" " ```gleam\n" " import dream_http_client/client.{host, stream_messages, cancel_stream}\n" "\n" " let assert Ok(req_id) = client.new\n" " |> host(\"api.example.com\")\n" " |> stream_messages()\n" "\n" " // Later, cancel the stream\n" " cancel_stream(req_id)\n" " ```\n" ). -spec cancel_stream(request_id()) -> nil. cancel_stream(Request_id) -> {request_id, Internal_id} = Request_id, dream_httpc_shim:cancel_stream(Internal_id).