From ab7fa763a87baf1fca1924f0ce9386c2dd27794a Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:58:25 -0300 Subject: [PATCH 1/6] feat: allow choosing http2 as protocol Despite the field name and typespec claiming it to be a list of protocols, it only actually works with a singleton... https://github.com/emqx/gun/blob/5dff085cd14923fb77eda9849accd3d7b9bd8ff7/src/gun.erl#L1074 --- src/ehttpc.erl | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/ehttpc.erl b/src/ehttpc.erl index bc2b110..aec6d00 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -482,6 +482,8 @@ gun_opts([{retry_timeout, _} | Opts], Acc) -> gun_opts(Opts, Acc); gun_opts([{connect_timeout, ConnectTimeout} | Opts], Acc) -> gun_opts(Opts, Acc#{connect_timeout => ConnectTimeout}); +gun_opts([{protocols, Protocols} | Opts], Acc) -> + gun_opts(Opts, Acc#{protocols => Protocols}); gun_opts([{transport, Transport} | Opts0], Acc0) -> Acc1 = Acc0#{transport => Transport}, case lists:keytake(transport_opts, 1, Opts0) of From f6e0adc7ecccb678fe4596a87ec10e414923a5b1 Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Thu, 16 Jul 2026 11:59:26 -0300 Subject: [PATCH 2/6] feat: allow cancelling inflight streams --- src/ehttpc.erl | 54 ++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 52 insertions(+), 2 deletions(-) diff --git a/src/ehttpc.erl b/src/ehttpc.erl index aec6d00..6afc04c 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -30,6 +30,7 @@ request/4, request/5, request_async/5, + cancel/2, workers/1, health_check/2, check_pool_integrity/1, @@ -64,7 +65,15 @@ -type path() :: binary() | string(). -type headers() :: [{binary(), iodata()}]. -type body() :: iodata(). --type callback() :: {function(), list()}. +-type callback() :: + {function(), list()} + | #{ + %% where to send the final results + final_result := {function(), list()}, + %% optional; sends the gun stream ref and the worker pid so the caller may cancel + %% it. + stream_ref => {function(), list()} + }. -type request() :: path() | {path(), headers()} | {path(), headers(), body()}. -include_lib("snabbkaffe/include/snabbkaffe.hrl"). @@ -92,6 +101,8 @@ (is_binary(element(3, REQ)) orelse is_list(element(3, REQ)))) ). +-record(cancel, {stream_ref :: reference()}). + -record(state, { pool :: term(), id :: pos_integer(), @@ -211,7 +222,8 @@ mk_request(put = Method, Req, ExpireAt) when ?IS_BODY_REQ(Req) -> mk_request(delete = Method, Req, ExpireAt) when ?IS_HEADERS_REQ(Req) -> ?REQ(Method, Req, ExpireAt). -%% @doc Send an async request. The callback is evaluated when an error happens or http response is received. +%% @doc Send an async request. The callback is evaluated when an error happens or http +%% response is received. -spec request_async(pid(), method(), request(), timeout(), callback()) -> ok. request_async(Worker, Method, Request, Timeout, ResultCallback) when is_pid(Worker) -> ExpireAt = fresh_expire_at(Timeout), @@ -233,6 +245,9 @@ mk_async_request(put = Method, Req, ExpireAt, RC) when ?IS_BODY_REQ(Req) -> mk_async_request(delete = Method, Req, ExpireAt, RC) when ?IS_HEADERS_REQ(Req) -> ?ASYNC_REQ(Method, Req, ExpireAt, RC). +cancel(Worker, StreamRef) -> + gen_server:cast(Worker, #cancel{stream_ref = StreamRef}). + workers(Pool) -> gproc_pool:active_workers(name(Pool)). @@ -315,6 +330,10 @@ handle_call(Call, _From, State0) -> State = maybe_shoot(State0), {reply, {error, {unexpected_call, Call}}, State}. +handle_cast(#cancel{stream_ref = StreamRef}, State0) -> + State1 = handle_cancel_stream(State0, StreamRef), + State = maybe_shoot(State1), + {noreply, State}; handle_cast(_Msg, State0) -> State = maybe_shoot(State0), {noreply, State}. @@ -702,6 +721,9 @@ drop_expired(#{pending := Pending, pending_count := PC} = Requests, Now) -> maybe_reply_timeout({F, A}) when is_function(F) -> _ = erlang:apply(F, A ++ [{error, timeout}]), ok; +maybe_reply_timeout(#{final_reply := {F, A}}) when is_function(F) -> + _ = erlang:apply(F, A ++ [{error, timeout}]), + ok; maybe_reply_timeout(_) -> %% This is not a callback, but the gen_server:call's From %% The caller should have alreay given up waiting for a reply, @@ -870,6 +892,7 @@ shoot( } ) when is_pid(Client) -> StreamRef = do_request(Client, Method, Request, TunnelRef), + maybe_send_stream_ref(ReplyTo, StreamRef), ?tp(shot, #{from => ReplyTo, req => Request, reqs => Requests}), %% no need for the payload Req = ?SENT_REQ(ReplyTo, ExpireAt, ?undef), @@ -894,6 +917,22 @@ do_after_gun_up(State0 = #state{client = Client}, ExpireAt, Fun) -> {reply, {error, Reason}, State#state{client = ?undef}} end. +handle_cancel_stream(#state{requests = Requests0} = State0, StreamRef) -> + do_cancel_gun_stream(State0, StreamRef), + case take_sent_req(StreamRef, Requests0) of + error -> + State0; + {_, Requests} -> + %% caller explicitly requested the cancel; no need to reply to it. + State0#state{requests = Requests} + end. + +do_cancel_gun_stream(#state{client = Client}, StreamRef) when is_pid(Client) -> + _ = gun:cancel(Client, StreamRef), + ok; +do_cancel_gun_stream(_State, _StreamRef) -> + ok. + %% This is a copy of gun:await_up/3 %% with the '$gen_call' clause added so the calls in the mail box %% are collected into the queue in time @@ -1006,9 +1045,20 @@ handle_gun_reply(State, Client, StreamRef, IsFin, StatusCode, Headers, Data) -> reply({F, A}, Result) when is_function(F) -> _ = erlang:apply(F, A ++ [Result]), ok; +reply(#{final_reply := FinalReply}, Result) -> + %% assert + {F, A} = FinalReply, + _ = erlang:apply(F, A ++ [Result]), + ok; reply(From, Result) -> gen_server:reply(From, Result). +maybe_send_stream_ref(#{stream_ref := {F, A}}, StreamRef) when is_function(F) -> + _ = erlang:apply(F, [StreamRef, self() | A]), + ok; +maybe_send_stream_ref(_ReplyTo, _StreamRef) -> + ok. + peek_oldest_fn(#{prioritise_latest := true}) -> {fun queue:peek_r/1, fun queue:out_r/1}; peek_oldest_fn(_) -> From 6798cff0ed52e500a62273b56caf65cde95797f0 Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Thu, 16 Jul 2026 14:09:14 -0300 Subject: [PATCH 3/6] chore: bump app vsn --- src/ehttpc.app.src | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ehttpc.app.src b/src/ehttpc.app.src index b7e5237..7fda509 100644 --- a/src/ehttpc.app.src +++ b/src/ehttpc.app.src @@ -1,6 +1,6 @@ {application, ehttpc, [ {description, "HTTP Client for Erlang/OTP"}, - {vsn, "0.7.3"}, + {vsn, "0.7.5"}, {registered, []}, {applications, [ kernel, From f8c3c734e059e5619b1e675290bd366f4fe67aed Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Thu, 16 Jul 2026 14:22:05 -0300 Subject: [PATCH 4/6] docs: add changelog --- changelog.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/changelog.md b/changelog.md index 8af3e3d..5294ead 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,10 @@ # ehttpc changes +## 0.7.5 + +- Added the possibility of choosing HTTP2 as the protocol. +- Added `cancel/2` and new allowed callback API to enable cancelling HTTP2 streams. + ## 0.7.3 - Previously, we had a fixed restart intensity for the worker supervisor, meaning that if a From 3949f87712057d40c854e99bd0c82a6f6b4a4402 Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:26:29 -0300 Subject: [PATCH 5/6] fix: typespec --- src/ehttpc.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ehttpc.erl b/src/ehttpc.erl index 6afc04c..5773c12 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -69,7 +69,7 @@ {function(), list()} | #{ %% where to send the final results - final_result := {function(), list()}, + final_reply := {function(), list()}, %% optional; sends the gun stream ref and the worker pid so the caller may cancel %% it. stream_ref => {function(), list()} From 64e60c6bfa776b2330f4ffecc958f5732dbc43b7 Mon Sep 17 00:00:00 2001 From: Thales Macedo Garitezi <16166434+thalesmg@users.noreply.github.com> Date: Fri, 17 Jul 2026 09:26:36 -0300 Subject: [PATCH 6/6] doc: add missing entry --- changelog.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/changelog.md b/changelog.md index 5294ead..623c64d 100644 --- a/changelog.md +++ b/changelog.md @@ -5,6 +5,11 @@ - Added the possibility of choosing HTTP2 as the protocol. - Added `cancel/2` and new allowed callback API to enable cancelling HTTP2 streams. +## 0.7.4 + +- Fixed the supervision tree structure. Previously, some processes could reach maximum + restart intensity and never be restarted nor propagate the crash up the tree. + ## 0.7.3 - Previously, we had a fixed restart intensity for the worker supervisor, meaning that if a