leifg
I am currently trying to download a large file via HTTP streaming. That’s hard enough as the only examples I find usually involve a receive loop of some kind of receive loop. See the following example taken from this tutorial:
defp async_response(conn, id) do
:ok = :ibrowse.stream_next(id)
receive do
{:ibrowse_async_headers, ^id, '200', _headers} ->
conn = Plug.Conn.send_chunked(conn, 200)
# Here you might want to set proper headers to `conn`
# based on `headers` from a response.
async_response(conn, id)
{:ibrowse_async_headers, ^id, status_code, _headers} ->
{status_code_int, _} = :string.to_integer(status_code)
# If a service responded with an error, we still need to send
# this error to a client. Again, you might want to set
# proper headers based on response.
conn = Plug.Conn.send_chunked(conn, status_code_int)
async_response(conn, id)
{:ibrowse_async_response_timeout, ^id} ->
Plug.Conn.put_status(conn, 408)
{:error, :connection_closed_no_retry} ->
Plug.Conn.put_status(conn, 502)
{:ibrowse_async_response, ^id, data} ->
case Plug.Conn.chunk(conn, chunk) do
{:ok, conn} ->
async_response(conn, id)
{:error, :closed} ->
Logger.info "Client closed connection before receiving the last chunk"
conn
{:error, reason} ->
Logger.info "Unexpected error, reason: #{inspect(reason)}"
conn
end
{:ibrowse_async_response_end, ^id} ->
conn
end
end
Ideally I process the individual chunks further and use existing libraries for this.
So having a stream would help greatly.
So how do I convert the receive loop pattern into a stream. Or even better: is there an HTTP client which directly returns a stream after making a request?
Where Next?
Trending in Questions
Other Trending Topics
Latest on Elixir Forum
Sponsor Spotlight
Catch errors, track performance, monitor hosts and more.
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #library
- #deployment
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #elixirconf
- #channels
- #exunit
- #discussion
- #code-sync
- #podcasts
- #javascript
- #onsite
- #dialyzer
- #docker
- #authentication
- #ai
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #blog-post
- #elixirconf-us
- #elixir-ls
- #phoenix_html
- #iex
- #graphql
- #genstage
- #websockets
- #supervisor
- #advent-of-code
- #distillery
- #processes
- #elixirconf-eu
- #api
- #forms
- #security
- #metaprogramming
Our Sponsors
Build Elixir applications with speed and confidence.
Supporting innovation across the BEAM ecosystem.
We build reliable cloud platforms for business-critical systems.
Catch errors, track performance, monitor hosts and more.
Error tracking, logs, uptime, and Just Enough APM. Built for developers.
Master modern tools, write cleaner code, and tackle real-world challenges.
Producing high quality Elixir screencasts since 2017.
The deployment platform built for Elixir: PaaS ease, VPS control.
The team behind Membrane, Popcorn, LiveDebugger. Available for hire.
Courses that'll move you from confusion to "Aha, now I get it!"
Showing Posts 1 to 7- Show Best Posts
- Show All (oldest first)
- Show All (newest first)
chrismcg
Have a look at Stream.resource/3. I’m not up on current http client features so don’t know if any include what you’re looking for.
leifg
I know about Stream.resource but have no idea how I can reconclie the receive loop with this.
I’m wondering if anyone had a similar problem and has a working solution.
amarraja
I was going to try and do this, but my approach was wrong so I did a little digging and found this article:
Basically they create a process under a supervisor which uses the HTTPoison stream functionality, then setup stream resource to request the next chunks. There is a linked Github repo which looks fairly generic so you could just lift-and-shift
https://github.com/esl/flex
leifg
That looks exactly like what I am looking for, thank you so much!
peerreynders
The term stream is underspecified. The solution you are describing buffers data being pushed by HTTP onto the process which is captured via a receive loop so that it can wait in the buffer for a consumer to pull the data out of the buffer. The point being there are
The stream module furnishes pull streams: Streams are composable, lazy enumerables - i.e. a consumer drains data that is ready and waiting.
In functional programming available data is typically pushed by calling a function to process it (which is what the receive loop does) or in process oriented programming sent (forwarded) to the process that needs to process is (see GenStage and Flow) - so push streams are much less common.
amarraja
Thanks @peerreynders, that got me thinking. Unfortunately I read your reply a few times and I think I confused myself!
So how would the original problem look as a push stream?
I’m making the assumption that when a request is made to download a large file, the client just receives data as it comes over the wire.
Let’s say we used GenStage with a single producer and consumer. The http request would “push” data into the producer’s buffer and the consumer will “pull” when ready. I say “pull”, I know the data gets pushed, but the consumer still needs to indicate it can handle the demand - much the same way a stream will ask or wait for the next item.
Not sure I see the fundemental difference in this case.
peerreynders
That isn’t entirely accurate. The consumer simply communicates that it is ready to have a certain amount data pushed to it - it doesn’t actively pull data.
So the difference is:
GenServerwhich is a “client” than and for example aGenStageconsumer (which has one extremely narrowly defined responsibility anyway).GenStageuses “push with backpressure”, i.e. consumers are never blocked nor are they overwhelmed as the producer is supposed to limit the flow through theGenStagepipeline.So in my mind “pulling” always expresses the “risk” of being blocked.
It aways helps to know when process execution will be blocked given that code is strictly sequential inside any single process. There is nothing inherently “bad” about a process being blocked because its capability could be highly focused and there could be absolutely nothing else for it to do when it is blocked. But in certain capacities some
GenServerbased processes cannot afford to be blocked to function well, so that they are forced to “out source” any “blocking activities” to other secondary processes (GenServer docs: "handle_cast ... should be used sparingly" Why?).In environments (without lightweight processes) that rely on heavyweight threads push based streams like ReactiveX are much more common.
Inside a process, data is pushed by calling a function, between processes data is pushed by casting a message from producer to consumer (
callbeing used to acknowledge receipt of data by the consumer).