novox/mesh-tools
3
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

broker: split events (mesh.events) from tool RPC (mesh.rpc)

Browse Source 
Per ADR 0047: two topic exchanges kept apart, so a '#' subscription on
mesh.events is a clean audit of module/mesh/node events without the
request/reply tool traffic. Header-in-AMQP, persistent messages, per-
consumer durable dead-lettered queues and prefetch are the remaining
alignment to 0047.

Claude-Session: https://claude.ai/code/session_01LrgweAeERJYBg88c5cKDzF
This commit is contained in:
jschoubben
2026-09-03 23:50:56 +02:00
parent 9eddcdb0a0
commit 201a413f69
1 changed files with 11 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/broker-amqp.ts
+11 -6
View File
@@ -6,7 +6,11 @@ import amqp from "amqplib";
import { randomUUID } from "node:crypto";
import type { Broker, Envelope } from "@novox/mesh-sdk/messaging";
const EXCHANGE = "mesh.tools"; // one topic exchange carries tool invocations and events
// Two topic exchanges, kept apart on purpose: tool invocations are request/reply and are not
// events, so an audit sink subscribing to `#` on the events exchange sees module, mesh and node
// events — never the RPC traffic.
const RPC_EXCHANGE = "mesh.rpc";
const EVENTS_EXCHANGE = "mesh.events";
interface Reply {
result?: unknown;
@@ -17,7 +21,8 @@ interface Reply {
export async function connectAmqp(url: string): Promise<Broker> {
const conn = await amqp.connect(url);
const ch = await conn.createChannel();
await ch.assertExchange(EXCHANGE, "topic", { durable: true });
await ch.assertExchange(RPC_EXCHANGE, "topic", { durable: true });
await ch.assertExchange(EVENTS_EXCHANGE, "topic", { durable: true });
// Request/reply: one exclusive reply queue, correlationId → resolver.
const { queue: replyQueue } = await ch.assertQueue("", { exclusive: true });
@@ -48,7 +53,7 @@ export async function connectAmqp(url: string): Promise<Broker> {
else resolve(r.result as Res);
});
});
ch.publish(EXCHANGE, key, Buffer.from(JSON.stringify(body)), {
ch.publish(RPC_EXCHANGE, key, Buffer.from(JSON.stringify(body)), {
correlationId: id,
replyTo: replyQueue,
});
@@ -57,7 +62,7 @@ export async function connectAmqp(url: string): Promise<Broker> {
async handle<Req, Res>(key: string, handler: (body: Req) => Promise<Res>): Promise<() => void> {
const { queue } = await ch.assertQueue(`serve.${key}`, { durable: true });
await ch.bindQueue(queue, EXCHANGE, key);
await ch.bindQueue(queue, RPC_EXCHANGE, key);
const consumer = await ch.consume(queue, (msg) => {
if (!msg) return;
void (async () => {
@@ -79,12 +84,12 @@ export async function connectAmqp(url: string): Promise<Broker> {
},
async publish<T>(env: Envelope<T>): Promise<void> {
ch.publish(EXCHANGE, env.key, Buffer.from(JSON.stringify(env.body)));
ch.publish(EVENTS_EXCHANGE, env.key, Buffer.from(JSON.stringify(env.body)));
},
async subscribe<T>(pattern: string, handler: (env: Envelope<T>) => Promise<void>): Promise<() => void> {
const { queue } = await ch.assertQueue("", { exclusive: true });
await ch.bindQueue(queue, EXCHANGE, pattern);
await ch.bindQueue(queue, EVENTS_EXCHANGE, pattern);
const consumer = await ch.consume(queue, (msg) => {
if (!msg) return;
void (async () => {