rm-rf-etc
How to stream data from callback-based API?
I’m using WebSockex and Finch for a streaming client, and I want to emit every incoming message using an Elixir Stream, but how do I do that?
I imagine my code would work like this:
MyModule.new_client("wss://cool.feed.com/")
|> Stream.each(&IO.inspect/1)
I gather that new_client/1 needs to return a %Stream{} instance. It looks like Stream.resource/3 is the function to do this, but getting the next_fun to work with my client modules sounds complicated. Am I approaching this correctly? It seems I need my WebSockex callbacks to push every new message into a :queue, and my next_fun needs to call my WebSockex process to retrieve (and remove) the oldest message from the queue. If next_fun is called and the queue is empty (the next message hasn’t arrived yet), how do I complete this synchronous function call only after the next message has arrived?
Marked As Solved
rm-rf-etc
It looks like the solution is for next_fun to invoke GenServer.call/3 to get the oldest unhandled message from my WebSockex process, and – when the next message hasn’t yet arrived – to return {:noreply, new_state} from the handle_call/3 callback, and then call reply/2 once it has.
handle_call/3 takes a timeout value. Will my stream close if the call times out?
UPDATE: I had to add a GenServer for this to work. Originally I thought WebSockex implemented everything from GenServer, and that I could do all of this in my existing WebSockex server. But actually, WebSockex doesn’t implement GenServer.call/2, and so instead I had to add a GenServer to serve as a buffer, which I suppose is the correct architecture for this.
Last Post!
Aetherus
Today, I was searching for the answer, too, but can’t find any. Inspired by `Req.request(…, into: :self)`, I did it in an extremely unclean way. Assume the stream will be consumed in the same process creating it.
event_ref = make_ref()
parent_pid = self()
{child_pid, child_ref} =
spawn_monitor(fn ->
SomeModule.callback_api(fn event ->
send(parent_pid, {:my_event, event_ref, event})
end)
end)
stream = Stream.unfold(:whatever, fn _ ->
receive do
{:DOWN, ^child_ref, :process, ^child_pid, :normal} -> nil
{:my_event, ^event_ref, event} -> {event, :whatever}
end
end)
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
Move fast and fix things! Error tracking with less noise for Elixir devs who love to ship. Start your free dev account.
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
- #code-sync
- #javascript
- #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
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #hex
- #security
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 for Elixir devs who love to ship. Start your free account.
Practical resources that improve the lives of professional developers.
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!"