Receive messages effectively

Showing JavaScript examples.

This guide shows you how to structure the subscribe side of a PubNub application. You'll learn how to:

• Subscribe from a direct client connection instead of through a proxy.
• Keep a checkpoint of the last message you processed.
• Replay messages published while your client was disconnected, and skip the ones you already processed.
• Copy received data into your own systems without forking every message from the client.

Every call on this page needs an SDK instance initialized with your subscribe key. If you don't have a keyset yet, start with Set up your account. This guide assumes you already know how to register a subscription. If you don't, start with Receive messages.

Examples use the JavaScript, Swift, Java, Kotlin, and Python SDKs, which document every parameter this guide uses. For any other language, refer to Available SDKs.

Subscribe from the client, not through a proxy

Open the subscription from the client that needs the messages, and let the SDK hold its own connection to PubNub's nearest point of presence. A proxy in the middle can conflict with the encryption and the long-lived TCP connection the SDK already maintains, and it inherits the proxy's own downtime as its own. For the reasoning behind a direct connection, refer to Architectural choices.

Recover messages published while you were disconnected

Live delivery to subscribers is at-most-once by default. On a stable connection, a subscriber receives each message at most once. After a reconnect, a replayed message can arrive again with the same timetoken. A subscriber can also miss messages if its buffer overflows or if it's disconnected when someone publishes a message.

To recover the gap, keep a checkpoint of the last message you processed and replay history from it after a reconnect. The recipe has four parts:

• A durable checkpoint. Store the timetoken of the last message your handler finished, and write it only after the handler succeeds. Load it from storage when your app starts, so a restart recovers the same way a reconnect does.
• A hold on live delivery. When the status listener reports a disconnect, buffer live messages instead of processing them.
• Bounded, paged replay. When the client connects, page backward through history from now to the checkpoint, up to a page limit. If the gap needs more pages than the limit, reload state from your server instead of replaying.
• One merge path. Process the replayed messages first, then the buffered live messages in timetoken order. Every message goes through the same check, which skips any message at or before the checkpoint, so a message that arrives on both paths runs once.

In each example, the checkpoint storage, the message handler, and the reload function are placeholders for your own code. The handler throws on failure, which leaves the checkpoint where it was and starts a replay from it.

• JavaScript
• Swift
• Java
• Kotlin
• Python

1const CHANNEL = 'channel_1';
2const PAGE_SIZE = 100;
3const MAX_PAGES = 10;
4const RETRY_DELAY_MS = 5000;
5

6// Timetoken of the last message your handler finished, from durable storage.
7let checkpoint = loadCheckpoint();
8let holdLive = checkpoint !== null;
9let recoveryRunning = false;
10const liveBuffer = [];
11

12function processMessage(timetoken, message) {
13  if (checkpoint !== null && BigInt(timetoken) <= BigInt(checkpoint)) return;
14  handleMessage(message); // throws on failure, so the checkpoint doesn't move
15  checkpoint = timetoken;
16  saveCheckpoint(checkpoint);
17}
18

19subscription.onMessage = (event) => {
20  const timetoken = String(event.timetoken);
21  if (holdLive) {
22    liveBuffer.push({ timetoken, message: event.message });
23    return;
24  }
25  try {
26    processMessage(timetoken, event.message);
27  } catch (error) {
28    console.error('Handler failed, replaying from the checkpoint:', error);
29    recoverMissedMessages();
30  }
31};
32

33pubnub.addListener({
34  status: (event) => {
35    switch (event.category) {
36      case 'PNDisconnectedCategory':
37      case 'PNDisconnectedUnexpectedlyCategory':
38      case 'PNConnectionErrorCategory':
39      case 'PNNetworkDownCategory':
40        holdLive = true;
41        break;
42      case 'PNConnectedCategory':
43      case 'PNReconnectedCategory':
44        recoverMissedMessages();
45        break;
46    }
47  },
48});
49

50async function recoverMissedMessages() {
51  if (recoveryRunning) return;
52  recoveryRunning = true;
53  holdLive = true;
54  try {
55    if (checkpoint !== null) {
56      for (const item of await fetchSinceCheckpoint()) {
57        processMessage(String(item.timetoken), item.message);
58      }
59    }
60    liveBuffer.sort((a, b) => (BigInt(a.timetoken) < BigInt(b.timetoken) ? -1 : 1));
61    while (liveBuffer.length > 0) {
62      processMessage(liveBuffer[0].timetoken, liveBuffer[0].message);
63      liveBuffer.shift();
64    }
65    holdLive = false;
66    recoveryRunning = false;
67  } catch (error) {
68    recoveryRunning = false;
69    if (error.gapTooLarge) {
70      reloadStateFromYourServer();
71      return;
72    }
73    console.error('Recovery failed, retrying:', error);
74    setTimeout(recoverMissedMessages, RETRY_DELAY_MS);
75  }
76}
77

78// Pages backward from now to the checkpoint and returns messages oldest first.
79async function fetchSinceCheckpoint() {
80  const pages = [];
81  let start;
82  for (let page = 0; page < MAX_PAGES; page++) {
83    const response = await pubnub.fetchMessages({
84      channels: [CHANNEL],
85      end: checkpoint,
86      start,
87      count: PAGE_SIZE,
88    });
89    const items = response.channels[CHANNEL] ?? [];
90    pages.unshift(items);
91    if (items.length < PAGE_SIZE) return pages.flat();
92    start = String(items[0].timetoken);
93  }
94  throw Object.assign(new Error('Gap exceeds MAX_PAGES'), { gapTooLarge: true });
95}

show all 95 lines

The examples recover one channel. For several channels, keep one checkpoint and one live buffer per channel. The Swift example relies on the SDK's default of running callbacks on the main queue. If you set a different callback queue, serialize access to the recovery state, as the Java, Kotlin, and Python examples do with a lock.

The SDK retries a dropped connection on its own, and a short outage may not produce a disconnect status at all. When the client reconnects, the subscriber buffer delivers what it holds:

The subscriber message buffer queues messages for a reconnecting client. It holds 100 messages for up to 16 minutes by default, and discards the oldest first (FIFO) when a burst exceeds that size. Larger buffers, for example 300 or 500 messages, can be provisioned per keyset by PubNub Support.

The examples start recovery on a status event or a handler failure. If you don't want to rely on a status event for every gap, also call the recovery function on a schedule or when your app returns to the foreground. It's safe to call at any time, because the checkpoint check skips messages you already processed.

The checkpoint moves only after the handler returns. If your handler fails partway through, for example after it wrote to a database, it runs again on the same message, so make its side effects safe to repeat. For publish-side options, refer to Exactly-once processing.

For how to register the status listener in your SDK, refer to Monitor and respond to connection status changes. For the fetch call's parameters, page limits, and how far back it can reach, refer to Retrieve message history.

Copy received data to your own systems without forking every message

Don't have the client forward a second copy of every message to your server on top of handling it. That doubles the mobile data and battery cost of every message, on top of whatever the client already does with it.

Use an After Publish Function to forward a copy from PubNub's own network instead. Or read the data back later through Message Persistence, rather than capturing a copy of each message as it arrives. Refer to Copy published data to your own systems without publishing twice for both options. The choice is the same regardless of which side of the connection triggers it.

Related tasks

• Receive messages. Create a subscription, register handlers, and subscribe to several channels at once.
• Monitor and respond to connection status changes. Attach the status listener this guide's recovery step depends on.
• Retrieve message history. Page through a long history and count unread messages.
• Send messages effectively. The publish-side half of this pattern, including the After Publish Function that avoids client-side forking.
• Connection management. What the SDK retries on its own and what each status means.
• Architectural choices. The reasoning behind a direct client connection and where a data copy belongs.