Connecting APIs with Flow

This article explains how to build an API connector with Elixir’s Flow.

Many services need to consume events and act on them. A recent client needed an API connector that subscribed to AMQP events, mapped their data into the target API’s structure, and posted the result.

Given the project’s deadline and budget, I wanted predictable failures, useful error reports, and thorough test coverage. Elixir’s Flow handled that combination well.

The flow must go on

Most examples of Elixir flows focus on ad hoc use in scripts or the console. Flow.from_enumerable/1 connects a flow to an enumerable, while Enum.into/2 turns it into a straightforward parallel processing pipeline.

A continuously running application can instead use Flow.from_stage/1 and Flow.start_link/1. The processing pipeline looks like this:

defmodule Pipeline do

  def start_link() do
    pipeline = Flow.from_stage(Pipeline.Dispatcher, max_demand: 100, stages: 10)
               |> flow
    Flow.start_link(pipeline, name: __MODULE__)
  end

  def flow(input) do
    input
    |> put(:item, &Poison.decode/1)
    |> put(:transformed, &transform/1)
    |> put(:response, &post/1)
    |> Flow.each(&report/1)
  end

  def transform(%{item: item}) do
    # returns {:ok, transformed} or {:error, ...}
    TargetApi.object_from_item(item)
  end
  
  def post(%{transformed: object}) do
    # also returns {:ok, ...} or {:error, ...}
    TargetApi.post(object)
  end
  
  def report({:ok, _state}), do: Logger.info("posted successfully")
  
  def report({:error, key, error, state}) do
    Logger.error("processing errored at #{key}: #{inspect error}, #{inspect state}")
  end
  
  defp put(flow, key, fun) do
    Flow.map(flow, fn
      {:ok, state} ->
        case fun.(state) do
          {:ok, result} -> {:ok, Map.put(state, key, result)}
          error -> {:error, key, error, state}
        end
      pass -> pass
    end)
  end

end

This design provides two main benefits:

The pipeline is straightforward to test.
Failures retain enough context for useful error handling.

Testability

Passing the flow’s input as an argument makes the pipeline easy to isolate in tests. There is no need to connect or start the Pipeline.Dispatcher GenStage. In tests, I use Flow.from_enumerable/1 to provide input and append Enum.into/2 to collect the pipeline’s output.

My tests look like this:

defmodule PipelineTest do

  test "flow fails when provided invalid JSON" do
    assert {:error, :item, _error, _state} = run("not json")
  end

  def run(data) do
    [{:ok, data}]
    |> Flow.from_enumerable
    |> Pipeline.flow
    |> Enum.into([])
    |> List.first
  end
  
end

Because the pipeline mostly calls other modules, their respective test suites already cover most of its logic. As long as those functions honor the tagged :ok and :error return values, the pipeline can carry their results consistently into the error-handling stage.

Error handling

This flow is a sequence of state transformations. Each result receives its own key and can provide input for later actions. The resulting map is the protocol between functions that depend on previous output. Callers can look up successful values, while errors retain the stage at which they occurred.

Elixir offers several complementary approaches to failure. The “let it crash” philosophy relies on process isolation and supervision so that one unexpected failure does not corrupt unrelated processes. Tagged :ok and :error return values instead describe failures that callers are expected to handle directly.

Choosing between those approaches depends on the application’s users. Here, the first group consists of end users, who should experience as little disruption as possible. The pipeline runs after they interact with the front-facing application and feeds a notification system. They do not see its error output, although a failure may prevent the intended follow-up action.

The second group consists of the developers who maintain that front-facing application. When processing fails and users do not receive the intended result, those developers need to identify the failed stage quickly.

I therefore chose tagged :ok and :error results for every failure I expected. Each error identifies both what went wrong and where it happened. Unexpected crashes remain under supervision, where process isolation keeps the application running and allows failed parts to restart.

Dispatcher

A queueing GenStage provides the pipeline’s continuous input and dispatches events on demand. In this case, the AMQP consumer forwards incoming events to the dispatcher like this:

Dispatcher.async_push(Pipeline.Dispatcher, {:ok, data})

For completeness, here is the implementation of the queueing GenStage producer:

defmodule Dispatcher do
  use GenStage

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

  def init(_) do
    {:producer, {Queue.new, 0}, dispatcher: GenStage.DemandDispatcher}
  end

  def async_push(name, event) do
    GenStage.cast(name, {:push, event})
  end

  def queue_size(name) do
    GenStage.call(name, :queue_size)
  end

  def handle_call(:queue_size, _from, {queue, _demand} = state) do
    {:reply, Queue.size(queue), [], state}
  end

  def handle_cast({:push, event}, {queue, demand}) do
    dispatch_events(Queue.put_front(queue, event), demand, [])
  end

  def handle_demand(incoming_demand, {queue, demand}) when incoming_demand > 0 do
    dispatch_events(queue, demand + incoming_demand, [])
  end

  defp dispatch_events(queue, demand, events) do
    with true <- demand > 0,
      {event, remaining_queue} <- Queue.pop(queue)
    do
      dispatch_events(remaining_queue, demand - 1, [event | events])
    else
      _ -> {:noreply, Enum.reverse(events), {queue, demand}}
    end
  end

end