# `ExternalService.Flow`
[🔗](https://github.com/jvoegele/external_service/blob/main/lib/external_service/flow.ex#L5)

[`Flow`](https://hexdocs.pm/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.

> #### Optional dependency {: .info}
>
> `ExternalService.Flow` exists only when the optional
> [`:flow`](https://hex.pm/packages/flow) dependency is present. Add it to your
> application's deps to use this module:
>
> ```elixir
> {: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.

# `mapper`

```elixir
@type mapper() :: (term() -> ExternalService.retriable_function_result())
```

A function applied to each element, returning a retriable result.

# `source`

```elixir
@type source() :: Enumerable.t() | Flow.t()
```

An enumerable source or an existing `Flow` to continue.

# `map`

```elixir
@spec map(source(), ExternalService.service(), mapper()) :: Flow.t()
```

Maps each element of `source` through a guarded `ExternalService` call.

`source` is either an enumerable (used as a `Flow` source via
`Flow.from_enumerable/2`) or an existing `Flow` (whose stage configuration
already applies, so `flow_opts` are ignored in that case). Returns a `Flow`.

`retry_opts` are the same per-call retry options accepted by
`ExternalService.call/3` (a keyword list of overrides merged onto the service's
defaults, or a `t:ExternalService.RetryOptions.t/0` struct). `flow_opts` are
passed straight to `Flow.from_enumerable/2` (for example `:stages`,
`:min_demand`, `:max_demand`).

# `map`

```elixir
@spec map(
  source(),
  ExternalService.service(),
  ExternalService.RetryOptions.t() | keyword(),
  mapper()
) ::
  Flow.t()
```

# `map`

```elixir
@spec map(
  source(),
  ExternalService.service(),
  ExternalService.RetryOptions.t() | keyword(),
  mapper(),
  keyword()
) :: Flow.t()
```

---

*Consult [api-reference.md](api-reference.md) for complete listing*