bus

In-process publish/subscribe that delivers each published message to its matching subscribers synchronously and in a defined order. Subscribers attach a (pattern, handler) registration; messages are matched by topic atom, head name, or an arity-1 boolean predicate. A re-entrant FIFO emission queue drains every message a handler publishes during dispatch before the next externally-published message, keeping delivery deterministic and replayable. The bus has no socket, thread, or timer of its own and introduces no capability — every effect is performed by the handlers it invokes.

bus:bus

[$bus:bus] -> element — Construct a fresh in-process bus with no subscriptions.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :ping [?fn ($m) $m]] $s
    [?let [= $d [$bus:emit $b [do :ping]]] $d@delivered]]]
          
            1

bus:close

[$bus:close] -> null — Close the bus, cancelling every active subscription — idempotent, never raises.

            [?lib 'cx-stdlib/bus']
[?let [= $b [$bus:bus]]
  [= $c [$bus:close $b]]
  [$bus:emit $b [do :t]]]
          
            cx-err:CXER4664

bus:on

[$bus:on] -> element — Register a handler to fire on every future message matching the pattern, returning a [subscription] handle.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :ping [?fn ($m) $m]] $s
    [?let [= $d [$bus:emit $b [do :ping]]] $d@delivered]]]
          
            1

bus:off

[$bus:off] -> bool — Cancel a subscription by handle or id — true if an active one was cancelled, false if already gone.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?let [= $s [$bus:on $b :t [?fn ($m) $m]]]
    [$bus:off $b $s]]]
          
            true

bus:emit

[$bus:emit] -> element — Publish a message and synchronously drive the full ordered dispatch, returning a [dispatch] summary value.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :ping [?fn ($m) $m]] $s
    [?let [= $d [$bus:emit $b [do :ping]]] $d@delivered]]]
          
            1

bus:subscribers

[$bus:subscribers] -> [sequence element] — List the active subscriptions in dispatch order, optionally filtered by a pattern.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :a [?fn ($m) $m]] $s1
    [?with-open [$bus:on $b :b [?fn ($m) $m]] $s2
      [$count [$bus:subscribers $b]]]]]
          
            2

bus:topics

[$bus:topics] -> [sequence element] — List the distinct topic atoms currently subscribed, in first-appearance order.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :a [?fn ($m) $m]] $s1
    [?with-open [$bus:on $b :b [?fn ($m) $m]] $s2
      [$bus:topics $b]]]]
          
            (:a, :b)

bus:match

[$bus:match] -> [sequence element] — Report which subscriptions would fire for a message, in fire order, without firing them.

            [?lib 'cx-stdlib/bus']
[?with-open [$bus:bus] $b
  [?with-open [$bus:on $b :t [?fn ($m) $m]] $s
    [$count [$bus:match $b [do :t]]]]]
          
            1

bus:topic

[$bus:topic] -> element — Compute a message's topic atom from its head and leading atom argument.

            [?lib 'cx-stdlib/bus']
[$bus:matches [do :refund] [?fn ($m) [= [$bus:topic $m] :refund]]]
          
            true

bus:matches

[$bus:matches] -> bool — Report whether a message matches a pattern — the bare matcher.

            [?lib 'cx-stdlib/bus']
[$bus:matches [do :order.placed] :order.placed]
          
            true