fly49

fly49

How to properly stop Producer -> ConsumerSupervisor?

I am quite new to Elixir and trying to understand GenStage flow.

I created a producer that emits a finite list of numbers

defmodule Producer do
  use GenStage

  def start_link(_) do
    GenStage.start_link(Producer, Enum.to_list(1..10), name: Producer)
  end

  def init(list) do
    {:producer, list}
  end

  def handle_demand(demand, state) when demand > 0 do
    case Enum.split(state, demand) do
      {[],[]} ->
        GenStage.async_info(self(), :stop)
        {:noreply, [], []}

      {pulled, remaining} ->
        {:noreply, pulled, remaining}
    end
  end

  def handle_info(:stop, state) do
    {:stop, :normal, state}
  end
end

and ConsumerSupervisor that starts new children for each number

defmodule ConSupervisor do
  use ConsumerSupervisor

  @opts [
    strategy: :one_for_one,
    subscribe_to: [{Producer, max_demand: 2, min_demand: 0}],
    max_restarts: 20
  ]

  def start_link(arg) do
    ConsumerSupervisor.start_link(__MODULE__, arg)
  end

  def init(_arg) do
    children = [%{id: Consumer, start: {Consumer, :start_link, []}, restart: :transient}]

    ConsumerSupervisor.init(children, @opts)
  end
end

Children are just tasks with 30% rate of success

defmodule Consumer do
  def start_link(event) do
    Task.start_link(fn ->
      Process.sleep(1000)

      if :rand.uniform(3) == 3 do
        IO.puts("task failed with #{event}")
        raise("ooops")
      end

      num = Kernel.inspect(event)
      IO.puts("Inserted #{num}")
    end)
  end
end

I want to terminate Producer the way that it can be assured that all Tasks are completed. For now when it terminates it emits cancel for ConsumerSupervisor that discard all Tasks retrying to finish successfully.

[info] GenStage consumer #PID<0.433.0> is stopping after receiving cancel from producer #PID<0.431.0> with reason: :normal

I feel like it should be tricky and require some consecutive cast/call messaging, but maybe there is more straightforward way to do this?

Marked As Solved

NeutronStein

NeutronStein

As @Ljzn mentioned you can check active workers with count_children and stop if there are none active.

  def handle_info(:stop, state) do
    %{active: active} = ConsumerSupervisor.count_children(consumer_pid_or_registered_name)
    if active > 0 do
      Process.send_after(self(), :stop, 5000)
      {:noreply, [], state}
    else
      {:stop, :normal, state}
    end
  end

Also Liked

Ljzn

Ljzn

I’m not very familiar with genState. Maybe you can use ConsumerSupervisor.count_children/1 repeatedly to check is all tasks completed.

Where Next?

Popular in Questions Top

JDanielMartinez
Hi! May someone helps me, please! I have two apps into an umbrella project: the first one is Database, which manages queries, and the se...
New
sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
itssasanka
Hi all, Trying to get some more clarity over utc_datetime and naive_datetime for Ecto: https://hexdocs.pm/ecto/Ecto.Schema.html#module-...
New
ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New
ovidiubadita
Hey all, I discovered Elixir and I love it. I always wanted to learn a functional programming and I intended to go for Haskell, but afte...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
Mooodi
Given a string, how can I get access to its character by index? Enum.at("my_string", 2) doesn't work. Or rather, not char, but a substr...
New
jay1
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New
idi527
I’ve been re-reading swift book again and noticed that multiline strings there don’t have a trailing line break, unlike in elixir iex(2)...
New

Other popular topics Top

dotdotdotPaul
Okay, I'm having a heck of a time trying to figure out how to best handle the validation of belongs_to associations in Ecto. I'm sure I'...
New
Tee
can someone please explain to me how Enum.reduce works with maps
New
sergio
I couldn’t find any guides that worked well with Phoenix 1.6.0 and esbuild. I hope this helps people test the waters and eases you into t...
New
hariharasudhan94
I would like to know what is the best IDE for elixir development?
New
grych
Hi folks, Few months ago I have announced the proof-of-concept of the library to manipulate the browsers DOM objects directly from Elixi...
639 49522 488
New
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
ashish173
I am using Ecto timestamps with postgres, I can see the timestamps() use the :naive_dateime but for my use case I wanted to store the ti...
New
alice
Hey, Just curious what are the main benefits of Elixir compared to Clojure? When is Elixir more useful than Clojure and vice versa? Th...
New
Qqwy
Original source of discussion: This topic on the Pragmatic Programmers' Functional Web Development with Elixir, OTP, and Phoenix forum. ...
New

We're in Beta

About us Mission Statement