GenStageのcastやcallもdispatchするんですね
以前ふむふむ眺めただけだったので触ってみたら微ハマりした。
defmodule Producer do use GenStage ... def enqueue(item) do GenStage.cast(__MODULE__, {:enqueue, item}) end def handle_cast({:enqueue, item}, state) do {:noreply, [item], state} end end
こういうとき、Producer.enqueue(:something)
するとconsumerへ1つイベントが流れていく。
こうやって見ると普通だけど、consumerへイベントを流す指示ができるのはhandle_demand
だと思いこんでいて、最初は↓の実装をしてた。
defmodule Producer do use GenStage ... def enqueue(item) do GenStage.cast(__MODULE__, {:enqueue, item}) end def handle_cast({:enqueue, item}, queue) do {:noreply, [], :queue.in(item, queue)} end def handle_demand(demand, queue) do case :queue.out(queue) do {:empty, queue} -> {:noreply, [], queue} {{:value, item}, queue} -> {:noreply, [item], queue} end end