Centrifugo Integration for Real-Time Messages on Website

Centrifugo Integration for Real-Time Messages on Website

Centrifugo is a self-hosted real-time server, compatible with any backend via HTTP API. Written in Go, minimal resource consumption, supports WebSocket, SSE, HTTP streaming and SockJS.

Integration Architecture

Your backend (any language)
    │── HTTP API → Centrifugo (publishing)
    │── gRPC API → Centrifugo
    │
Client (browser) ◄── WebSocket ──► Centrifugo

Installation via Docker

# docker-compose.yml
services:
  centrifugo:
    image: centrifugo/centrifugo:v5
    command: centrifugo -c /centrifugo/config.json
    ports:
      - "8000:8000"
    volumes:
      - ./centrifugo-config.json:/centrifugo/config.json

// centrifugo-config.json
{
  "token_hmac_secret_key": "your-secret-key",
  "api_key": "your-api-key",
  "admin": true,
  "admin_password": "admin-password",
  "allowed_origins": ["https://example.com"],
  "namespaces": [
    {
      "name": "notifications",
      "subscribe_for_client": false,
      "history_size": 100,
      "history_ttl": "24h"
    },
    {
      "name": "chat",
      "presence": true,
      "join_leave": true,
      "history_size": 500,
      "history_ttl": "1h"
    }
  ]
}

Server: JWT Generation and Publishing

import jwt from 'jsonwebtoken';
import axios from 'axios';

// Token for client
function generateCentrifugoToken(userId: string): string {
  return jwt.sign(
    {
      sub: userId,
      exp: Math.floor(Date.now() / 1000) + 3600
    },
    process.env.CENTRIFUGO_SECRET,
    { algorithm: 'HS256' }
  );
}

// Publish message via HTTP API
async function publishMessage(channel: string, data: unknown): Promise<void> {
  await axios.post(`${process.env.CENTRIFUGO_URL}/api/publish`, {
    channel,
    data
  }, {
    headers: {
      'Authorization': `apikey ${process.env.CENTRIFUGO_API_KEY}`,
      'Content-Type': 'application/json'
    }
  });
}

// Send notification to user
await publishMessage(`notifications#${userId}`, {
  type: 'order_shipped',
  orderId: '12345',
  message: 'Your order shipped'
});

// Batch publish
await axios.post(`${process.env.CENTRIFUGO_URL}/api/batch`, {
  commands: userIds.map(id => ({
    publish: {
      channel: `notifications#${id}`,
      data: notification
    }
  }))
});

Client (JavaScript)

import { Centrifuge } from 'centrifuge';

const centrifuge = new Centrifuge(process.env.NEXT_PUBLIC_CENTRIFUGO_URL, {
  token: await fetchCentrifugoToken(),  // from your backend
  debug: process.env.NODE_ENV === 'development'
});

centrifuge.on('connect', (ctx) => {
  console.log('Connected to Centrifugo', ctx.transport);
});

centrifuge.on('disconnect', (ctx) => {
  console.log('Disconnected', ctx.reason);
});

// Subscribe to personal notifications
const sub = centrifuge.newSubscription(`notifications#${userId}`, {
  getToken: async (ctx) => {
    // Get subscription token for private channel
    const resp = await fetch('/api/centrifugo/subscription-token', {
      method: 'POST',
      body: JSON.stringify({ channel: ctx.channel })
    });
    return (await resp.json()).token;
  }
});

sub.on('publication', (ctx) => {
  const notification = ctx.data;
  showNotification(notification.message);
});

sub.on('subscribing', (ctx) => console.log('Subscribing...'));
sub.on('subscribed', (ctx) => console.log('Subscribed'));

sub.subscribe();
centrifuge.connect();

Presence — Online Statuses

// Who's currently reading document (collaborative editing)
const docSub = centrifuge.newSubscription(`chat:doc-${documentId}`);

docSub.presence().then((presence) => {
  const onlineUsers = Object.values(presence.clients);
  setOnlineUsers(onlineUsers);
});

docSub.on('join', ({ info }) => {
  console.log(`${info.user} joined`);
});

docSub.on('leave', ({ info }) => {
  console.log(`${info.user} left`);
});

Timeline

Centrifugo + Node.js SDK + client with private channels: 3–5 days.