katipo (katipo v2.0.0-rc.2)

View Source

An HTTP/HTTP2/HTTP3 client library for Erlang built around libcurl-multi and libevent.

Quick Start

{ok, _} = application:ensure_all_started(katipo).
{ok, _} = katipo_pool:start(my_pool, 2, [{pipelining, multiplex}]).
{ok, #{status := 200, body := Body}} = katipo:get(my_pool, <<"https://example.com">>).

Request Options

Options can be passed as the third argument to HTTP method functions, or included directly in the request map passed to req/2.

See opts/0 for all available options and request/0 for the full request map type.

Responses

Synchronous request functions return response/0:

{ok, #{status := pos_integer(), headers := headers(), cookiejar := cookiejar(), body := body()}}
{error, #{code := error_code(), message := error_msg()}}

Async Requests

Async functions (async_get/2,3, async_req/2, etc.) return {ok, Ref} immediately and deliver the response as a message to the calling process (or the pid specified by the reply_to option):

{katipo_response, Ref, #{status := pos_integer(), headers := headers(), ...}}
{katipo_error, Ref, #{code := error_code(), message := error_msg()}}

Use await/1,2 to block until the response arrives, or cancel/2 to abort an in-flight request (no response is then delivered).

If the pool worker handling an in-flight async request dies (e.g. its port crashes), a {katipo_error, Ref, #{code => worker_died}} message is delivered so the caller fails fast instead of blocking until the request timeout. If the request cannot be handed to a worker at all (the worker is dead or being restarted), the async function returns {error, #{code => worker_died}} instead of {ok, Ref} -- an accepted request always produces exactly one terminal message. Handing an async request to a worker is itself bounded (5s): a wedged or deeply backed-up worker yields {error, #{code => admission_timeout}} rather than blocking the caller. (Sync requests are bounded by the request timeout instead -- their single call carries the whole response.)

With stream => true the body is delivered incrementally as {katipo_headers, Ref, _}, {katipo_chunk, Ref, _}* and a terminal {katipo_done, Ref, _} instead of one buffered response; see async_req/2.

Async requests emit the same OTel span (HTTP <METHOD>, parented to the caller's context) and metrics as their synchronous counterparts; the span covers the full request and is finished when the response, a timeout, or a worker failure arrives.

Summary

Types

async_response()
body()
connecttimeout_ms()
cookiejar()
curlmopts()
curlopt_http_version()
curlopt_sslversion()
doh_url()
error_code()
error_msg()
header()
headers()
http_auth()
interface()
method()
metrics()
opts()
pipelining()
pool_opts()
proxy()
qs_vals()
req_body()
request()
response()
ssl_verifyhost()
ssl_verifypeer()
sslcert()
sslkey()
sslkey_blob()
status()
tcp_fastopen()
unix_socket_path()
url()
userpwd()

Functions

async_delete(PoolName, Url)

Equivalent to async_delete/3.

async_delete(PoolName, Url, Opts)

Performs an async HTTP DELETE request. Returns {ok, Ref} immediately.

async_get(PoolName, Url)

Equivalent to async_get/3.

async_get(PoolName, Url, Opts)

Performs an async HTTP GET request. Returns {ok, Ref} immediately. The response is delivered as a {katipo_response, Ref, Response} message.

async_head(PoolName, Url)

Equivalent to async_head/3.

async_head(PoolName, Url, Opts)

Performs an async HTTP HEAD request. Returns {ok, Ref} immediately.

async_options(PoolName, Url)

Equivalent to async_options/3.

async_options(PoolName, Url, Opts)

Performs an async HTTP OPTIONS request. Returns {ok, Ref} immediately.

async_patch(PoolName, Url)

Equivalent to async_patch/3.

async_patch(PoolName, Url, Opts)

Performs an async HTTP PATCH request. Returns {ok, Ref} immediately.

async_post(PoolName, Url)

Equivalent to async_post/3.

async_post(PoolName, Url, Opts)

Performs an async HTTP POST request. Returns {ok, Ref} immediately.

async_put(PoolName, Url)

Equivalent to async_put/3.

async_put(PoolName, Url, Opts)

Performs an async HTTP PUT request. Returns {ok, Ref} immediately.

async_req(PoolName, Opts)

Performs an async HTTP request using the full request map.

await(Ref)

Equivalent to await/2.

await(Ref, Timeout)

Blocks until an async response for Ref arrives or the timeout expires. Only for buffered async requests: a streamed response (stream => true) arrives as a message sequence, which await does not understand -- receive those messages directly.

cancel(PoolName, Ref)

Cancels the async request identified by Ref (returned by async_get/2,3, async_req/2, etc.). The Ref routes directly to the worker holding the request; the pool argument is retained for API compatibility and not used.

check_opts(Opts)

Validates request options without performing the request. Cross-field rules are checked under the async API's rules (the superset): stream => true passes here but is additionally rejected by req/2 and the synchronous wrappers.

delete(PoolName, Url)

Equivalent to delete/3.

delete(PoolName, Url, Opts)

Performs an HTTP DELETE request.

doh_url_available()

Returns whether DNS-over-HTTPS is available (curl >= 7.62.0).

get(PoolName, Url)

Equivalent to get/3.

get(PoolName, Url, Opts)

Performs an HTTP GET request.

head(PoolName, Url)

Equivalent to head/3.

head(PoolName, Url, Opts)

Performs an HTTP HEAD request.

http3_available()

Returns whether HTTP/3 is available (curl >= 7.66.0).

options(PoolName, Url)

Equivalent to options/3.

options(PoolName, Url, Opts)

Performs an HTTP OPTIONS request.

patch(PoolName, Url)

Equivalent to patch/3.

patch(PoolName, Url, Opts)

Performs an HTTP PATCH request.

post(PoolName, Url)

Equivalent to post/3.

post(PoolName, Url, Opts)

Performs an HTTP POST request.

put(PoolName, Url)

Equivalent to put/3.

put(PoolName, Url, Opts)

Performs an HTTP PUT request.

req(PoolName, Opts)

Performs an HTTP request using the full request map.

sslkey_blob_available()

Returns whether SSL key blob is available (curl >= 7.71.0).

tcp_fastopen_available()

Returns whether TCP Fast Open is available (curl >= 7.49.0).

unix_socket_path_available()

Returns whether Unix socket paths are available (curl >= 7.40.0).

update_flow(PoolName, Ref, N)

Grants N more chunk-message credits to the streaming request Ref, which must have been started with a bounded stream_window. Routed like cancel/2 (directly to the owning worker; the pool argument is unused) and equally best-effort: granting credits to an unknown, completed, or unbounded-window request is a harmless no-op. Granting in batches (e.g. half the window at a time) amortizes the per-grant messaging.

Types

async_response()

-type async_response() :: {ok, reference()} | {error, #{code := error_code(), message := error_msg()}}.

body()

-type body() :: binary().

connecttimeout_ms()

-type connecttimeout_ms() :: pos_integer().

cookiejar()

-opaque cookiejar()

curlmopts()

-type curlmopts() ::
          [{pipelining, pipelining()} |
           {max_total_connections, non_neg_integer()} |
           {max_concurrent_streams, non_neg_integer()}].

curlopt_http_version()

-type curlopt_http_version() ::
          curl_http_version_none | curl_http_version_1_0 | curl_http_version_1_1 |
          curl_http_version_2_0 | curl_http_version_2tls | curl_http_version_2_prior_knowledge |
          curl_http_version_3.

curlopt_sslversion()

-type curlopt_sslversion() ::
          sslversion_default | sslversion_tlsv1 | sslversion_tlsv1_0 | sslversion_tlsv1_1 |
          sslversion_tlsv1_2 | sslversion_tlsv1_3.

doh_url()

-type doh_url() :: binary().

error_code()

-type error_code() ::
          ok | unsupported_protocol | failed_init | url_malformat | not_built_in |
          couldnt_resolve_proxy | couldnt_resolve_host | couldnt_connect | ftp_weird_server_reply |
          remote_access_denied | ftp_accept_failed | ftp_weird_pass_reply | ftp_accept_timeout |
          ftp_weird_pasv_reply | ftp_weird_227_format | ftp_cant_get_host | http2 |
          ftp_couldnt_set_type | partial_file | ftp_couldnt_retr_file | obsolete20 | quote_error |
          http_returned_error | write_error | obsolete24 | upload_failed | read_error | out_of_memory |
          operation_timedout | obsolete29 | ftp_port_failed | ftp_couldnt_use_rest | obsolete32 |
          range_error | http_post_error | ssl_connect_error | bad_download_resume |
          file_couldnt_read_file | ldap_cannot_bind | ldap_search_failed | obsolete40 |
          function_not_found | aborted_by_callback | bad_function_argument | obsolete44 |
          interface_failed | obsolete46 | too_many_redirects | unknown_option | telnet_option_syntax |
          obsolete50 | got_nothing | ssl_engine_notfound | ssl_engine_setfailed | send_error |
          recv_error | obsolete57 | ssl_certproblem | ssl_cipher | peer_failed_verification |
          bad_content_encoding | ldap_invalid_url | filesize_exceeded | use_ssl_failed |
          send_fail_rewind | ssl_engine_initfailed | login_denied | tftp_notfound | tftp_perm |
          remote_disk_full | tftp_illegal | tftp_unknownid | remote_file_exists | tftp_nosuchuser |
          conv_failed | conv_reqd | ssl_cacert_badfile | remote_file_not_found | ssh |
          ssl_shutdown_failed | again | ssl_crl_badfile | ssl_issuer_error | ftp_pret_failed |
          rtsp_cseq_error | rtsp_session_error | ftp_bad_file_list | chunk_failed |
          no_connection_available | obsolete16 | ssl_pinnedpubkeynotmatch | ssl_invalidcertstatus |
          http2_stream | recursive_api_call | auth_error | http3 | proxy | ssl_clientcert |
          quic_connect_error | unrecoverable_poll | too_large | ech_required | curl_last | bad_opts |
          await_timeout | worker_died | overload | admission_timeout.

error_msg()

-type error_msg() :: binary().

header()

-type header() :: {binary(), iodata()}.

headers()

-type headers() :: [header()].

http_auth()

-type http_auth() :: basic | digest | ntlm | negotiate.

interface()

-type interface() :: binary().

method()

-type method() :: get | post | put | head | options | patch | delete.

metrics()

-type metrics() :: proplists:proplist().

opts()

-type opts() ::
          #{reply_to => pid(),
            headers => headers(),
            cookiejar => cookiejar(),
            body => req_body(),
            connecttimeout_ms => connecttimeout_ms(),
            followlocation => boolean(),
            ssl_verifyhost => ssl_verifyhost(),
            ssl_verifypeer => ssl_verifypeer(),
            capath => binary(),
            cacert => binary(),
            timeout_ms => pos_integer(),
            maxredirs => -1 | non_neg_integer(),
            http_auth => http_auth(),
            username => binary(),
            password => binary(),
            proxy => proxy(),
            tcp_fastopen => tcp_fastopen(),
            interface => interface(),
            unix_socket_path => unix_socket_path(),
            doh_url => doh_url(),
            http_version => curlopt_http_version(),
            sslversion => curlopt_sslversion(),
            verbose => boolean(),
            sslcert => sslcert(),
            sslkey => sslkey(),
            sslkey_blob => sslkey_blob(),
            keypasswd => binary(),
            userpwd => userpwd(),
            dns_cache_timeout => integer(),
            ca_cache_timeout => integer(),
            pipewait => boolean(),
            stream => boolean(),
            stream_window => pos_integer() | infinity}.

pipelining()

-type pipelining() :: nothing | http1 | multiplex.

pool_opts()

-type pool_opts() ::
          [{pipelining, pipelining()} |
           {max_total_connections, non_neg_integer()} |
           {max_concurrent_streams, non_neg_integer()} |
           {max_in_flight, pos_integer() | infinity}].

proxy()

-type proxy() :: binary().

qs_vals()

-type qs_vals() :: [{unicode:chardata(), unicode:chardata() | true}].

req_body()

-type req_body() :: iodata() | qs_vals().

request()

-type request() ::
          #{url := binary(),
            method := method(),
            reply_to => pid(),
            headers => headers(),
            cookiejar => cookiejar(),
            body => req_body(),
            connecttimeout_ms => connecttimeout_ms(),
            followlocation => boolean(),
            ssl_verifyhost => ssl_verifyhost(),
            ssl_verifypeer => ssl_verifypeer(),
            capath => binary(),
            cacert => binary(),
            timeout_ms => pos_integer(),
            maxredirs => -1 | non_neg_integer(),
            http_auth => http_auth(),
            username => binary(),
            password => binary(),
            proxy => proxy(),
            tcp_fastopen => tcp_fastopen(),
            interface => interface(),
            unix_socket_path => unix_socket_path(),
            doh_url => doh_url(),
            http_version => curlopt_http_version(),
            sslversion => curlopt_sslversion(),
            verbose => boolean(),
            sslcert => sslcert(),
            sslkey => sslkey(),
            sslkey_blob => sslkey_blob(),
            keypasswd => binary(),
            userpwd => userpwd(),
            dns_cache_timeout => integer(),
            ca_cache_timeout => integer(),
            pipewait => boolean(),
            stream => boolean(),
            stream_window => pos_integer() | infinity}.

response()

-type response() ::
          {ok, #{status := status(), headers := headers(), cookiejar := cookiejar(), body := body()}} |
          {error, #{code := error_code(), message := error_msg()}}.

ssl_verifyhost()

-type ssl_verifyhost() :: boolean().

ssl_verifypeer()

-type ssl_verifypeer() :: boolean().

sslcert()

-type sslcert() :: binary().

sslkey()

-type sslkey() :: binary().

sslkey_blob()

-type sslkey_blob() :: binary().

status()

-type status() :: pos_integer().

tcp_fastopen()

-type tcp_fastopen() :: boolean().

unix_socket_path()

-type unix_socket_path() :: binary().

url()

-type url() :: binary().

userpwd()

-type userpwd() :: binary().

Functions

async_delete(PoolName, Url)

-spec async_delete(katipo_pool:name(), url()) -> async_response().

Equivalent to async_delete/3.

async_delete(PoolName, Url, Opts)

-spec async_delete(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP DELETE request. Returns {ok, Ref} immediately.

async_get(PoolName, Url)

-spec async_get(katipo_pool:name(), url()) -> async_response().

Equivalent to async_get/3.

async_get(PoolName, Url, Opts)

-spec async_get(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP GET request. Returns {ok, Ref} immediately. The response is delivered as a {katipo_response, Ref, Response} message.

async_head(PoolName, Url)

-spec async_head(katipo_pool:name(), url()) -> async_response().

Equivalent to async_head/3.

async_head(PoolName, Url, Opts)

-spec async_head(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP HEAD request. Returns {ok, Ref} immediately.

async_options(PoolName, Url)

-spec async_options(katipo_pool:name(), url()) -> async_response().

Equivalent to async_options/3.

async_options(PoolName, Url, Opts)

-spec async_options(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP OPTIONS request. Returns {ok, Ref} immediately.

async_patch(PoolName, Url)

-spec async_patch(katipo_pool:name(), url()) -> async_response().

Equivalent to async_patch/3.

async_patch(PoolName, Url, Opts)

-spec async_patch(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP PATCH request. Returns {ok, Ref} immediately.

async_post(PoolName, Url)

-spec async_post(katipo_pool:name(), url()) -> async_response().

Equivalent to async_post/3.

async_post(PoolName, Url, Opts)

-spec async_post(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP POST request. Returns {ok, Ref} immediately.

async_put(PoolName, Url)

-spec async_put(katipo_pool:name(), url()) -> async_response().

Equivalent to async_put/3.

async_put(PoolName, Url, Opts)

-spec async_put(katipo_pool:name(), url(), opts()) -> async_response().

Performs an async HTTP PUT request. Returns {ok, Ref} immediately.

async_req(PoolName, Opts)

-spec async_req(katipo_pool:name(), request()) -> async_response().

Performs an async HTTP request using the full request map.

Returns {ok, Ref} once a pool worker has accepted the request. The response is delivered as a {katipo_response, Ref, ResponseMap} or {katipo_error, Ref, ErrorMap} message to the process specified by the reply_to option (defaults to self()). If no worker could accept the request, returns {error, #{code => worker_died | overload | admission_timeout}} (dead/restarting worker, pool at max_in_flight capacity, or a wedged worker not answering the bounded admission call) and no message is delivered.

Use await/1,2 to block until the response arrives.

With stream => true the response body is delivered incrementally instead of as one buffered binary. The message flow is:

{katipo_headers, Ref, #{status := pos_integer(), headers := headers()}}
{katipo_chunk, Ref, binary()}     %% zero or more, in order
{katipo_done, Ref, #{status := pos_integer(), cookiejar := cookiejar()}}

A {katipo_error, Ref, ErrorMap} message is terminal and can arrive at any point, including after headers and chunks (e.g. a request timeout mid-body). cancel/2 works as for buffered async requests. await/1,2 does not apply to streamed requests; receive the messages directly. Streaming is only available through the async API -- req/2 and the synchronous wrappers reject stream => true.

By default chunks are delivered as fast as the transfer produces them. Pass stream_window => N to bound that: the transfer pauses (propagating backpressure to the server via TCP or the HTTP/2/3 stream window) once N chunk messages are outstanding, and update_flow/3 grants more. The request timer keeps running while a transfer is paused, so a consumer that stops granting credits eventually receives operation_timedout.

Caveat shared with buffered mode: when following redirects (followlocation => true), a redirect response that itself carries a body surfaces that body through the write path -- in streaming form the katipo_headers message may then describe the redirect response rather than the final one. Real-world redirect responses rarely carry bodies.

await(Ref)

-spec await(reference()) -> response().

Equivalent to await/2.

await(Ref, Timeout)

-spec await(reference(), timeout()) -> response().

Blocks until an async response for Ref arrives or the timeout expires. Only for buffered async requests: a streamed response (stream => true) arrives as a message sequence, which await does not understand -- receive those messages directly.

cancel(PoolName, Ref)

-spec cancel(katipo_pool:name(), reference()) -> ok.

Cancels the async request identified by Ref (returned by async_get/2,3, async_req/2, etc.). The Ref routes directly to the worker holding the request; the pool argument is retained for API compatibility and not used.

Best-effort: once the cancel takes effect no {katipo_response, Ref, _} or {katipo_error, Ref, _} message is delivered. A message that was already delivered before the cancel raced in may still be in the receiver's mailbox, so callers should be prepared to flush a late one. Cancelling an unknown or already-completed Ref is a harmless no-op.

check_opts(Opts)

-spec check_opts(request()) -> ok | {error, map()}.

Validates request options without performing the request. Cross-field rules are checked under the async API's rules (the superset): stream => true passes here but is additionally rejected by req/2 and the synchronous wrappers.

delete(PoolName, Url)

-spec delete(katipo_pool:name(), url()) -> response().

Equivalent to delete/3.

delete(PoolName, Url, Opts)

-spec delete(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP DELETE request.

doh_url_available()

Returns whether DNS-over-HTTPS is available (curl >= 7.62.0).

get(PoolName, Url)

-spec get(katipo_pool:name(), url()) -> response().

Equivalent to get/3.

get(PoolName, Url, Opts)

-spec get(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP GET request.

head(PoolName, Url)

-spec head(katipo_pool:name(), url()) -> response().

Equivalent to head/3.

head(PoolName, Url, Opts)

-spec head(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP HEAD request.

http3_available()

Returns whether HTTP/3 is available (curl >= 7.66.0).

options(PoolName, Url)

-spec options(katipo_pool:name(), url()) -> response().

Equivalent to options/3.

options(PoolName, Url, Opts)

-spec options(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP OPTIONS request.

patch(PoolName, Url)

-spec patch(katipo_pool:name(), url()) -> response().

Equivalent to patch/3.

patch(PoolName, Url, Opts)

-spec patch(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP PATCH request.

post(PoolName, Url)

-spec post(katipo_pool:name(), url()) -> response().

Equivalent to post/3.

post(PoolName, Url, Opts)

-spec post(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP POST request.

put(PoolName, Url)

-spec put(katipo_pool:name(), url()) -> response().

Equivalent to put/3.

put(PoolName, Url, Opts)

-spec put(katipo_pool:name(), url(), opts()) -> response().

Performs an HTTP PUT request.

req(PoolName, Opts)

-spec req(katipo_pool:name(), request()) -> response().

Performs an HTTP request using the full request map.

sslkey_blob_available()

Returns whether SSL key blob is available (curl >= 7.71.0).

tcp_fastopen_available()

Returns whether TCP Fast Open is available (curl >= 7.49.0).

unix_socket_path_available()

Returns whether Unix socket paths are available (curl >= 7.40.0).

update_flow(PoolName, Ref, N)

-spec update_flow(katipo_pool:name(), reference(), pos_integer()) -> ok.

Grants N more chunk-message credits to the streaming request Ref, which must have been started with a bounded stream_window. Routed like cancel/2 (directly to the owning worker; the pool argument is unused) and equally best-effort: granting credits to an unknown, completed, or unbounded-window request is a harmless no-op. Granting in batches (e.g. half the window at a time) amortizes the per-grant messaging.