Persistent Pub/Sub

Overview

This tutorial builds a publisher and subscriber on a KubeMQ Events Store channel — reach for this pattern when a subscriber can't guarantee it's listening the instant a message is published. Plain events are fire-and-forget: publish with no one subscribed and the message is gone. Events Store persists every event to a durable, ordered log, so a subscriber connecting seconds or a full restart later still catches up — useful for anything needing a complete history, like an audit trail or event-sourced state.

The two calls involved: sendEventStore() publishes and returns a result confirming storage plus a broker-assigned sequence number, and subscribeToEventsStore() takes a required startFrom position telling the broker where to start — new events only, from the first stored event (EventStoreStartPosition.StartFromFirst, used here), or a given sequence or time. Production subscribers usually resume from a saved checkpoint instead of replaying from the beginning.

Gotchas: replaying from the first event on every restart replays the whole log, which gets costly on a busy channel — track the last sequence you processed instead. Starting from new-only has the opposite risk: anything published earlier is silently skipped, so don't rely on a fixed delay like this sample's to paper over that race in production. Persistence isn't consumer coordination: each independent subscriber gets its own full replay unless grouped with a consumer group.

Prerequisites

• KubeMQ server running on localhost:50000
• Node.js SDK installed (npm install kubemq-js)

Code

/**
 * Example: Persistent Event Publish/Subscribe
 *
 * Demonstrates event store messaging where events are persisted and can be
 * replayed by subscribers. Unlike regular events, subscribers can connect
 * after events are published and still receive them.
 *
 * Prerequisites:
 *   - KubeMQ server running on localhost:50000
 *
 * Run: npx tsx examples/events-store/persistent-pubsub.ts
 */
import { KubeMQClient, createEventStoreMessage, EventStoreStartPosition } from 'kubemq-js';

async function main(): Promise<void> {
  const client = await KubeMQClient.create({
    address: 'localhost:50000',
    clientId: 'js-events-store-persistent-pubsub-client',
  });

  try {
    // Publish events first — they are persisted in the store.
    for (let i = 1; i <= 3; i++) {
      await client.sendEventStore(
        createEventStoreMessage({
          channel: 'js-events-store.persistent-pubsub',
          body: `User action #${i}: login from 192.168.1.${i}`,
          tags: { action: 'login', sequence: String(i) },
        }),
      );
      console.log(`Published event #${i}`);
    }

    // Subscribe starting from the first stored event — replays all three.
    const subscription = client.subscribeToEventsStore({
      channel: 'js-events-store.persistent-pubsub',
      startFrom: EventStoreStartPosition.StartFromFirst,
      onEvent: (event) => {
        console.log(`Received [seq=${event.sequence}]:`, new TextDecoder().decode(event.body));
      },
      onError: (err) => {
        console.error('Subscription error:', err.message);
      },
    });

    await new Promise((resolve) => setTimeout(resolve, 2000));
    subscription.cancel();
  } finally {
    await client.close();
  }
}

main().catch(console.error);

// Expected output:
// Published event #1
// Published event #2
// Published event #3
// Received [seq=<sequence>]: User action #1: login from 192.168.1.1
// Received [seq=<sequence>]: User action #2: login from 192.168.1.2
// Received [seq=<sequence>]: User action #3: login from 192.168.1.3

How It Works

• sendEventStore() persists each event on the server — unlike sendEvent(), messages survive subscriber disconnection.
• subscribeToEventsStore() with StartFromFirst replays all stored events from the very beginning of the channel, even though the publisher finished before the subscriber connected.
• The event.sequence field is a server-assigned monotonic integer; it can be used to resume replay from a checkpoint (see Replay from Sequence).
• The 2-second wait gives the subscriber time to receive all replayed events before cancel() closes the stream.

Related

• Pattern overview
• Node.js SDK Reference
• Cancel Subscription
• Consumer Group