GenStage playground

glue_all_together.ex

import Tizio

{:ok, source} = Tizio.Source.start_link(%{:spam => "eggs"})
{:ok, consumer} = Tizio.Consumer.start_link(42)
{:ok, another_consumer} = Tizio.AnotherConsumer.start_link(42)

GenStage.sync_subscribe(consumer, to: source)
GenStage.sync_subscribe(another_consumer, to: source);

oneliner.ex

import Tizio; Tizio.Source.start_link(); Tizio.DBManager.start_link(); Tizio.TwitterManager.start_link(); Tizio.Source.broadcast_update(42);

tizio.ex

defmodule Tizio do
  defmodule Source do
    use GenStage

    def start_link(update) do
      GenStage.start_link(Tizio.Source, update)
    end

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

    def handle_demand(_demand, update) do
      { :noreply, [ %{ :update => update } ], update }
    end
  end

  defmodule Consumer do
    use GenStage
    
    def start_link(_unused) do
      GenStage.start_link(Tizio.Consumer, :ok)
    end

    def init(:ok) do
      { :consumer, :ok }
    end

    def handle_events(events, _from, state) do
      IO.puts "WRITE TO DB: #{IO.inspect Enum.at(events, 0).update.spam}"

      { :noreply, [], state }
    end
  end

  defmodule AnotherConsumer do
    use GenStage
    
    def start_link(_unused) do
      GenStage.start_link(Tizio.AnotherConsumer, :ok)
    end

    def init(:ok) do
      { :consumer, :ok }
    end

    def handle_events(events, _from, state) do
      IO.puts "SEND TO TWITTER: #{IO.inspect Enum.at(events, 0).update.spam}"

      { :noreply, [], state }
    end
  end
end