Async mediator

The mediator/async package (@mediator_async) is the async form of the mediator. A handler is an async function, so it can wait or do I/O through moonbitlang/async. The shape is the same as the core: typed values, and the compiler is the registry. There is no text key and no runtime lookup.

Import the package with an explicit alias, because its last path segment is async, the same as moonbitlang/async:

import {
  "moonrockz/moon-lander/mediator/async" @mediator_async,
  "moonbitlang/async",
}

In your package, moonbitlang/async keeps its default alias @async. The examples on this page run as tests of the package itself, where async is the name of the package, so they import moonbitlang/async as @aio. Write @async.sleep where an example has @aio.sleep.

Each mbt check block on this page runs as a test with mise run test on wasm, js, and native.

Send a request

@mediator_async.sender takes one async handler and gives a Sender[C, Req, Res] value. The handler takes the context value and the request, can wait, and gives the reply. Sender::send waits for the reply.

///|
struct Ping(String)

///|
async test "send a ping to a handler that waits" {
  let ping : @mediator_async.Sender[Unit, Ping, String] = @mediator_async.sender((
    _,
    req,
  ) => {
    @aio.sleep(10)
    let Ping(text) = req
    "Pong: " + text
  })
  assert_eq(ping.send((), Ping("Hello")), "Pong: Hello")
}

A sender calls its one handler one time for each send. An error that the handler raises passes out of send unchanged. The mediator does not turn an error into a reply.

Middleware

An async middleware wraps the next step of a send. It gets next, the context value, and the request. It can wait, run code before and after next, change the request or the reply, give a reply without a call to next, or catch the error of next. The helper names and the order rule are the same as in the core: Sender::use_ gives a new sender with one more middleware, and the first middleware that you attach is the outermost step.

The finally_ hook is a sync function, so it runs in a defer also when the caller cancels the send. A core @mediator.Middleware does not fit an async sender, because its next step is sync.

///|
async test "log middleware around an async send" {
  let log = []
  let ping : @mediator_async.Sender[Unit, Ping, String] = @mediator_async.sender((
    _,
    req,
  ) => {
    @aio.sleep(10)
    log.push("handler")
    let Ping(text) = req
    "Pong: " + text
  })
  let logged = ping
    .use_(
      @mediator_async.middleware((next, ctx, req) => {
        log.push("log in")
        let reply = next(ctx, req)
        log.push("log out")
        reply
      }),
    )
    .use_(@mediator_async.after((_, _, reply) => log.push("reply " + reply)))
    .use_(@mediator_async.finally_((_, _) => log.push("finally")))
  assert_eq(logged.send((), Ping("Hello")), "Pong: Hello")
  assert_eq(log, [
    "log in", "handler", "finally", "reply Pong: Hello", "log out",
  ])
}

Publish an event

@mediator_async.topic takes the first async event handler and gives a Topic[C, Ev] value. Topic::add gives a new topic with one more handler at the end, and Topic::use_ gives a new topic with one more middleware around the whole publish. A topic has two publish operations. Both follow the Foundatio.Mediator contract of the core: every handler runs, and a failing handler does not stop the others.

When one or more handlers raise, both operations raise one @mediator.PublishFailed(handlers~, failures~) after the last handler ends. The failures are in add order, also when the handlers of a concurrent publish end in another order. A topic middleware gets that one error from its next and can catch it.

No publish returns before its handlers end. There is no fire-and-forget publish, because a handler error must not be lost and moonbitlang/async lets no task outlive its task group. An application that wants background work starts its own task group.

///|
struct Joined(String)

///|
suberror AuditError {
  AuditClosed(String)
}

///|
async test "a concurrent publish runs every handler and lists the failure" {
  let log = []
  let joined : @mediator_async.Topic[Unit, Joined] = @mediator_async.topic((
    _,
    ev,
  ) => {
    @aio.sleep(50)
    let Joined(name) = ev
    log.push("welcome " + name)
  })
  let all = joined.add((_, ev) => {
    @aio.sleep(10)
    let Joined(name) = ev
    raise AuditClosed(name)
  })
  try all.publish_concurrent((), Joined("Ann")) catch {
    @mediator.PublishFailed(handlers~, failures~) as error => {
      assert_eq(handlers, 2)
      assert_eq(failures.length(), 1)
      assert_eq(failures[0].position, 2)
      assert_true(failures[0].error is AuditClosed("Ann"))
      assert_eq(
        error.message(),
        "Publish failed: 1 of 2 handlers raised an error (position 2)",
      )
    }
    other => fail("publish raised another error: \{other}")
  } noraise {
    _ => fail("publish did not raise")
  }
  assert_eq(log, ["welcome Ann"])
}

Cancellation and timeouts

A caller cancels a send or a publish through moonbitlang/async, for example with with_timeout_opt around the call, or when the task group of the caller ends. The mediator never catches, wraps, or counts a cancellation:

A cancellation is never a failure in PublishFailed. catch cannot capture a cancellation, so a middleware that catches every error does not stop it. The failures that a sequential publish held before the cancellation are not reported. A caller that needs them publishes without a cancellation scope around the publish. The finally_ hook runs also during a cancellation. A handler that reacts to its own cancellation with a protected cleanup, and raises an ordinary error from it, gives that error from send, as a direct call of the handler does. A publish does not start the next handler after a cancellation and does not raise PublishFailed for it.

A time limit that a handler sets for itself, for example with with_timeout, raises an ordinary error. send passes it out, and a publish lists it as a failure and runs the other handlers.

@mediator_async.timeout(after_ms) is a middleware that gives the next step after_ms milliseconds. When the step does not end in time, the middleware cancels it and raises @mediator.TimedOut(after_ms~). Its message is Request timed out after <after_ms> ms and holds no request values.

///|
async test "a timeout stops a slow send, and a caller cancellation is not an error" {
  let log = []
  let slow : @mediator_async.Sender[Unit, Ping, String] = @mediator_async.sender((
    _,
    req,
  ) => {
    @aio.sleep(1000)
    log.push("handler ended")
    let Ping(text) = req
    "Pong: " + text
  })
  let limited = slow.use_(@mediator_async.timeout(50))
  try limited.send((), Ping("Hello")) catch {
    @mediator.TimedOut(_) as error =>
      assert_eq(error.message(), "Request timed out after 50 ms")
    other => fail("send raised another error: \{other}")
  } noraise {
    reply => fail("send gave a reply: \{reply}")
  }
  let reply = @aio.with_timeout_opt(50, () => slow.send((), Ping("Hello")))
  assert_eq(reply, None)
  assert_eq(log, [])
}

The context value and the core mediator

Each send and each publish takes a context value of type C. The mediator gives that exact value to each middleware and to each handler, also after a wait, and does not keep it after the call. A consumer service, such as the Morphir workbench or moonrockz/smithy, puts its services in a record and gives that record as the context.

The async package reuses the core error type @mediator.MediatorError, and it works with core values in two ways:

After add on a wrapped core topic, the core publish counts as one handler at position 1, and its error is the failure at that position. A core @mediator.Middleware does not fit an async sender; use the async helpers. There is no way to call an async sender from sync code.

///|
struct Store {
  users : Array[String]
}

///|
struct Services {
  store : Store
  greeting : String
}

///|
struct Register(String)

///|
/// A sync handler that the service already has.
fn greet(services : Services, req : Register) -> String {
  let Register(name) = req
  services.greeting + " " + name
}

///|
async test "a service registers a user and greets through the core sender" {
  let greeter : @mediator.Sender[Services, Register, String] = @mediator.sender((
    services,
    req,
  ) => greet(services, req))
  let greet_async = @mediator_async.Sender::from_sync(greeter)
  let register : @mediator_async.Sender[Services, Register, String] = @mediator_async.sender((
    services,
    req,
  ) => {
    @aio.sleep(10)
    let Register(name) = req
    services.store.users.push(name)
    greet_async.send(services, req)
  })
  let direct : @mediator_async.Sender[Services, Register, String] = @mediator_async.sender((
    services,
    req,
  ) => greet(services, req))
  let services = { store: { users: [] }, greeting: "Welcome" }
  assert_eq(register.send(services, Register("Ann")), "Welcome Ann")
  assert_eq(direct.send(services, Register("Bo")), "Welcome Bo")
  assert_eq(services.store.users, ["Ann"])
}