@bytemend/mfebus

In-memory pub-sub for Node. Synchronous same-process dispatch, plus pluggable transports so the bus can extend across child_process, worker_threads, MessagePort, or — when running in the browser — postMessage boundaries.

When the package is loaded inside a fork()-spawned child (process.send defined), it auto-binds to the parent channel. No setup needed for the common worker-pool pattern.

Install

npm i @bytemend/mfebus

Same-process

import { emit, on } from '@bytemend/mfebus';

const off = on('job:done', (payload, meta) => {
  console.log('job', payload.id, 'finished', meta.local ? '(local)' : `(from ${meta.senderId})`);
});

emit('job:done', { id: 42 });
off();

Node IPC — worker_threads

import { Worker } from 'node:worker_threads';
import { addIpcTarget, on } from '@bytemend/mfebus';

const worker = new Worker(new URL('./worker.js', import.meta.url));
const remove = addIpcTarget(worker);

on('worker:status', (s) => console.log('[main]', s));

// inside worker.js — `addIpcTarget(parentPort)` plus `emit('worker:status', …)`
// is enough to reach the listener above.

Node IPC — child_process.fork

import { fork } from 'node:child_process';
import { addIpcTarget } from '@bytemend/mfebus';

const child = fork('./child.js');
addIpcTarget(child);

// inside child.js the package auto-binds to `process` because
// `process.send` is defined under fork(). Just import and use.

Custom transport (Redis, NATS, etc.)

import { addTransport } from '@bytemend/mfebus';

addTransport({
  send: (envelope) => redis.publish('events', JSON.stringify(envelope)),
  onMessage: (cb) => {
    const onMsg = (_, raw) => cb(JSON.parse(raw));
    redis.on('message', onMsg);
    redis.subscribe('events');
    return () => redis.off('message', onMsg);
  },
});

Browser cross-frame (optional)

When loaded in a browser context, the bus also listens for window.postMessage. Register the peer window once:

import { addCrossFrameTarget, setAllowedOrigins } from '@bytemend/mfebus';

const frame = document.querySelector('#peer');
if (frame?.contentWindow) addCrossFrameTarget(frame.contentWindow);

// Tighten in production:
setAllowedOrigins(['https://peer.example.com']);

API

emit<T>(event: string, payload?: T): void
on<T>(event: string, handler: (p: T, meta) => void): () => void
once<T>(event: string, handler: (p: T, meta) => void): () => void
off(event: string, handler?: Handler): void
clear(event?: string): void

addTransport(transport: { send(env), onMessage?(cb) → off }): () => void
addIpcTarget(channel): () => void   // Worker | ChildProcess | MessagePort

addCrossFrameTarget(target): void
removeCrossFrameTarget(target): void
setAllowedOrigins(origins: string[]): void

senderId: string

Handler meta

type DispatchMeta = {
  senderId: string;        // random per-load id of the emitter
  local: boolean;          // came from the same process / frame?
  origin?: string;         // window origin (browser cross-frame only)
  ts?: number;             // posting timestamp (transport dispatches)
  transport?: unknown;     // handle of the transport that delivered the envelope
};

Notes

• Handlers run in registration order. Exceptions are caught and logged — one bad subscriber can't kill the bus.
• The bus drops envelopes whose senderId matches the local id, so transports that loop back are safe.
• setAllowedOrigins(['*']) is the dev default; tighten in production for browser frames.

License

MIT.