GenStage.from_enumerable зависает с прерывистым потоком

Question

У меня есть поток, который не производит данные так быстро, как потребляется.

Итак, у меня есть продюсер, который определяется так:

def start_link() do
  create_stream
  |> GenStage.from_enumerable(name: Producer)
end

Тогда мой производитель-потребитель подписывается на него

  def init(:ok) do
    {:producer_consumer, :the_state_does_not_matter, subscribe_to: [Producer]}
  end

А мой потребитель подписывается на мю производителя-потребителя

  def init(:ok) do
    {:consumer, :the_state_does_not_matter, subscribe_to: [ProducerConsumer]}
  end

Проблема, с которой я сталкиваюсь, заключается в зависании потребителя, я думаю, потому что в какой-то момент производителю не удалось получить новые данные и, как указано в документации:

Когда перечислимое заканчивается или останавливается, сцена выйдет с: нормальной причиной. Это означает, что, если потребитель подписывается на перечисляемую стадию, а для параметра: cancel установлено значение: постоянное (по умолчанию), потребитель также завершит работу с: обычной причиной

Поэтому я читаю больше, и он предлагает добавить опцию cancel:: transient не закончить сцену. Я добавил это так, но это не работает, я что-то упустил?

|> GenStage.from_enumerable(name: Producer, cancel: :transient)

Первоначально я использовал Flow.into_stages(flow, [ProducerConsumer]) но я не могу сделать это, потому что я не могу ссылаться (или я не знаю, как) на ProducerConsumer из моего дерева супервизоров

children = [
  {Producer, []},
  {ProducerConsumer, []},
  {Consumer, []}
]

Обновить

Обновление передаваемой ссылки на Flow.into_stages из дочернего определения

children = [
  {Producer, [name: ProducerConsumer]},
  {ProducerConsumer, []},
  {Consumer, []}
]

def start_link(producer_consumer) do
  create_stream
  |> Flow.into_stages(producer_consumer)
end

** (Mix) Не удалось запустить тест приложения: Application.start(:normal, []) возвратил ошибку: shutdown: не удалось запустить child: Producer ** (EXIT) завершился в: GenServer.call({:name, ProducerConsumer}, {:"$subscribe", nil, #PID<0.2031.0>, [cancel::transient]}, 5000) ** (EXIT) нет связи с Elixir.ProducerConsumer

Answer 1

Ошибка:

** (Mix) Не удалось запустить тест приложения: Application.start(:normal, []) возвратил ошибку: shutdown: не удалось запустить child: Producer ** (EXIT) завершился в: GenServer.call({:name, ProducerConsumer}, {:"$subscribe", nil, #PID<0.2031.0>, [cancel::transient]}, 5000) ** (EXIT) нет связи с Elixir.ProducerConsumer

Просто означает, что когда Flow.into_stages пытается синхронизировать с предоставленным потребителем, этот потребитель уже должен быть запущен.

Итак, порядок важен при контроле, примерно так:

children = [
  Consumer,
  FlowProducerWorker # worker which implements Flow.into_stages(flow, [Consumer])
]