# Realtime channel pool

Publish messages concurrently through a fixed pool of realtime connections, while maintaining message order on each channel.

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](https://ably.com/docs/platform/architecture/message-ordering.md#rest-considerations).

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:

<Code>

#### Javascript

```
// 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();
  }
}
```
</Code>

## 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](https://ably.com/docs/pub-sub/advanced.md#transient-publish). 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:

<Code>

### Shell

```
npm install @ably/pubsub-server
```
</Code>

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

<Code>

### Javascript

```
// 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();
```
</Code>

The first argument is a [client options](https://ably.com/docs/api/realtime-sdk.md#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:

| Option | Description | Default |
| --- | --- | --- |
| `size` | The number of realtime clients, and so of connections, that the pool holds. | `4` |
| `failedRetryTimeout` | How long, in milliseconds, the pool waits before it reconnects a client whose connection has [failed](#connection-states). | `30000` |

`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.

> **Important**
>
> Clients created by `@ably/pubsub-server` declare themselves as server-side clients. With an API key, nothing more is needed. If the pool authenticates with token auth, through `authUrl` or `authCallback`, each token must carry a signed `x-ably-clientType=server` claim, or Ably rejects the connection. Add the claim to the tokens your auth server issues before you deploy.

### Choose the pool size 

Each realtime connection accepts a limited [inbound message rate](https://ably.com/docs/platform/pricing/limits.md#connection). 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](https://ably.com/docs/platform/pricing/limits.md#connection). 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:

<Code>

### Javascript

```
// 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();
}
```
</Code>

`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:

<Code>

### Javascript

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

`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:

| Member | Description |
| --- | --- |
| `name` | The channel name. |
| `publish()` | Publishes a message or an array of messages, as [`publish()`](https://ably.com/docs/pub-sub.md#publish) on a realtime channel does. |
| `updateMessage()`, `deleteMessage()`, `appendMessage()` | [Updates, deletes or appends to](https://ably.com/docs/messages/updates-deletes.md) a message. |
| `annotations.publish()`, `annotations.delete()` | Publishes or deletes an [annotation](https://ably.com/docs/messages/annotations.md) on a message. |
| `release()` | Drops this handle's hold on the channel. |

> **Note**
>
> `get()` takes no channel options, because a channel shared by concurrent operations cannot carry options that differ between them. This means a handle cannot publish to an encrypted channel.

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](https://ably.com/docs/connect/states.md):

| Connection state | What the client does | What happens to a call |
| --- | --- | --- |
| `connecting` or `disconnected` | Retries the connection by itself. | Queued, and sent once the client connects. |
| `suspended` | Retries the connection by itself, less often. | Rejects with the connection's `errorReason`. |
| `failed` | Does not retry, because the cause, such as an invalid or revoked key, is still there on a retry. The pool calls `connect()` on it every `failedRetryTimeout`, so a long-running process recovers once the cause is fixed. | Rejects with the connection's `errorReason`. |

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:

<Code>

### Javascript

```
// Server-side code.
await pool.close();
```
</Code>

`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`](https://ably.com/docs/platform/errors.md) in the following cases:

| Code | Cause |
| --- | --- |
| `40000` | A method was called on a handle before `open()` was called on the pool, or after the handle was released. |
| `80017` | A method was called on a handle after `close()` was called on the pool. `open()` also rejects with this code if `close()` is called before it completes, or if the pool has already been closed. |
| `40003` | `createRealtimeChannelPool()` was given an invalid `size`, which must be a whole number of 1 or more, or an invalid `failedRetryTimeout`, which must be a number of milliseconds of 0 or more. Thrown when the pool is created. |
| `90000` | A call was made while the connection was connecting or disconnected, with `queueMessages` set to `false`, and the connection had no `errorReason`. |

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](https://ably.com/docs/pub-sub/temporal.md) 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:

<Code>

#### Javascript

```
// 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();
      }
    },
  };
}
```
</Code>

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:

<Code>

#### Javascript

```
// 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();
}
```
</Code>

### 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:

<Code>

#### Javascript

```
// 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 };
}
```
</Code>

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.

> **Further reading**
>
> [AI Transport](https://ably.com/docs/ai-transport/durable-execution/temporal.md) builds on this pattern for AI agents. Its activities add headers to every message they publish, so that clients replace a failed attempt's partial response with the retry's.

## 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](https://ably.com/docs/pub-sub/temporal.md): call the Ably REST API from Temporal workflow code.
- [AI Transport with Temporal](https://ably.com/docs/ai-transport/durable-execution/temporal.md): stream AI agent responses to users from Temporal activities.
- [Message ordering](https://ably.com/docs/platform/architecture/message-ordering.md): the ordering guarantees Ably provides.
- [Limits](https://ably.com/docs/platform/pricing/limits.md#connection): the connection and message rate limits that size the pool.

## Related Topics

- [Temporal plugin](https://ably.com/docs/pub-sub/temporal.md): Call the Ably REST API from Temporal workflow code. Each call runs as a Temporal activity on your worker, with deterministic message IDs so that a retried publish is not applied twice.

## Documentation Index

To discover additional Ably documentation:

1. Fetch [llms.txt](https://ably.com/llms.txt) for the canonical list of available pages.
2. Identify relevant URLs from that index.
3. Fetch target pages as needed.

Avoid using assumed or outdated documentation paths.