Realtime channel pool

For occasional or discrete server-side publishes, the REST API is a good choice. For high-throughput or latency-sensitive publishing, a realtime connection avoids the overhead of individual HTTP requests. Publishing successive messages through separate REST requests also does not guarantee their order.

A realtime channel pool makes realtime publishing practical for concurrent server workloads. It maintains a fixed number of connected clients, distributing channel names across them to increase publishing capacity without creating a connection for every task or request. Publishes to the same channel use the same connection while it remains connected, preserving message order across callers within the process.

The pool is part of the Ably Pub/Sub SDK for servers, @ably/pubsub-server, and is available only in JavaScript and TypeScript, on Node.js 22 or later.

The following server code streams a reply through a pool:

// Server-side code.
import { createRealtimeChannelPool } from '@ably/pubsub-server';

const pool = createRealtimeChannelPool({ key: process.env.ABLY_API_KEY }, { size: 4 });
await pool.open();

async function streamReply(channelName, chunks) {
  const channel = pool.get(channelName);
  try {
    await Promise.all(chunks.map((chunk) => channel.publish('chunk', chunk)));
  } finally {
    channel.release();
  }
}

How it works

The pool is part of the Ably Pub/Sub SDK for servers, @ably/pubsub-server. It works as follows:

1. createRealtimeChannelPool() takes the same client options as a realtime client, and the number of clients to hold.
2. open() creates and connects every client, once, when the process starts.
3. get() returns a handle on a channel. The handle does not choose a client until you call a method on it.
4. The first call on a channel name binds the name to the connected client with the fewest names bound to it. Later calls on that name, from any handle, go to the same client while it stays connected.
5. A handle publishes without attaching to the channel, using transient publishing. It holds no subscriptions and no channel state, so concurrent operations can share it without affecting each other.
6. release() drops a handle's hold on the channel. Once every handle on a name is released and its publishes have settled, the pool forgets the name.

Clients remain connected between operations, so an operation does not wait for a connection, and the clients keep state such as the current fallback host.

Publishes to the same channel are made through the same connection, in call order. This holds while that connection remains connected. If it disconnects, new calls may move to another client.

Create and open the pool

Install @ably/pubsub-server, which requires Node.js 22 or later:

npm install @ably/pubsub-server

Create one pool per process and open it before using it:

// Server-side code.
import { createRealtimeChannelPool } from '@ably/pubsub-server';

const pool = createRealtimeChannelPool(
  { key: process.env.ABLY_API_KEY },
  { size: 4, failedRetryTimeout: 30000 },
);

await pool.open();

The first argument is a client options object, an API key or a token, used by every client in the pool. The second argument sets options for the pool itself:

open() resolves once each client's first connection attempt has an outcome, whether the client connected or cannot connect yet. A client that cannot connect yet, such as during a network outage, keeps retrying by itself, so a transient outage does not stop your process from starting.

If a client enters the failed state during its first attempt, such as with an invalid API key, open() rejects with that client's error and closes the pool. Repeated calls to open() return the same promise.

Choose the pool size

Each realtime connection accepts a limited inbound message rate. Size the pool for the rate at which your process publishes, estimating the number of clients as:

clients ≥ (concurrent operations × messages per second per operation) ÷ per-connection limit

For example, 20 streams publishing 25 messages per second each generate 500 messages per second. With a limit of 100 per connection, that requires at least 5 clients.

The pool balances channel names, not message rates, so busy channels can be unevenly distributed. Allow headroom, such as 8 clients for this example.

Each pool client counts towards your account's concurrent connection limit. A pool of 8 in each of 10 processes uses 80 connections.

Publish on a channel

Call get() with a channel name to get a handle, publish through it, then release it when you are finished:

// Server-side code.
const channel = pool.get('chat:123');
try {
  await channel.publish('agent.status', { status: 'thinking' });
  await channel.publish('agent.reply', { text: 'Hello' });
} finally {
  channel.release();
}

get() does not fail, because the handle chooses its client when you call a method on it. It is bound to the pool, so const { get } = pool works.

In TypeScript 5.2 or later, or in JavaScript on Node.js 24 or later, declare the handle with using instead, which releases it when the block exits:

// Server-side code, Node.js 24 or later.
{
  using channel = pool.get('chat:123');
  await channel.publish('agent.reply', { text: 'Hello' });
}

release() only affects the handle you call it on, and a second call does nothing. A method called on a released handle rejects with error 40000.

Use the supported operations

A handle publishes, but cannot attach to the channel or subscribe to it. It has the following members:

To read history, query presence, use LiveObjects, or subscribe, use a separate client created with createHttpClient() or createRealtimeClient() from @ably/pubsub-server.

Handle connection states

Each call on a handle runs on the client that its channel name is bound to while that client is connected. If the bound client is not connected, the name moves to the connected client with the fewest names bound to it. If no client is connected, the call goes to the bound client, or to the client with the fewest names if the name is not yet bound, and its outcome depends on that client's connection state:

If queueMessages is false in the client options, a call is not queued. While the connection is connecting or disconnected, it rejects with the connection's errorReason, or with error 90000 if there is none.

Each call resolves its client afresh, so if you retry a call that rejected, the retry goes to a connected client if there is one.

The pool does not expose its clients' connection state. Instead, handle connection errors where calls fail.

To monitor the pool's connections, use the clients' logs. A client whose connection fails logs the failure and its reason at error level, which the default logLevel of 1 outputs. Set logLevel to 2 in the client options to also log disconnections and suspensions, and set logHandler to send the logs to your own logging.

Close the pool

Call close() once the work that uses the pool has finished, for example when your process shuts down:

// Server-side code.
await pool.close();

close() waits for the calls in progress to settle, then closes every client, and resolves once every connection is closed. A second call returns the promise of the first. A closed pool cannot be reopened, so call createRealtimeChannelPool() again if the process needs a new one.

Handle errors

The pool and its handles reject with an ErrorInfo in the following cases:

Other errors are those of the underlying client and channel, such as a connection's errorReason or a rejected publish.

Edge cases

Consider the following when you publish through a pool:

• Order holds only while a channel's client stays connected. If it disconnects, calls already made on it are sent once it reconnects, while new calls on that name move to another connected client, so messages published either side of the move can arrive out of order.
• A call made before open() is called rejects with error 40000. A call made while open() is in progress goes ahead, and meets the connection state of its client. Await open() before you start the work that uses the pool, so that a misconfiguration such as an invalid key stops the process from starting.
• Each process holds its own pool, so two processes that publish on the same channel do so through different connections, and their messages are not ordered relative to each other.

Stream from a Temporal activity

A Temporal worker runs many activities at once, and an activity that calls a model can publish its output as the model produces it. The pool lets those activities share a few realtime connections, created when the worker starts. Temporal recommends creating clients such as this once per worker and passing them to the activities, so that activity code is safe to run concurrently and does not create connections of its own.

The Temporal plugin suits discrete calls from workflow code, such as status updates. The pool suits the stream itself, because separate REST publishes from the plugin are not guaranteed to arrive in order, while publishes through the pool on one channel are.

Create the activities

Write the activities as a factory that takes the pool, which is Temporal's pattern for passing dependencies to activities in TypeScript and JavaScript. The following activity streams a reply from the Vercel AI SDK, publishing each chunk as it arrives:

// Activity code, which runs on the worker.
import { streamText } from 'ai';

export function createAgentActivities(pool) {
  return {
    async streamReply({ channelName, prompt }) {
      const channel = pool.get(channelName);
      try {
        const { textStream } = streamText({ model: 'anthropic/claude-sonnet-4.5', prompt });

        const published = [];
        const track = (promise) => {
          // Mark the publish as handled, so that a rejection during the stream is not an
          // unhandled rejection. Promise.all below still rejects with it.
          promise.catch(() => {});
          published.push(promise);
        };

        let text = '';
        for await (const chunk of textStream) {
          text += chunk;
          track(channel.publish('agent.chunk', chunk));
        }
        track(channel.publish('agent.done', { text }));

        await Promise.all(published);
        return text;
      } finally {
        channel.release();
      }
    },
  };
}

The activity does not wait for each publish before reading the next chunk, so the stream is not slowed by a round trip per message. The publishes go out over one connection in the order they were made. Awaiting them all before the activity returns means that a failed publish fails the activity, and Temporal retries it.

Open the pool with the worker

Open the pool before the worker starts, pass it to the activity factory, and close it once worker.run() returns. Temporal waits up to the worker's shutdownGraceTime for running activities to finish before run() returns, so every activity has released its handles by the time the pool closes:

// Worker code.
import { fileURLToPath } from 'node:url';
import { NativeConnection, Worker } from '@temporalio/worker';
import { createRealtimeChannelPool } from '@ably/pubsub-server';
import { AblyHttpPlugin } from '@ably/pubsub-server/temporal';
import { createAgentActivities } from './activities.js';

const pool = createRealtimeChannelPool({ key: process.env.ABLY_API_KEY }, { size: 4 });
await pool.open();

const connection = await NativeConnection.connect({ address: 'localhost:7233' });

const worker = await Worker.create({
  connection,
  namespace: 'default',
  taskQueue: 'agents',
  workflowsPath: fileURLToPath(new URL('./workflows.js', import.meta.url)),
  activities: createAgentActivities(pool),
  // Optional: lets workflow code make discrete Ably calls, such as status updates.
  plugins: [new AblyHttpPlugin({ clientOptions: { key: process.env.ABLY_API_KEY } })],
});

try {
  await worker.run();
} finally {
  await pool.close();
}

Call the activity from the workflow

The workflow schedules the streaming activity like any other. The following workflow also uses the Temporal plugin to publish status updates around it:

// Workflow code.
import { proxyActivities, workflowInfo } from '@temporalio/workflow';
import { TemporalAblyHttpClient } from '@ably/pubsub-server/temporal/workflow';

const { streamReply } = proxyActivities({ startToCloseTimeout: '5 minutes' });

const ably = new TemporalAblyHttpClient();

export async function replyWorkflow({ prompt }) {
  const channelName = `chat:${workflowInfo().workflowId}`;
  const channel = ably.channels.get(channelName);

  await channel.publish('agent.status', { status: 'thinking' });
  const text = await streamReply({ channelName, prompt });
  await channel.publish('agent.status', { status: 'complete' });

  return { text };
}

Because an activity depends only on the object passed to the factory, a unit test can pass a fake pool whose get() returns a handle that records what is published, with no Ably connection and no Temporal server.

Handle a retried activity

When a publish fails, the activity fails and Temporal retries it. The retry gets a new handle, and each call on it resolves its client afresh, so it publishes through a connected client if there is one.

A retried activity publishes its stream again. Subscribers receive a second copy of the chunks, after any that the failed attempt published. Include the attempt number, from Context.current().info.attempt in @temporalio/activity, in the messages so that clients can discard an earlier attempt's output.

FAQ

Why not use one realtime client for the whole process?

One connection accepts a limited rate of inbound messages, which a process streaming several responses at once soon reaches. A shared client also shares its channels, so one operation's channel options or attach state affect every other operation on the same channel. The pool spreads channel names across several connections and gives each operation a handle that holds no channel state.

Why not create a realtime client in each operation?

A new client opens a new connection, requests a new token if you use token auth, and starts again from the primary host. During an outage of the primary host, each new client waits for a request to time out before it tries a fallback host. The pool keeps its clients, and their state, for the life of the process.

Can I subscribe through the pool?

No. A handle only publishes, which keeps it free of channel state that concurrent operations could conflict over. Use a separate realtime client to subscribe.

Read next

• Temporal plugin: call the Ably REST API from Temporal workflow code.
• AI Transport with Temporal: stream AI agent responses to users from Temporal activities.
• Message ordering: the ordering guarantees Ably provides.
• Limits: the connection and message rate limits that size the pool.