Batch Processing

Process items with controlled concurrency using `Task.async_stream`.

send_campaignPending

bulk_emailer.ex

defmodule MyApp.BulkEmailer do
  use Durable
  use Durable.Helpers

  workflow "send_campaign" do
    step :load, fn input ->
      recipients = Subscribers.active(input["campaign_id"])
      {:ok, %{campaign_id: input["campaign_id"], recipients: recipients}}
    end

    step :send_emails, fn ctx ->
      results =
        ctx.recipients
        |> Task.async_stream(
          fn recipient ->
            case Mailer.send_campaign(recipient, ctx.campaign_id) do
              :ok -> {:ok, recipient}
              {:error, reason} -> {:error, {recipient, reason}}
            end
          end,
          max_concurrency: 10,
          timeout: :infinity
        )
        |> Enum.map(fn {:ok, r} -> r end)

      sent = for {:ok, _} <- results, do: 1
      failed = for {:error, _} <- results, do: 1

      {:ok, ctx
      |> assign(:sent_count, length(sent))
      |> assign(:failed_count, length(failed))}
    end
  end
end

The code is the README example. The run is simulated in your browser, with time sped up.