diff --git a/changelog.md b/changelog.md index 8af3e3d..623c64d 100644 --- a/changelog.md +++ b/changelog.md @@ -1,5 +1,15 @@ # 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.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 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, diff --git a/src/ehttpc.erl b/src/ehttpc.erl index bc2b110..5773c12 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_reply := {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}. @@ -482,6 +501,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 @@ -700,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, @@ -868,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), @@ -892,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 @@ -1004,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(_) ->