leifg
Convert receive loop into Stream
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?
Marked As Solved
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
Also Liked
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
- push streams and
- pull streams
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.
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:
- In a pull scenario the client can be blocked if no data is available - i.e. it isn’t going to get anything else done. This is more important for a
GenServerwhich is a “client” than and for example aGenStageconsumer (which has one extremely narrowly defined responsibility anyway). - In an unconstrained push scenario a client process can be overwhelmed with data being pushed to it, which usually leads to the process mailbox growing which will have undesirable side effects.
- This is why
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 GenServer based 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 (call being used to acknowledge receipt of data by the consumer).
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.
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
Deploy Phoenix to your own VPS with PaaS ease: push-to-deploy, zero downtime, auto SSL, managed Postgres. 14-day free trial at potions.io
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #javascript
- #code-sync
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #performance
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.
Real-time error tracking, performance insights, and observability for devs.
Practical resources that improve the lives of professional developers.
Producing high quality Elixir screencasts since 2017.
Enabling companies to succeed by building software people love.
Courses that'll move you from confusion to "Aha, now I get it!"