centrifuge-js

Map subscriptions deliver a real-time key-value collection managed by Centrifugo. Clients receive a full snapshot on subscription and observe live updates to individual keys.

Key Features:

• Events: Emits a sync event when a fresh snapshot is delivered (initial subscribe or unrecoverable recovery). Emits update events for individual key changes (set/remove).
• Variants: newMapClientsSubscription and newMapUsersSubscription provide presence-style maps backed by $clients:* and $users:* channels.
• Mutation: Use publish(key, data) and remove(key) to change values.

WARNING

Experimental: Requires Centrifugo >= v6.8.0. Behavior may change in future minor releases.

const mapSub = centrifuge.newMapSubscription('my_map_channel');

mapSub.on('sync', (snapshot) => {
  // Handle full snapshot
});

mapSub.on('update', (update) => {
  // Handle individual key change
});

mapSub.publish('key1', { value: 'data' });
mapSub.remove('key1');

Shared poll subscriptions move the polling logic from the client to Centrifugo. Centrifugo polls the backend at a configurable interval and fans out changes to all interested clients, reducing backend load.

Key Features:

• Management: Use track(keys) and untrack(keys) on the returned SharedPollSubscription to manage which items the client is interested in.
• Authorization: Provide a getSignature callback in the options to authorize tracked keys via HMAC signatures.
• Events: Emits an update event per item when the backend reports a new version. Includes removed: true events when a key is revoked.

WARNING

Experimental: Requires Centrifugo >= v6.8.0. Behavior may change in future minor releases.

const pollSub = centrifuge.newSharedPollSubscription('poll_channel', {
  getSignature: (key) => 'your_hmac_signature'
});

pollSub.on('update', (update) => {
  // Handle update or removal
});

pollSub.track(['key1', 'key2']);
pollSub.untrack(['key1']);

While WebSocket is the primary transport, you can provide an array of TransportEndpoint objects to enable fallbacks (like HTTP-streaming or SSE) in environments where WebSockets are blocked by firewalls or proxies.

The client will attempt the transports in the order provided during the initial handshake. Once a transport succeeds, it will be used for subsequent reconnects.

Supported transport types:

• websocket
• http_stream (requires server-side bidirectional emulation)
• sse (requires server-side bidirectional emulation)
• sockjs (Deprecated: removed in Centrifugo v6 and will be removed in centrifuge-js v6)
• webtransport (Experimental; currently only supported by Centrifugo)

const transports = [
    {
        transport: 'websocket',
        endpoint: 'ws://example.com/connection/websocket'
    },
    {
        transport: 'http_stream',
        endpoint: 'http://example.com/connection/http_stream'
    },
    {
        transport: 'sse',
        endpoint: 'http://example.com/connection/sse'
    }
];
const centrifuge = new Centrifuge(transports);
centrifuge.connect();

To handle dynamic authentication or token refreshing, provide a getToken function in the Centrifuge options. The SDK will call this function when it needs a token (e.g., during initial connection or when the current token expires).

Key Behaviors:

• If token is not provided but getToken is, the SDK assumes token authentication is required and will attempt to fetch a token before the initial connection.
• If your getToken function throws a Centrifuge.UnauthorizedError(), the client will move to a disconnected state (useful for logout/permission loss).
• Any other error thrown will cause the SDK to retry the token refresh after a jittered delay.

Note: The connection token must be generated on your application backend.

import { Centrifuge, UnauthorizedError } from 'centrifuge';

async function getToken() {
    if (!loggedIn) {
        // Throwing UnauthorizedError disconnects the client
        throw new UnauthorizedError();
    }
    const res = await fetch('/centrifuge/connection_token');
    if (!res.ok) {
        throw new Error(`Unexpected status code ${res.status}`);
    }
    const data = await res.json();
    return data.token;
}

const client = new Centrifuge(
    'ws://localhost:8000/connection/websocket',
    {
        getToken: getToken
    }
);

Server-side subscriptions are created by the server upon connection establishment. While the client has less control over them than client-side subscriptions, the SDK maintains them in an internal registry and provides event hooks.

Available Events for Server-side Subscriptions:

• subscribed: Called when the client moves to a connected state or receives a Subscribe push.
• subscribing: Called during reconnection or explicit disconnection.
• unsubscribed: Called when the server sends an unsubscribe push or the subscription disappears upon reconnection.
• publication: Called when the server sends a Publication over the channel.

Top-level methods for server-side channels:

• publish(channel, data)
• history(channel, options)
• presence(channel)
• presenceStats(channel)

const client = new Centrifuge('ws://localhost:8000/connection/websocket', {});

client.on('subscribed', (ctx) => {
    console.log('subscribed to server-side channel', ctx.channel);
});

client.on('publication', (ctx) => {
    console.log('publication receive from server-side channel', ctx.channel, ctx.data);
});

client.connect();

You can provide a subscription token (typically a JWT generated on your backend) when creating a subscription. If the token includes an expiration, the SDK can automatically refresh it using the getToken option.

How getToken works:

1. The SDK calls your provided getToken function when a new token is needed.
2. Your function must return the new token string.
3. If your function returns an empty string, the SDK assumes the user no longer has permission and will unsubscribe the client.
4. If your function throws an error, the SDK will retry the refresh after a jittered delay.

Note: If you provide getToken but no initial token, the SDK assumes you are using token authorization and will attempt to fetch a token before the initial subscription attempt.

import { Centrifuge, UnauthorizedError } from 'centrifuge';

async function getToken(ctx) {
    // ctx contains the channel name
    const res = await fetch('/centrifuge/subscription_token', {
        method: 'POST',
        headers: new Headers({ 'Content-Type': 'application/json' }),
        body: JSON.stringify(ctx)
    });

    if (!res.ok) {
        if (res.status === 403) {
            // Throwing UnauthorizedError stops further refresh attempts and unsubscribes
            throw new UnauthorizedError();
        }
        throw new Error(`Unexpected status code ${res.status}`);
    }

    const data = await res.json();
    return data.token;
}

const client = new Centrifuge('ws://localhost:8000/connection/websocket', {});

const sub = client.newSubscription('news', {
    token: 'JWT-GENERATED-ON-BACKEND-SIDE',
    getToken: getToken,
});

sub.subscribe();