One Event System, Browser to Message Broker

A few years ago I was working at a marketing SaaS company building whitelabel mobile apps. React Native + web. The job was analytics tracking — capturing user behavior across different surfaces and routing events to various destinations.

I needed a cross-platform event emitter. EventTarget technically works everywhere, but it felt like a hack — string-only events, no type safety, no pattern matching. And I needed pattern matching badly. When your event names look like analytics:screen:home, analytics:tap:cta:signup, analytics:scroll:pricing, you don't want to register 40 individual listeners. You want /^analytics:/.

observer.on(/^analytics:/, ({ event, data }) => {
    // catches everything in the analytics namespace
    sendToMixpanel(event, data)
})

That worked. But then I hit the real problem: I had no idea what was happening. Events would silently not fire, or fire twice, or listeners would leak, and I'd spend hours adding console.log everywhere trying to figure out what was wired wrong.

Observability first

Event-driven code has a reputation for being hard to debug, and it's earned. The indirection that makes events powerful is the same indirection that makes them invisible. You wire things up, something doesn't fire, and you're left grepping through code hoping to find the disconnect. I didn't want a debugger for events — I wanted observability built into the system itself.

And thus spy() was born:

const observer = new ObserverEngine<AppEvents>({
    spy: (action) => {
        // every .on(), .off(), .emit() — all visible
        // action.fn, action.event, action.data, action.context
        console.log(`${action.context.name} → ${action.fn}(${String(action.event)})`)
    }
})

// Or introspect at any point
observer.$has('user:login')   // are there listeners?
observer.$facts()             // listener counts, regex counts
observer.$internals()         // full internal state, cloned and safe

No more guessing. You just look. Every .on(), .off(), .emit() passes through the spy. You can pipe it to your telemetry, your logger, or just console.log it during development. The point is that nothing is hidden from you. If an event fires and nobody's listening, you know immediately.

Platform-agnostic by design

I was using this in React, but I deliberately kept React out of the core. I write a lot of Node.js servers, processing scripts, and ETL pipelines. I wanted the same event system everywhere — browser, server, mobile, scripts. The moment you couple an event system to a rendering framework, you lose it everywhere else. React can consume the observer just fine from the outside. It doesn't need to be inside it.

This turned out to be one of the better design decisions I've made with this library. The same ObserverEngine instance that powers analytics tracking in a React Native app is the same one managing queue processing on a Node server. Same API, same types, same debugging tools. No mental context switching between platforms.

The evolution

As JS matured and my utilities grew, I kept adding what I needed and what I thought would be cool to use and JS-standards-esque (eg: AbortController).

AbortSignal support — just like EventEmitter, I can now do on('event', handler, { signal }) on the frontend too. Works with AbortSignal.timeout(). This alone solved an entire class of listener-leak bugs in component lifecycles.

Async generators — for await (const data of observer.on('event')) with internal FIFO buffering so nothing drops while you're doing async work between iterations. This is the pattern I reach for most in processing pipelines.

Event promises — const data = await observer.once('ready') lets you await a single event with cleanup built in. Useful for initialization sequences where you need to wait for something to be ready before proceeding.

Event queues — concurrency control, rate limiting, backpressure, priority ordering. You create a queue from any event and the observer manages the processing pipeline. This replaced a lot of hand-rolled queue logic I had scattered across projects.

Component observation — observer.observe(anyObject) extends any plain object with event capabilities. You get .on(), .off(), .emit(), and .once() grafted onto whatever you pass in, with its own isolated event space.

The design problem behind ObserverRelay

I've wanted to split the observer across a network boundary for a long time. The use case was always obvious: you have services that need to communicate via events, and you want the same ergonomics — pattern matching, queues, generators, spy — regardless of whether the event came from the same process or a different machine.

The problem was never motivation. It was figuring out how to do it without leaking transport concerns into the abstraction.

The questions that kept me stuck: How do you handle ack and nack abstractly? Different transports have completely different acknowledgment semantics — RabbitMQ has explicit ack/nack on the channel, Kafka has offset commits, SQS has receipt handle deletion, Redis Pub/Sub has no acknowledgment at all. If you bake any of those into the core, you've coupled yourself to a transport. If you ignore them, you've built a toy.

Dead letter queues were another one. Some transports handle DLQ at the infrastructure level, others expect your application to manage it. Where does that responsibility live in an abstract emitter?

The answer I landed on was a context object. When a message arrives from the transport, the subclass provides a ctx alongside the event data. That context carries whatever the transport needs — ack(), nack(), metadata, headers, whatever. The relay doesn't know or care what's in it. It just passes it through to your listeners.

// The type signature tells the whole story
type RelayEvents<TEvents, TCtx> = {
    [K in keyof TEvents]: { data: TEvents[K]; ctx: TCtx }
}

Emitting is pure data — your code doesn't think about transports. Receiving is data plus context — your code handles transport concerns exactly where it should, at the consumption point.

ObserverRelay in practice

ObserverRelay is an abstract class with two internal ObserverEngine instances — one for publishing (outbound), one for subscribing (inbound). You subclass it, implement send() for outbound events, and call receive() when the transport delivers inbound messages. Everything else — pattern matching, queues, generators, spy — works across the boundary automatically.

Same process — WorkerThreads

I'm using this right now for parallel processing with worker threads. The parent and worker share the same event API:

class ThreadRelay extends ObserverRelay<TaskEvents, ThreadCtx> {

    #port: MessagePort | Worker

    constructor(port: MessagePort | Worker) {

        super({ name: 'thread' })
        this.#port = port

        port.on('message', (msg) => {

            this.receive(msg.event, msg.data, { port })
        })
    }

    protected send(event: string, data: unknown) {

        this.#port.postMessage({ event, data })
    }
}

// parent.ts
const worker = new Worker('./processor.js')
const relay = new ThreadRelay(worker)

relay.emit('task:run', { id: '123', payload: rawData })

// Queue results with concurrency control
relay.queue('task:result', async ({ data }) => {

    await saveResult(data)
}, { concurrency: 3, name: 'result-writer' })

// Or consume as an async stream
for await (const { data } of relay.on('task:progress')) {

    updateProgressBar(data.percent)
}

// processor.ts (worker)
const relay = new ThreadRelay(parentPort!)

relay.on('task:run', ({ data }) => {

    const result = heavyComputation(data.payload)
    relay.emit('task:result', { id: data.id, result })
})

Across the network — RabbitMQ

Same concept, but now you're horizontally scaling. The subclass wires the transport, and the rest of your code doesn't care whether the event came from the same process or a different continent:

class AmqpRelay extends ObserverRelay<OrderEvents, AmqpCtx> {

    #channel: AmqpChannel

    constructor(channel: AmqpChannel, queues: QueueBinding[]) {

        super({ name: 'amqp' })
        this.#channel = channel

        for (const q of queues) {

            channel.consume(q.queue, (msg) => {

                if (!msg) return

                const { event, data } = JSON.parse(msg.content.toString())
                this.receive(event, data, {
                    ack: () => channel.ack(msg),
                    nack: () => channel.nack(msg),
                })
            }, q.config)
        }
    }

    protected send(event: string, data: unknown) {

        this.#channel.sendToQueue(
            event,
            Buffer.from(JSON.stringify(data))
        )
    }
}

const relay = new AmqpRelay(channel, [
    { queue: 'orders.placed', config: { noAck: false } },
    { queue: 'orders.shipped', config: { noAck: false } },
])

// Emit is just data. No transport concerns.
relay.emit('order:placed', { id: '123', total: 99.99 })

// Subscribe with transport context for ack/nack
relay.on('order:placed', ({ data, ctx }) => {

    processOrder(data)
    ctx.ack()
})

// Concurrency-controlled processing with rate limiting
relay.queue('order:placed', async ({ data, ctx }) => {

    await fulfillOrder(data)
    ctx.ack()
}, { concurrency: 5, rateLimitCapacity: 100, rateLimitIntervalMs: 60_000 })

It's just an abstract class — it doesn't ship with transport implementations. But you can wire it to Redis Pub/Sub, Kafka, SQS, WebSockets, Postgres LISTEN/NOTIFY, whatever. You implement send(), you call receive(), and all the observer abstractions just work across the wire.

Where this sits

This isn't a replacement for EventEmitter in simple cases. If you need to fire an event and listen for it in the same file, EventEmitter is fine. Node's built-in tooling is solid for straightforward use cases.

Where this earns its keep is in systems where events get complex — namespaced patterns across multiple domains, backpressure on high-throughput processing, cross-process communication that needs the same debugging tools as local code. The kind of codebase where you spend more time figuring out why something isn't firing than writing the logic itself.

That's the codebase I was working in when I started building this. And this is what I wished existed.

Documentation · GitHub · npm