Elixir GenStage PartitionDispatcher Example

genstage_partition_example.ex

#
# GenStage PartitionDispatcher Example
#
# reference: http://www.elixirfbp.org/2016/08/genstage-example-no-3-dispatching.html

defmodule Comsumer do
  use GenStage

  def start_link(state) do
    GenStage.start_link(__MODULE__, state)
  end
  def init(sleeping_time) do
    {:consumer, sleeping_time}
  end
  def handle_events(events, _from, sleeping_time) do
    IO.inspect(Comsumer: {self(), events |> Enum.map( &Integer.to_string/1 ), sleeping_time})
    Process.sleep(elem(sleeping_time,0))
    {:noreply, [], sleeping_time}
  end
end
defmodule Splitter do
  use GenStage

  def start_link(_) do
    GenStage.start_link(__MODULE__, :ok)
  end
  def init(:ok) do
    hash_fun = fn event ->
      {event, case rem(event, 3) do
                 0 -> :第一
                 1 -> :第二
                 2 -> :第三
               end
      } end
    {:producer_consumer, %{},
      dispatcher: {GenStage.PartitionDispatcher,
                    partitions: [:第一, :第二, :第三],
                    hash: hash_fun}
    }
  end
  def handle_events(events, _from, state) do
    # IO.inspect Splitter: state
    {:noreply, events, state}
  end
end
defmodule PartitionTest do
  def run do
    {:ok, import}    = GenStage.from_enumerable(1..10)
    {:ok, splitter}  = Splitter.start_link([])
    {:ok, c_first}   = Comsumer.start_link({2_000, :第一})
    {:ok, c_second}  = Comsumer.start_link({2_000, :第二})
    {:ok, c_third}   = Comsumer.start_link({2_000, :第三})

    GenStage.sync_subscribe(c_first,  to: splitter, partition: :第一, max_demand: 1)
    GenStage.sync_subscribe(c_second, to: splitter, partition: :第二, max_demand: 1)
    GenStage.sync_subscribe(c_third,  to: splitter, partition: :第三, max_demand: 1)
    GenStage.sync_subscribe(splitter, to: import, max_demand: 1)
  end
end