GenStage sync and async

sync_or_async.exs

events = 1..5

defmodule SyncOrAsync do
  use GenStage

  def start_link(mode) when mode == :async or mode == :sync do
    GenStage.start_link(__MODULE__, mode, name: __MODULE__)
  end

  # GenStage API

  def init(mode), do: {:producer_consumer, mode}

  def handle_events(events, _from, state) do
    case state do
      :sync ->
        {:noreply, events, state}

      :async ->
        async_op(events)
        {:noreply, [], state}
    end
  end

  def handle_info({:event, event, stop}, state) do
    if stop do
      parent = self()
      spawn(fn -> GenStage.stop(parent) end)
    end

    {:noreply, [event], state}
  end

  def async_op(events) do
    parent = self()
    last = List.last(events)

    Enum.each(events, fn event ->
      IO.puts("Working on #{event}")
      spawn(fn -> send(parent, {:event, event, last == event}) end)
    end)
  end
end

# Sync

{:ok, producer} = GenStage.from_enumerable(events)
{:ok, sync} = SyncOrAsync.start_link(:sync)

GenStage.sync_subscribe(sync, to: producer)

[{sync, cancel: :transient}]
|> GenStage.stream()
|> Enum.to_list()
|> IO.inspect()

# Async

{:ok, producer} = GenStage.from_enumerable(events)
{:ok, async} = SyncOrAsync.start_link(:async)

# Omiting `cancel: :transient` when subscribing prevents `SyncOrAsync` to handle
# any events, as it will be stopped before the async work is done.
GenStage.sync_subscribe(async, to: producer, cancel: :transient)

[{async, cancel: :transient}]
|> GenStage.stream()
|> Enum.to_list()
|> IO.inspect()