Building A continues data processing system

fetcher.ex

defmodule Archiver.Fetcher do
  use GenStage

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

  def init(state), do: {:producer, state}

  def handle_demand(demand, state) do
    items = Database.get_items()

    {:noreply, items, state}
  end
end

uploader.ex

defmodule Archiver.Uploader do
  use GenStage

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

  def init(_state) do
    {:consumer, :the_state_does_not_matter, subscribe_to: [Archiver.Fetcher]}
  end

  def handle_events(items, _from, _state) do
    items
    |> Enum.map(&Task.start_link(Dropbox, :upload, [&1]))

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