Messaging APIs

Complete guide to real-time messaging with publish/subscribe patterns, advanced subscription options, and message history retrieval.

Real-Time Messaging

`subscribe(topic, handler, options?): Promise<void>`

Subscribe to messages on a topic with full acknowledgement support.

Parameters:

topic: string - Topic name to subscribe to
handler: (message: MessageBody) => void - Message callback
onAck?: SubscriptionCallback - Optional subscription acknowledgement callback
timeoutMs?: number - Subscription timeout in milliseconds (default: 10000)
options?: SubscribeOptions - Optional configuration

Message Format:

interface MessageBody {
  id: string; // Unique message ID
  topic: string; // Topic name
  senderId: string; // Sender's client ID
  seq: string; // Sequence number (ULID)
  sentAt: Date; // Server timestamp
  payload: string; // Message content
  clientMsgId?: string; // Optional client correlation ID
}

SubscribeOptions:

interface SubscribeOptions {
  streamOldMessages?: boolean; // Deliver missed messages (default: false)
}

type SubscriptionCallback = (response: {
  success: boolean;
  error?: { code: string; message: string };
}) => void;

Examples:

// Simple subscription
await client.subscribe("chat:lobby", (msg) => {
  console.log(`${msg.senderId}: ${msg.payload}`);
});

// With options
await client.subscribe("chat:lobby", (msg) => console.log(msg), {
  streamOldMessages: true,
});

// With acknowledgement callback
await client.subscribe(
  "chat:lobby",
  (msg) => console.log("Message:", msg.payload),
  (response) => {
    if (response.success) {
      console.log("Subscribed successfully");
    } else {
      console.error("Subscription failed:", response.error?.message);
    }
  },
);

// Full control: acknowledgement, timeout, and options
await client.subscribe(
  "chat:lobby",
  (msg) => console.log("Message:", msg.payload),
  (response) => {
    if (response.success) {
      console.log("Subscription confirmed");
    }
  },
  5000, // 5 second timeout
  { streamOldMessages: true },
);

`unsubscribe(topic: string): void`

Unsubscribe from a topic.

Parameters:

topic: string - Topic to unsubscribe from

// Simple unsubscribe
client.unsubscribe("chat:lobby");

// With acknowledgement callback (optional)
client.unsubscribe(
  "chat:lobby",
  (response) => {
    if (response.success) {
      console.log("Unsubscribed successfully");
    }
  },
  5000, // timeout in ms
);

`publish(topic, payload): Promise<string>`

Publish a message without acknowledgement.

Parameters:

topic: string - Topic to publish to
payload: string - Message content

Returns: Promise<string> - Client message ID

const msgId = await client.publish("chat:lobby", "Hello!");

`publishWithAck(topic, payload, onAck, timeout?): Promise<string>`

Publish with server acknowledgement.

Parameters:

topic: string - Topic to publish to
payload: string - Message content
onAck: AckCallback - Acknowledgement callback
timeout?: number - Timeout in milliseconds (default: 3000)

AckCallback Type:

type AckCallback = (response: AckResponse) => void;

type AckResponse = {
  success: boolean;
  ack?: AckPacketType;
  seq?: string;
  serverMsgId?: string;
  topic: string;
  error?: {
    code: string;
    message: string;
  };
};

Example:

await client.publishWithAck(
  "chat:lobby",
  "Hello!",
  (ack) => {
    if (ack.success) {
      console.log("Message delivered! Seq:", ack.seq);
    } else {
      console.error("Failed:", ack.error?.message);
    }
  },
  5000,
);

Message History

Retrieve past messages from any topic with cursor-based pagination. Messages are stored for 3 days with up to 100 messages per topic.

Quick Start

// Fetch latest 50 messages
const history = await client.getHistory("chat:lobby");

history.items.forEach((msg) => {
  console.log(`[${msg.seq}] ${msg.senderId}: ${msg.payload}`);
});

// Paginate with cursor
if (history.nextCursor) {
  const nextPage = await client.getHistory("chat:lobby", {
    cursor: history.nextCursor,
    limit: 50,
  });
}

Pagination Iterator

For cleaner pagination code, use the history iterator:

const getNext = client.createHistoryIterator("chat:lobby", { limit: 50 });

const page1 = await getNext(); // { items: [...], hasMore: true }
const page2 = await getNext(); // { items: [...], hasMore: false }
const done = await getNext(); // null (exhausted)

ℹ️

For comprehensive history features, see the full guide:

Pagination patterns (infinite scroll, load more)
Catch-up after reconnection
Search in history
Time-based queries with ULID
Performance optimization tips

→ Read the Message History Guide

Advanced Messaging Patterns

Message Acknowledgment Patterns

Fire-and-Forget

// Simple publish without confirmation
await client.publish("notifications", "System update complete");

Reliable Delivery

// Publish with acknowledgment
await client.publishWithAck(
  "critical-alerts",
  "Server maintenance starting",
  (ack) => {
    if (ack.success) {
      console.log("Alert delivered successfully");
    } else {
      // Retry logic or fallback notification
      console.error("Alert delivery failed:", ack.error?.message);
    }
  },
  5000,
);

Batch Publishing

async function publishBatch(topic: string, messages: string[]) {
  const promises = messages.map((msg) =>
    client.publishWithAck(topic, msg, (ack) => {
      console.log(`Message ${msg}: ${ack.success ? "✅" : "❌"}`);
    }),
  );

  await Promise.all(promises);
  console.log("All messages sent");
}

Subscription Patterns

Topic Subscription with Cleanup

const subscriptions = new Set<string>();

async function subscribeWithCleanup(
  topic: string,
  handler: (msg: MessageBody) => void,
) {
  await client.subscribe(topic, handler);
  subscriptions.add(topic);

  // Return unsubscribe function
  return () => {
    client.unsubscribe(topic);
    subscriptions.delete(topic);
  };
}

// Cleanup all subscriptions
function unsubscribeAll() {
  for (const topic of subscriptions) {
    client.unsubscribe(topic);
  }
  subscriptions.clear();
}

Pattern-Based Subscriptions

// Subscribe to multiple related topics
const roomTopics = ["room-123", "room-456", "room-789"];

for (const topic of roomTopics) {
  await client.subscribe(topic, (msg) => {
    handleRoomMessage(msg.topic, msg.payload);
  });
}

Subscription with Buffering

class MessageBuffer {
  private buffer: MessageBody[] = [];
  private batchSize = 10;
  private flushInterval = 1000; // 1 second

  constructor(
    topic: string,
    private processBatch: (messages: MessageBody[]) => void,
  ) {
    client.subscribe(topic, (msg) => this.addMessage(msg));
    setInterval(() => this.flush(), this.flushInterval);
  }

  private addMessage(msg: MessageBody) {
    this.buffer.push(msg);
    if (this.buffer.length >= this.batchSize) {
      this.flush();
    }
  }

  private flush() {
    if (this.buffer.length > 0) {
      this.processBatch([...this.buffer]);
      this.buffer.length = 0;
    }
  }
}

Error Handling

Publishing Errors

try {
  await client.publish("topic", "message");
} catch (error) {
  if (error.message.includes("Not connected")) {
    console.error("Connection lost - will reconnect automatically");
  } else if (error.message.includes("Invalid topic")) {
    console.error("Topic name must be non-empty string");
  }
}

Subscription Errors

// With callback for error handling
await client.subscribe(
  "topic",
  (msg) => console.log(msg),
  (response) => {
    if (!response.success) {
      console.error("Subscription failed:", response.error?.message);
      // Implement retry logic
    }
  },
);

// Or use try-catch
try {
  await client.subscribe("topic", (msg) => console.log(msg));
} catch (error) {
  console.error("Subscription failed:", error.message);
}

Performance Optimization

Message Batching

Batch multiple publishes to reduce overhead:

class MessageBatcher {
  private queue: Array<{ topic: string; payload: string }> = [];
  private batchTimeout?: NodeJS.Timeout;

  publish(topic: string, payload: string) {
    this.queue.push({ topic, payload });

    if (this.queue.length >= 10) {
      this.flush();
    } else if (!this.batchTimeout) {
      this.batchTimeout = setTimeout(() => this.flush(), 100);
    }
  }

  private async flush() {
    if (this.batchTimeout) {
      clearTimeout(this.batchTimeout);
      this.batchTimeout = undefined;
    }

    const batch = [...this.queue];
    this.queue.length = 0;

    // Send all messages in parallel
    await Promise.all(
      batch.map(({ topic, payload }) => client.publish(topic, payload)),
    );
  }
}

Next Steps

Message History Guide - Comprehensive guide to history API, pagination patterns, and advanced use cases
Presence Tracking - Monitor user activity and awareness
Connection Management - Connection lifecycle and error handling
Types Reference - Complete TypeScript definitions

Messaging APIs

On this page