defmodule OneFlight do @moduledoc """ Collapses concurrent duplicate calls into one supervised execution per key. `one_flight` is not a cache. It only shares work that is already in progress; after a flight completes, the result is discarded. ## Example iex> {:ok, _pid} = start_supervised({OneFlight, name: ExampleFlights}) iex> OneFlight.run(:key, fn -> :value end, name: ExampleFlights) {:ok, :value} """ @type key :: term() @type option :: {:name, GenServer.server()} | {:timeout, timeout()} | {:on_abandoned, :cancel | :complete} | {:scope, :node} @doc false def child_spec(opts) do id = Keyword.get(opts, :name, __MODULE__) %{id: id, start: {OneFlight.Supervisor, :start_link, [opts]}} end @doc false def start_link(opts), do: OneFlight.Supervisor.start_link(opts) @doc """ Runs `fun` once for concurrent callers using the same key. """ @spec run(key(), (-> term()), [option()]) :: {:ok, term()} | {:error, OneFlight.Error.t()} def run(key, fun, opts \\ []) when is_function(fun, 0) do with :ok <- validate_scope(Keyword.get(opts, :scope, :node)) do timeout = Keyword.get(opts, :timeout, 5_000) key |> async(fun, opts) |> await(timeout) end end @doc """ Runs `fun` once and unwraps a successful result. Raises `OneFlight.Error` when the non-bang API would return an error tuple. """ @spec run!(key(), (-> term()), [option()]) :: term() def run!(key, fun, opts \\ []) when is_function(fun, 0) do case run(key, fun, opts) do {:ok, result} -> result {:error, error} -> raise error end end @doc """ Starts a flight and returns a handle that can be awaited. """ @spec async(key(), (-> term()), [option()]) :: OneFlight.Flight.t() def async(key, fun, opts \\ []) when is_function(fun, 0) do name = Keyword.get(opts, :name, __MODULE__) coordinator = OneFlight.Supervisor.coordinator_name(name) {:wait, ref} = OneFlight.Coordinator.run(coordinator, key, fun, opts) %OneFlight.Flight{ref: ref, coordinator: coordinator} end @doc """ Waits for an async flight result. """ @spec await(OneFlight.Flight.t(), timeout()) :: {:ok, term()} | {:error, OneFlight.Error.t()} def await(%OneFlight.Flight{ref: ref, coordinator: coordinator}, timeout \\ 5_000) do receive do {^ref, result} -> result after timeout -> OneFlight.Coordinator.waiter_timed_out(coordinator, ref) {:error, OneFlight.Error.timeout()} end end @doc """ Detaches `key` so the next caller starts a fresh flight. """ @spec forget(key(), [option()]) :: :ok def forget(key, opts \\ []) do name = Keyword.get(opts, :name, __MODULE__) OneFlight.Coordinator.forget(OneFlight.Supervisor.coordinator_name(name), key) end @doc """ Returns local in-flight stats for `key`. """ @spec stats(key(), [option()]) :: %{in_flight?: boolean(), waiters_count: non_neg_integer()} def stats(key, opts \\ []) do name = Keyword.get(opts, :name, __MODULE__) OneFlight.Coordinator.stats(OneFlight.Supervisor.coordinator_name(name), key) end defp validate_scope(:node), do: :ok defp validate_scope(:cluster) do {:error, OneFlight.Error.invalid_option("scope: :cluster is not supported in one_flight v0.1")} end end