Recovery & reliability

const sub = client.newSubscription(`orderbook_v3_event:${eventId}`, {
  recoverable: true,
});

from centrifuge import SubscriptionOptions

options = SubscriptionOptions(recoverable=True)
sub = client.new_subscription(f"orderbook_v3_event:{event_id}", handler, options)

sub.on("subscribed", (ctx) => {
  if (ctx.wasRecovering && ctx.recovered) {
    // Reconnected and gap was filled via history replay.
    // No need to re-fetch from REST — all missed messages were replayed.
  } else if (ctx.wasRecovering && !ctx.recovered) {
    // Reconnected but history was pruned before recovery could complete.
    // Too much time passed — re-seed your local state from REST.
  } else {
    // Fresh connect (first connection, or after a clean disconnect).
    // Seed initial state from REST, then rely on the subscription for updates.
  }
});

from centrifuge import SubscribedContext, SubscriptionEventHandler

async def on_subscribed(ctx: SubscribedContext) -> None:
    if ctx.was_recovering and ctx.recovered:
        # Reconnected and gap was filled via history replay.
        # No need to re-fetch from REST — all missed messages were replayed.
        pass
    elif ctx.was_recovering and not ctx.recovered:
        # Reconnected but history was pruned before recovery could complete.
        # Too much time passed — re-seed your local state from REST.
        pass
    else:
        # Fresh connect (first connection, or after a clean disconnect).
        # Seed initial state from REST, then rely on the subscription for updates.
        pass

handler = SubscriptionEventHandler(on_subscribed=on_subscribed)

const seen = new Set();
const MAX_SEEN = 10_000;

sub.on("publication", (ctx) => {
  const id = ctx.tags?.messageId;
  if (id !== undefined) {
    if (seen.has(id)) return;
    seen.add(id);
    if (seen.size > MAX_SEEN) {
      seen.delete(seen.values().next().value);
    }
  }
  applyUpdate(ctx.data);
});

from collections import OrderedDict
from centrifuge import PublicationContext, SubscriptionEventHandler

MAX_SEEN = 10_000
seen: "OrderedDict[str, None]" = OrderedDict()

async def on_publication(ctx: PublicationContext) -> None:
    msg_id = (ctx.tags or {}).get("messageId")
    if msg_id is not None:
        if msg_id in seen:
            return
        seen[msg_id] = None
        if len(seen) > MAX_SEEN:
            seen.popitem(last=False)
    apply_update(ctx.data)

handler = SubscriptionEventHandler(on_publication=on_publication)

const resp = await sub.history({ limit: 50 });
for (const pub of resp.publications) {
  applyUpdate(pub.data);
}

resp = await sub.history(limit=50)
for pub in resp.publications:
    apply_update(pub.data)