Current section
Files
Jump to
Current section
Files
src/dream_http_client@client.erl
-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(
<<Body/binary>>
),
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 = <<<<<<Error_msg/binary,
" (also failed to decode request ID: "/utf8>>/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).