wojtekmach
commented
1 year ago Looking forward to this!
If you don't mind me reqing this thread, here's a proof of concept of using this branch: (thanks to the enhanced :finch_request option also done by @zachallaun :D)
Mix.install([
:req,
{:finch, github: "zachallaun/finch", branch: "async", override: true},
:bandit
])
defmodule MyPlug do
use Plug.Router
plug(:match)
plug(:dispatch)
get "/redirect" do
to = "/stream"
conn
|> Plug.Conn.put_resp_header("location", to)
|> Plug.Conn.send_resp(302, "redirecting to #{to}")
end
get "/stream" do
conn = Plug.Conn.send_chunked(conn, 200)
Enum.reduce_while(~w(a b c d), conn, fn chunk, conn ->
IO.inspect({:send, chunk})
case Plug.Conn.chunk(conn, chunk) do
{:ok, conn} ->
{:cont, conn}
{:error, :closed} ->
{:halt, conn}
end
end)
end
end
fun = fn request, finch_request, finch_name, finch_options ->
ref = Finch.async_request(finch_request, finch_name, finch_options)
receive do
{^ref, {:status, status}} ->
receive do
{^ref, {:headers, headers}} ->
body =
Stream.unfold(ref, fn ref ->
receive do
{^ref, {:data, data}} ->
IO.inspect({:recv, data})
{data, ref}
{^ref, :done} ->
nil
end
end)
{request, Req.Response.new(status: status, headers: headers, body: body)}
end
end
end
{:ok, _} = Bandit.start_link(plug: MyPlug, scheme: :http, port: 4000)
stream = Req.get!("http://localhost:4000/redirect", finch_request: fun).body
IO.inspect({:take, Enum.take(stream, 1)})
Process.sleep(1000)
IO.inspect({:take, Enum.take(stream, 10)})Outputs:
11:37:41.997 [info] Running MyPlug with Bandit 0.7.7 at 0.0.0.0:4000 (http)
11:37:42.057 [debug] follow_redirects: redirecting to /stream
{:send, "a"}
{:send, "b"}
{:send, "c"}
{:recv, "a"}
{:send, "d"}
{:take, ["a"]}
{:recv, "b"}
{:recv, "c"}
{:recv, "d"}
{:take, ["b", "c", "d"]}Again just a PoC, eventually this would be behind just a stream: true option or something. I'm curious if it's OK for Req to consume :status and :headers bits like that so it can return a response struct (which we can then pass through response steps)? Or that's undesirable/unsafe? I think I read somewhere that a server may return additional headers down the road. Not sure if that's true or how popular that is but it would break the model above.
Otherwise we'd go with some kind of Req.stream which is just a thin wrapper around this, users would need to handle :status, :headers, :data bits themselves, and we would NOT leverage response steps.
zachallaun
josevalim
sneako
hkrutzer
See #208 for context and past discussion.
Todo
Finch.HTTP2.Pool- fix significant duplication between:requestcall and:async_requestcast (fixed by 69d37d6)Finch.HTTP2.Pool- send errors that are currently sent as:gen_statemreplies incontinue_requests(fixed by 69d37d6)Finch.HTTP2.Pool- add tests for:connected_read_onlyand:disconnectedstateFinch.HTTP2.Pool- add tests for all handled error conditionsFinch.HTTP2.Pool- monitor caller process and cancel when caller exitsmainto ensure no performance regressions for sync requestsFinch.HTTP2.Pool-emit telemetry events(this will be addressed in a separate PR)To facilitate review, I'll try to list changes here that affect non-async requests as well so that they can be given careful consideration.
:errortuples that didn't match the public interface in certain cases, returning{:error, "read_only"}. In these cases, it now returns{:error, Finch.Error.exception(:read_only)}.:gen_statemunderlying the HTTP/2 pool no longer uses:replyactions to respond to callers, but:gen_statem.reply/2instead. This allows the same code-path either reply orsenddepending on whether the request was sync or async (see here).%RequestStream{}struct was accumulating state that was unrelated to streaming the request. This change simplifies the struct to only contain information required for streaming the body and stores it in a map of other state that the pool needs to maintain per request.