fabric

Platform-level eventing over the shipped primitives: one subscribe/emit surface with an explicit durability axis. The durable plane rides journal — publish appends an attributed event to a named stream (per-stream total order, hash-chained), subscribe/receive is a batched pull filtered by the bus pattern vocabulary, and ack commits a consumer group's offset cumulatively as store data, giving at-least-once delivery with redelivery of the uncommitted tail; a group may carry a redelivery policy (max-deliveries + dlq) that dead-letters a poison head-of-tail event to an ordinary stream and unblocks the group. The transient plane is latest-wins in-process channels — no history, no replay, no ack — and carries the request-reply call convention: respond registers an arity-1 callable answering a channel (sticky-exclusive), request is a blocking deadline-bounded call, serve pumps a remote responder. fabric introduces no capability of its own; persistence authority is the underlying journal/store's.

fabric:open

[$fabric:open] -> element — Open a fabric, returning a [fabric] handle — dual-form: a [journal] element composes the embedded tier over it (the journal stays caller-owned); an xsp://host:port URL string dials a fabric-serve daemon and attaches via XSP-AUTH (opts.tenant required; opts.did+seed = the proven identity, absent = anonymous; net-gated at the dial).

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  $f@state]

            'open'

fabric:close

[$fabric:close] -> null — Close the fabric: cancel subscriptions, drop transient channels — idempotent; the journal is untouched.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $c [$fabric:close $f]]
  [$fabric:read $f "acme/coord/map"]]

            cx-err:CXER4921

fabric:publish

[$fabric:publish] -> element — Append an attributed event to a named durable stream, returning a [receipt seq=N stream=…] — the journal append is the sequencing.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:publish $f "orders" [do :order.placed] {actor: "u" authority: "g"}]]
  $r@seq]

fabric:subscribe

[$fabric:subscribe] -> element — Register a pull subscription on a durable stream — pattern-filtered; with opts.group it resumes from the group's committed offset. Group subscriptions may declare a redelivery policy (opts.max-deliveries + opts.dlq, together or not at all): the policy persists as group state, later group subscriptions inherit it, a conflicting redeclaration refuses, and a head-of-tail event redelivered past the limit is published to the dlq stream as a [dead-letter] envelope and auto-committed past — the group unblocks, the original entry stays in the source journal.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:publish $f "orders" [do :order.placed] {actor: "u" authority: "g"}]]
  [= $s [$fabric:subscribe $f "orders" :order.*]]
  [$count [$fabric:receive $s]]]

fabric:observe

[$fabric:observe] -> element — Register a read-only subscription — receive works, ack refuses (the wire-tap shape; no offsets, no group).

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:publish $f "orders" [do :order.placed] {actor: "u" authority: "g"}]]
  [= $s [$fabric:observe $f "orders" :order.placed]]
  [$fabric:ack $s 1]]

            cx-err:CXER4922

fabric:receive

[$fabric:receive] -> [sequence element] — Batched pull: return the pattern-matching [entry] elements committed since the cursor (up to opts.max), advancing the cursor — never blocks; empty node-set when nothing.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:publish $f "orders" [do :order.placed] {actor: "u" authority: "g"}]]
  [= $s [$fabric:subscribe $f "orders" "do"]]
  [$count [$fabric:receive $s]]]

fabric:ack

[$fabric:ack] -> null — Commit the group's offset cumulatively through seq, persisted as store data — a later group subscription resumes past it; the uncommitted tail redelivers.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:publish $f "orders" [do :order.placed] {actor: "u" authority: "g"}]]
  [= $s1 [$fabric:subscribe $f "orders" :order.placed {group: "g1"}]]
  [= $e1 [$fabric:receive $s1]]
  [= $a [$fabric:ack $s1 1]]
  [= $s2 [$fabric:subscribe $f "orders" :order.placed {group: "g1"}]]
  [$count [$fabric:receive $s2]]]

fabric:emit

[$fabric:emit] -> null — Publish a value on a transient channel — latest-wins, no history, no replay, out of the durable audit.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $e1 [$fabric:emit $f "acme/coord/map" [viewport zoom=12]]]
  [= $e2 [$fabric:emit $f "acme/coord/map" [viewport zoom=13]]]
  [$fabric:read $f "acme/coord/map"]]

            [viewport zoom=13]

fabric:read

[$fabric:read] -> any — Read a transient channel's latest value — the empty node-set when the channel was never published (absence, never null).

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $e [$fabric:emit $f "acme/coord/map" [viewport zoom=12]]]
  [$fabric:read $f "acme/coord/map"]]

            [viewport zoom=12]

fabric:respond

[$fabric:respond] -> element — Register an arity-1 callable answering a request-reply channel, returning a [fabric-responder …] — sticky-exclusive per channel (a second respond while the holder lives refuses); the callable never travels: on the remote tier only the registration crosses the wire and the serve pump applies it here.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:respond $f "acme/svc/echo" [?fn ($v) $v]]]
  $r@channel]

            'acme/svc/echo'

fabric:request

[$fabric:request] -> any — Blocking request-reply call on a channel: the reply is the responder callable's return value; a responder err (raised or returned) travels verbatim; no live responder refuses immediately; opts.deadline bounds the remote wait (ms, default 10000). Embedded requests are answered synchronously at the call site.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:respond $f "acme/svc/sum" [?fn ($v) [+ $v/a $v/b]]]]
  [$fabric:request $f "acme/svc/sum" [args [a 2] [b 3]]]]

fabric:serve

[$fabric:serve] -> int — The remote responder's pump (pull is the primitive): drain pushed request frames, apply the registered callable, send the replies — returns the count of calls answered (opts: deadline ms, max). Embedded calls are answered at the request site, so the embedded pump reports 0 by construction.

            [?lib 'cx-stdlib/journal']
[?lib 'cx-fabric' :as fabric]
[?let [= $j [$journal:open "mem://t" "acme"]]
  [= $f [$fabric:open $j]]
  [= $r [$fabric:respond $f "acme/svc/echo" [?fn ($v) $v]]]
  [$fabric:serve $r]]