dli
GenServer + DynamicSupervisor to track streaming workers per topic?
Hey there,
I am building a system that:
- streams data from an external HTTP API endpoint for multiple topics
- starts streaming processes under a supervisor
- allows runtime introspection which topics are currently streaming (optional: + which ones errored)
The external endpoint has a param for passing multiple topics (topic count per request is limited though). To stream additional topics, I need to make a new request.
Something like StreamerManager.start(topics) would launch one or more streamer processes for topics that are not streaming yet. Topics that are already streamed by some process are skipped. FWIW, I already implemented the Streamer + Consumers using GenStage but this is likely irrelevant to my question.
I am considering a GenServer that tracks %{:topic_name => pid()} in its state and calls DynamicSupervisor to launch streamers as child processes with the appropriate args. However, the PIDs will become invalid when a streamer crashes and is restarted by the supervisor (not sure if they even matter if I rely on a supervisor).
Am I off the track here or what would be the appropriate architecture here?
Marked As Solved
dli
Here is what I am using now (thanks @shanesveller for the inspiration):
-
Start Registry in supervisor tree:
{Registry, keys: :unique, name: MyApp.TopicRegistry} -
When starting a
:producer/ “TopicStreamer” GenStage withtopics:- fetch all keys from Registry:
Registry.select(MyApp.TopicRegistry, [{{:"$1", :_, :_}, [], [:"$1"]}]) |> List.flatten() |> MapSet.new() Enum.rejectall topics that exist in set of keysMyApp.TopicStreamer.start_link(name: {:via, TopicRegistry, topics})
- fetch all keys from Registry:
This ensures that there are no TopicStreamers with overlapping topics.
Also Liked
shanesveller
You sound like you may be imagining something very similar to Registry and via-tuples, will that save you some wheel reinvention? It automatically deregisters processes that terminate.
It is single host only, not distributed, but scales very well IME especially if you’re on a new enough Elixir version to activate the partitioning support.
shanesveller
I don’t see any concerns with this design off the top of my head, though in a standard GenServer I would possibly move the logic into a handle_continue instead of directly in init.
Popular in Questions
Other popular topics
Latest on Elixir Forum
Sponsor Spotlight
We build reliable cloud platforms for business-critical systems.
Categories:
Sub Categories:
Forums
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.
Practical resources that improve the lives of professional developers.
Enabling companies to succeed by building software people love.
The deployment platform built for Elixir: PaaS ease, VPS control.
Courses that'll move you from confusion to "Aha, now I get it!"