English | 简体中文

async is an Erlang library for composing asynchronous request/reply workflows as monadic values. A promise may emit zero or more progress messages and must eventually emit one final reply. Workflows can be sequenced, transformed, observed from an OTP process, or aggregated concurrently with a bounded number of active jobs.

The public convenience API is async_m, an identity-specialized version of the generic async_t monad transformer.

• Compose asynchronous operations with ErLando do notation.
• Consume multiple {message, Message} events followed by one final reply.
• Propagate {error, Reason} replies through monadic chains.
• Apply per-promise or whole-wait timeouts.
• Aggregate lists or maps of promises with optional concurrency limits.
• Carry application state and callback-local state through continuations.
• Integrate pending promises into gen_server-style handle_info/2 loops.
• Create promises for gen_server, gen_fsm, RPC, and raw monitored calls.
• Use async_t over another inner monad when async_m is too specialized.

The application version in this repository is 0.6.0.

The project does not currently declare a minimum Erlang/OTP version. Its supervision tree uses map child specifications and ordinary one_for_one supervisors for both static and dynamic children. Applications should compile and test it against their chosen OTP release.

Add the Git repository to rebar.config:

{deps, [
    {async, {git, "https://github.com/slepher/async.git", {tag, "0.6.0"}}}
]}.

Then compile:

rebar3 compile

Start the application explicitly when using its supervised workers or channels:

ok = case async:start() of
         {ok, _Apps} -> ok;
         {error, {already_started, async}} -> ok
     end.

In an OTP release, adding async to the consuming application's applications list is normally preferable.

Include ErLando's do notation and build a workflow:

-module(example).
-export([run/0]).

-include_lib("erlando/include/do.hrl").

run() ->
    Workflow =
        do([async_m ||
               async_m:promise_sleep(25),
               return(done)
           ]),
    async_m:wait(Workflow).

The result is:

{ok, done}

async_m:return/1 creates a normal monadic success, while async_m:pure_return/1 injects an already-formed reply value without adding an {ok, ...} wrapper:

async_m:wait(async_m:return(value)).
%% => {ok, value}

async_m:wait(async_m:pure_return(value)).
%% => value

Use pure_return/1 mainly when implementing protocol-level combinators or when the unwrapped reply shape is intentional.

An asynchronous source is identified by a reference-like value. The supported identifiers are references, integers, and binaries. Replies delivered to the waiting process use these shapes:

{message, Ref, Message}  %% zero or more non-final messages
{Ref, FinalReply}        %% exactly one final reply
{'DOWN', Ref, _, _, Reason}

async_t exposes replies to user callbacks as:

{message, Message}
{ok, Value}
{error, Reason}
Value

The central invariant is:

A promise may emit any number of messages, but it must emit exactly one final reply.

Final replies remove the registered callback. Messages leave it registered so the same promise can continue producing events.

Wrap an existing request reference:

Ref = async_gen_server:call(Server, Request),
Promise = async_m:promise(Ref).

Or defer creation of the reference until the promise starts:

Promise =
    async_m:promise(
      fun() ->
          async_gen_server:call(Server, Request)
      end).

If the zero-arity action returns a reference, integer, or binary, the promise waits for replies associated with that identifier. Any other returned value is treated as an immediate final value.

A server can send progress through the standard From value and finish with a normal reply:

handle_call(work, From, State) ->
    async:message(From, started),
    async:message(From, halfway),
    {reply, {ok, finished}, State}.

Consume messages with wait_t/2:

Result =
    async_m:wait_t(
      async_gen_server:promise_call(Server, work),
      #{callback =>
            fun({message, Message}) ->
                    io:format("progress: ~p~n", [Message]),
                    ok;
               (FinalReply) ->
                    FinalReply
            end}).

Promises participate in the standard ErLando monad operations:

Workflow =
    do([async_m ||
           User <- async_gen_server:promise_call(user_server, get_user),
           Orders <- async_gen_server:promise_call(
                       order_server, {orders_for, User}),
           return({User, Orders})
       ]).

An {error, Reason} reply short-circuits the remaining monadic steps:

Workflow =
    do([async_m ||
           _ <- async_m:fail(invalid_request),
           return(unreachable)
       ]),

{error, invalid_request} = async_m:wait(Workflow).

Use monad_error:catch_error/2 to recover:

Recovered =
    monad_error:catch_error(
      async_m:fail(not_found),
      fun(not_found) -> async_m:return(default_value) end).

map_promises is the recommended aggregation primitive when every input promise contributes one final result.

List output preserves input position:

Delayed =
    fun(Milliseconds, Value) ->
        do([async_m ||
               async_m:promise_sleep(Milliseconds),
               async_m:pure_return(Value)
           ])
    end,

Promises = [
    Delayed(30, first),
    Delayed(10, second),
    Delayed(20, third)
],

[first, second, third] =
    async_m:wait(async_m:map_promises(Promises)).

Map output preserves keys:

Promises = #{
    profile => async_gen_server:promise_call(profile_server, UserId),
    settings => async_gen_server:promise_call(settings_server, UserId)
},

#{profile := Profile, settings := Settings} =
    async_m:wait(async_m:map_promises(Promises)).

Messages from map values are tagged with their key:

{message, {profile, Progress}}

The list form uses internal numeric keys and removes them before forwarding messages to the caller.

Pass limit to cap active work:

Result =
    async_m:wait(
      async_m:map_promises(
        Promises,
        #{limit => 4})).

limit => 0 is the default and means no limit. Positive limits use an internal Working/Pending/Completed lifecycle:

1. Up to limit keys enter Working.
2. Remaining keys stay Pending.
3. A Working promise may emit multiple messages.
4. Its final-reply callback must finish before the result enters Completed.
5. The freed Working slot starts one Pending promise.
6. The aggregate returns only after every initial working chain has drained.

The final-reply callback is part of the work unit. If a custom callback is asynchronous, its Working slot remains occupied until the callback finishes. Different Working chains still run concurrently.

The map form accepts:

#{
    limit => non_neg_integer(),
    acc0 => InitialAccumulator,
    cc => fun(Key, Reply) -> AsyncT end
}

The default callback:

• forwards {message, Message} as {message, {Key, Message}};
• stores final replies in a map under Key.

A custom callback is invoked for both messages and final replies and must return an async_t value. It may be asynchronous; map_promises waits for it before advancing that Working chain.

Promise = async_m:promise(RequestRef, 1000).

Or with a transport adapter:

Promise = async_gen_server:promise_call(Server, Request, 1000).

If the timer wins, the promise receives:

{error, timeout}

Result = async_m:wait_t(Workflow, #{timeout => 5000}).

When this timeout expires, all callbacks still registered in the current state are driven with {error, timeout}. This differs from assigning an independent timer when each promise is created.

async_m exposes two related state layers:

• get_state/0, put_state/1, modify_state/1 operate on the complete runtime state while preserving registered callbacks.
• get_local/0, put_local/1, modify_local/1 operate on data associated with the current continuation reference.

Example:

Workflow =
    do([async_m ||
           async_m:put_local([]),
           Value <- async_m:promise(RequestRef),
           async_m:modify_local(fun(Values) -> [Value | Values] end),
           async_m:get_local()
       ]).

local_ref/2 and local/2 provide scoped access to another local context. The lower-level find_ref/1, get_ref/2, put_ref/2, and remove_ref/1 functions operate directly on the callback/reference store.

The simplest runner blocks until no registered callbacks remain:

async_m:wait(Promise).

wait_t/2 accepts:

#{
    callback => Callback,
    cc => AsyncRContinuation,
    state => InitialState,
    offset => CallbackMapTupleIndex,
    timeout => MillisecondsOrInfinity
}

callback may have arity 0, 1, or 2. An arity-2 callback receives Callback(Reply, State). When it returns the same state shape, that value becomes the next state; returning another shape completes with that value.

For custom tuple/record state, offset identifies the tuple element holding the callback map:

-record(state, {callbacks = #{}, events = []}).

async_m:wait_t(
  Promise,
  #{
      state => #state{},
      offset => #state.callbacks,
      callback =>
          fun({message, Event}, #state{events = Events} = State) ->
                  State#state{events = [Event | Events]};
             (Final, #state{events = Events}) ->
                  {Final, lists:reverse(Events)}
          end
  }).

wait/1 is convenient at process boundaries but blocks the caller. To keep an OTP server responsive, register the workflow with exec/4 and route incoming messages through handle_info/3.

-record(state, {callbacks = #{}, value}).

start_request(Promise, State) ->
    async_m:exec(
      Promise,
      fun({message, Progress}, S) ->
              io:format("~p~n", [Progress]),
              S;
         (Reply, S) ->
              S#state{value = Reply}
      end,
      #state.callbacks,
      State).

handle_info(Info, State) ->
    case async_m:handle_info(Info, #state.callbacks, State) of
        unhandled ->
            {noreply, State};
        NewState when is_record(NewState, state) ->
            {noreply, NewState}
    end.

async_gen_server:call(Server, Request).
async_gen_server:promise_call(Server, Request).
async_gen_server:promise_call(Server, Request, Timeout).

async_gen_fsm:promise_sync_send_event(Process, Event, Timeout).
async_gen_fsm:promise_sync_send_all_state_event(Process, Event, Timeout).

async_rpc:promise_call/4 calls a remote function. If the remote function returns an async_t, the helper forwards its messages and final reply back to the caller as a local promise.

async_rpc:promise_call(Node, Module, Function, Args).

async_channel provides a supervised request queue with a pool-size limit:

{ok, _Pid} = async_channel:start(my_channel, 8),
Promise =
    async_gen_server:promise_channel_call(
      my_channel, Server, Request, Timeout).

Ref = async:call(Process, Label, Request),
Promise = async:promise_mref(Ref, Timeout).

Process may be a PID, local registered name, {global, Name}, or {Name, Node}.

The generic transformer stack is:

AsyncT s r m a
  = ReplyT Message Error
      (ContT r
        (AsyncRT s m)) a

AsyncRT s m a
  = StateT s
      (ReaderT Reference
        (ReaderT CallbacksGS m)) a

Responsibilities:

Use async_m unless you specifically need to choose another inner monad. When using async_t directly, pass the monad descriptor explicitly:

AT = async_t:new(identity),
Promise = async_t:promise(RequestRef, AT),
ResultMonad = async_t:wait(Promise, AT),
Result = identity:run(ResultMonad).

• lift_reply/1 exposes messages and final replies as values to a higher reply layer.
• lift_final_reply/1 exposes only final replies as values; messages continue outward.
• with_message/2, with_final/2, and with_all/2 attach handlers at different reply boundaries.
• handle_message/2 consumes messages with a callback/continuation.
• pass/0 deliberately returns from the runtime layer without producing a final async_t reply.
• par/1 is low level: only one branch may produce a final reply; other branches must emit messages or use pass/0.
• progn_par/1 follows that low-level rule and returns the last branch's value.
• map_promises/1,2 should be preferred for normal all-results aggregation.

Violating the single-final-reply invariant can invoke a continuation more than once, leave internal counters behind, or complete an aggregate too early.

rebar3 compile
rebar3 ct
rebar3 xref
rebar3 dialyzer

The test suites cover transformer state, promise chaining, errors, messages, timeouts, asynchronous accumulation, bounded concurrency, and historical API versions.

For Docker-based compatibility testing across multiple Erlang/OTP versions:

.\ci_scripts\sync_ci.ps1
.\ci_scripts\build.ps1
.\ci_scripts\run.ps1 -NoView

ci_scripts/sync_ci.ps1 first runs rebar3 get-deps, then resolves the selected astranaut dependency through _build/default. A fetched dependency is read from lib/astranaut; a local checkout is discovered through the checkouts/astranaut/src link, preserving uncommitted checkout changes. This repository owns ci_scripts/ci-env.conf.example; the ignored ci_scripts/ci-env.conf contains local overrides. The sync script never replaces these project-specific configuration files. Re-run it when the upstream CI file set changes.

On Bash, use bash ./ci_scripts/sync_ci.sh instead.

On modern OTP releases, Dialyzer may report legacy opaque/generated-code warnings that are independent of compilation and Common Test results.

BSD 3-Clause License. See LICENSE.