Anonymized producer/consumer for triggering events from a controller

feed_item_flow.ex

defmodule FeedItemProducer do
  use GenStage

  def start_link(:nameless) do
    GenStage.start_link(__MODULE__, :ok)
  end

  def start_link() do
    GenStage.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    {:producer, [], [buffer_size: 50_000]}
  end

  def notify(item = %FeedItem{}, pid \\ __MODULE__) do
    GenStage.cast(pid, {:notify, item})
  end

  def handle_cast({:notify, item}, state) do
    {:noreply, [%{item: item, at: Utils.Time.unix_now()}], state}
  end

  def handle_demand(demand, keys) when demand > 0 do
    {:noreply, [], keys}
  end
end

defmodule FeedItemConsumer do
  use ConsumerSupervisor

  def start_link() do
    ConsumerSupervisor.start_link(__MODULE__, :ok, name: __MODULE__)
  end

  def init(:ok) do
    children = [
      worker(FeedItemServer, [], restart: :transient)
    ]

    # max concurrency is 3
    opts = [strategy: :one_for_one, subscribe_to: [{FeedItemProducer, max_demand: 3, min_demand: 1}]]
    ConsumerSupervisor.init(children, opts)
  end
end

defmodule FeedItemServer do
  def start_link(%{item: event_feed_item, at: unix_inserted_at}) do
    Task.start_link(fn ->
      FeedItem.notify(event_feed_item)
    end)
  end
end

# start the consumer & producer in your Application supervisor, as per the GenStage docs

# You can call FeedItemProducer.notify(item) at any time, from a controller or elsewhere