A realtime connection manager for
every browser tab, one socket.

# Core manager
pnpm add @connexis/core

# Transport adapters
pnpm add @connexis/transport-websocket @connexis/transport-sse @connexis/transport-polling

# React bindings (optional)
pnpm add @connexis/react

import { createRealtimeClient } from '@connexis/core';
import { WebSocketTransport } from '@connexis/transport-websocket';

const client = createRealtimeClient({
  transport: new WebSocketTransport('wss://api.example.com/events'),
  connectionPolicy: 'isolated',
});

import { SSETransport } from '@connexis/transport-sse';

// Standard GET stream
const client = createRealtimeClient({
  transport: new SSETransport('https://api.example.com/stream', {
    publishUrl: 'https://api.example.com/publish', // fallback POST for publish()
  }),
});

// QUERY stream with headers and a body payload
const scopedClient = createRealtimeClient({
  transport: new SSETransport('https://api.example.com/stream', {
    method: 'QUERY',
    headers: { Authorization: 'Bearer my-token' },
    body: { topics: ['news', 'alerts'], userId: '123' },
  }),
});

import { PollingTransport } from '@connexis/transport-polling';

const client = createRealtimeClient({
  transport: new PollingTransport('https://api.example.com/poll', {
    pollInterval: 5000, // ms
  }),
});

const transport = new WebSocketTransport('wss://api.example.com/events', {
  // Called on the initial connect and every retry
  authToken: async () => {
    const token = await fetchFreshJWT();
    return token;
  },
});

const client = createRealtimeClient({
  transport,
  reconnectOptions: {
    maxAttempts: 5,       // give up after 5 attempts
    delay: 1000,          // wait 1s between attempts
    stableThreshold: 5000 // must stay open 5s to count as stable
  },
});

const client = createRealtimeClient({
  transport: new WebSocketTransport('wss://api.example.com/events'),
  connectionPolicy: (sub) => {
    if (sub.metadata?.tier === 'premium') return 'premium-conn';
    return 'default-shared-conn';
  },
});

import { Coordinator } from '@connexis/coordinator';

const client = createRealtimeClient({
  transport: new WebSocketTransport('wss://api.example.com/events'),
  connectionPolicy: 'shared',
  coordinator: new Coordinator('my_app_namespace'),
});

client.use(async (context, next) => {
  if (context.direction === 'outbound' && context.topic === 'orders') {
    context.payload.timestamp = Date.now();
  }

  if (context.direction === 'inbound' && context.payload.isSpam) {
    throw new Error('Rejected spam message'); // stops the pipeline
  }

  await next();
});

import { RealtimeProvider } from '@connexis/react';
import { client } from './client';

export default function App() {
  return (
    <RealtimeProvider client={client}>
      <Dashboard />
    </RealtimeProvider>
  );
}

import { useConnection, useSubscription, usePublish } from '@connexis/react';

function Dashboard() {
  const { state, metrics } = useConnection();
  const publish = usePublish();

  useSubscription('orders', { filter: { region: 'US' } }, (order) => {
    console.log('New US order:', order);
  });

  return (
    <div>
      <h3>Status: {state}</h3>
      <p>Latency: {metrics.latency.toFixed(1)} ms</p>
      <button onClick={() => publish('alerts', { text: 'Emergency alert' })}>
        Broadcast alert
      </button>
    </div>
  );
}

import { useState } from 'react';
import { useConnection, useSubscription, usePublish } from '@connexis/react';

type Column = 'todo' | 'progress' | 'done';
interface Card { id: string; title: string; column: Column; }

function KanbanBoard() {
  const { state } = useConnection();
  const publish = usePublish();
  const [cards, setCards] = useState<Card[]>([
    { id: '1', title: 'Design DB schema', column: 'todo' },
    { id: '2', title: 'Implement auth pipeline', column: 'progress' },
  ]);

  useSubscription('kanban_moves', (payload: { id: string; column: Column }) => {
    setCards((prev) =>
      prev.map((c) => (c.id === payload.id ? { ...c, column: payload.column } : c))
    );
  });

  const moveCard = async (id: string, column: Column) => {
    setCards((prev) => prev.map((c) => (c.id === id ? { ...c, column } : c)));
    await publish('kanban_moves', { id, column });
  };

  return (
    <div>
      <p>Connection: {state}</p>
      {/* columns + moveCard() calls on drop */}
    </div>
  );
}

class RealtimeClient<TEvents = any> extends EventEmitter {
  constructor(config: RealtimeClientConfig);

  /** Registers Koa-style middleware */
  use(middleware: Middleware): this;

  /** Current connection state */
  readonly state: ConnectionState;

  /** Current aggregated metrics */
  readonly metrics: Metrics;

  /** Subscribes to a topic; resolves to an unsubscribe function */
  subscribe<K>(topic: K, handler: (data: any) => void): Promise<() => Promise<void>>;
  subscribe<K>(topic: K, options: SubscribeOptions, handler: (data: any) => void): Promise<() => Promise<void>>;

  /** Publishes a payload to a topic */
  publish<K>(topic: K, payload: any): Promise<void>;

  /** Tears down the client and closes all connections */
  destroy(): Promise<void>;
}

client.on('stateChange', ({ state }) => {
  console.log(`Connection state is now: ${state}`);
});

client.on('metricsChange', (metrics) => {
  console.log(`Round-trip latency: ${metrics.latency}ms`);
});