Flow-based parallel processing of an enumerable (or another Flow) through guarded ExternalService calls.

This is for the case where a guarded call is one stage of a larger Flow pipeline — partitioned, back-pressured processing with downstream map/filter/reduce stages. For a simple ordered, bounded-concurrency parallel map, prefer ExternalService.call_async_stream/5; Flow only earns its keep when you're building a pipeline.

{:flow, "~> 1.2"}

Example

[order1, order2, order3]
|> ExternalService.Flow.map(MyApp.Stripe, fn order ->
  case Stripe.charge(order) do
    {:error, %{status: s}} when s in 500..599 -> :retry
    other -> other
  end
end)
|> Flow.filter(&match?({:ok, _}, &1))
|> Enum.to_list()

map/5 accepts either an enumerable (which it turns into a Flow source) or an existing Flow, and returns a Flow so you can keep composing.

Semantics

Each element is processed with ExternalService.call/2,3, so retries, the circuit breaker, rate limiting, and telemetry behave exactly as they do for a direct call. A few consequences worth knowing:

• Errors are elements. Because call/3 returns structured errors rather than raising them, a failed element comes through the Flow as the {:error, %ExternalService.RetriesExhausted{}} / {:error, %CircuitBreakerOpen{}} / {:error, %ServiceNotStarted{}} tuple that call/3 returns — filter/partition on them downstream. (This module never uses call!, which would crash a Flow stage.)

• Unordered. Flow partitions reorder elements. If you need results in input order, use ExternalService.call_async_stream/5 instead.

• Rate-limit pacing. Throttling blocks the worker (it sleeps and re-checks), which in a Flow naturally back-pressures upstream. The rate-limit bucket is global per service, so the configured limit is honored across all stages. Because a sleeping call stalls the rest of its demand batch, a small :max_demand gives smoother pacing under a rate limit.