@portable/websockets

Functions

`WebSockets.connect`

Opens a WebSocket that reconnects when the connection drops, and returns its readable and writable sides. Both streams stay the same across reconnects.

Writes made while disconnected are queued and sent after the next connection's open finishes. Messages the server sent while the connection dropped, and writes the old socket accepted but never delivered, are lost. Add acknowledgements or sequence numbers to your protocol if that matters.

function connect(
  url: connect.Url,
  options: connect.Options = {},
): Duplex;

Option	Type	Default	Description
`protocols`	`string \| readonly string[]`		Subprotocols to request, as for `new WebSocket(url, protocols)`.
`signal`	`AbortSignal`		Closes the socket with code 1000 and fails both streams with `signal.reason`.
`delay`	`any`	`{ initial: 1000, max: 30000 }`	Wait before each reconnect.
`attempts`	`number`	`Infinity`	The number of consecutive failed attempts, counting the first, after which `connect` stops reconnecting and fails the streams with the last error. A connection that finishes `open` resets the count.
`timeout`	`number`	`10000`	Time limit for each attempt to open and finish `open`, in ms.
`open`	`((context: Context) => void \| PromiseLike<void>)`		Runs on every connection, after the socket opens and before queued writes are sent. Use it to authenticate or resubscribe. If it throws, the socket closes and reconnects after `delay`.
`reconnectIf`	`((close: Close, retry: { attempt: number; }) => boolean \| PromiseLike<boolean>)`		Called when the connection closes without the client calling `close`, with the close and `{ attempt }`, the number of failures in a row including this one. It returns whether `connect` reconnects. By default, codes 1000, 1002, 1003, 1007, 1008, 1009 and 1010 are final and every other code reconnects. If it throws, both streams fail with that error.
`maxQueued`	`number`	`1000`	Writes to hold while disconnected. Further writes wait.
`highWaterMark`	`number`	`1048576`	Writes wait while the socket's `bufferedAmount` is above this many bytes.
`maxBuffered`	`number`	`Infinity`	Most bytes (characters for text) of received messages to hold while the reader is behind. Past it, both streams fail with `MaxBufferedError` and the socket closes.
`heartbeat`	`Heartbeat`		Sends `ping` every `interval` ms and reconnects when no `pong` arrives within `timeout` ms.
`WebSocket`	`Constructor`	`globalThis.WebSocket`	The WebSocket class to use.

import * as Streams from '@portable/streams';
import * as WebSockets from '@portable/websockets';
 
const { readable, writable } = WebSockets.connect(
  'wss://example.com/feed',
  {
    async open({ send }) {
      send(JSON.stringify({ subscribe: 'prices' }));
    },
  },
);
const writer = writable.getWriter();
await writer.write('hello');
for await (const message of Streams.toAsyncIterable(readable)) {
  console.log(message);
}

`WebSockets.fromSocket`

Wraps an existing WebSocket, such as one a server accepted, in a readable and a writable stream. It never reconnects. Writes wait until the socket opens.

The readable closes when the socket closes with code 1000, 1001 or 1005, and fails with a CloseError for any other code. The writable fails with a CloseError once the socket closes.

function fromSocket(
  socket: Socket,
  options: fromSocket.Options = {},
): Duplex;

Option	Type	Default	Description
`signal`	`AbortSignal`		Closes the socket with code 1000 and fails both streams with `signal.reason`.
`highWaterMark`	`number`	`1048576`	Writes wait while the socket's `bufferedAmount` is above this many bytes.
`maxBuffered`	`number`	`Infinity`	Most bytes (characters for text) of received messages to hold while the reader is behind. Past it, both streams fail with `MaxBufferedError` and the socket closes.

import * as WebSockets from '@portable/websockets';
 
export default {
  fetch(request: Request) {
    const [client, server] = Object.values(new WebSocketPair());
    server.accept();
    const { readable, writable } = WebSockets.fromSocket(server);
    readable.pipeTo(writable); // echo
    return new Response(null, { status: 101, webSocket: client });
  },
};

Errors

`WebSockets.CloseError`

Thrown when the other side closes the connection, or when the connection closes and connect stops reconnecting.

class CloseError extends Errors.BaseError {
  constructor(options: { code: number; reason: string });
  readonly kind: string;
  readonly code: number;
  readonly reason: string;
}

import * as Streams from '@portable/streams';
import * as WebSockets from '@portable/websockets';
 
const { readable } = WebSockets.fromSocket(
  new WebSocket('wss://example.com/feed'),
);
try {
  for await (const message of Streams.toAsyncIterable(readable)) {
    console.log(message);
  }
} catch (error) {
  if (error instanceof WebSockets.CloseError) {
    console.log(error.code, error.reason);
  }
}

`WebSockets.MaxBufferedError`

Thrown when received messages waiting to be read exceed maxBuffered.

class MaxBufferedError extends Errors.BaseError {
  constructor(options: { maxBuffered: number });
  readonly kind: string;
  readonly maxBuffered: number;
}

import * as Streams from '@portable/streams';
import * as WebSockets from '@portable/websockets';
 
const { readable } = WebSockets.connect('wss://example.com/feed', {
  maxBuffered: 1_000_000,
});
try {
  for await (const message of Streams.toAsyncIterable(readable)) {
    await handle(message);
  }
} catch (error) {
  if (error instanceof WebSockets.MaxBufferedError) {
    console.log(`fell ${error.maxBuffered} behind`);
  }
}

Types

`WebSockets.Duplex`

The readable and writable sides of a WebSocket. Where the runtime has Symbol.asyncDispose, [Symbol.asyncDispose]() closes the socket with code 1000, stops reconnecting, and fails both streams with an AbortError DOMException, as aborting signal does, so a connection declared with await using is closed when the scope ends. Without the symbol, abort signal to close it, even while a reader or writer holds a lock.

type Duplex = AsyncDisposeMethod & {
  readable: ReadableStream<Message>;
  writable: WritableStream<string | Uint8Array>;
};

`WebSockets.Message`

A received message: text as a string, binary data as a Uint8Array.

type Message = string | Uint8Array<ArrayBuffer>;

`WebSockets.Socket`

The part of the platform WebSocket this package uses. Any object with this shape works.

type Socket = {
  readonly readyState: number;
  readonly bufferedAmount: number;
  binaryType: string;
  send(data: string | Uint8Array<ArrayBuffer>): void;
  close(code?: number, reason?: string): void;
  addEventListener(
    type: string,
    listener: (event: SocketEvent) => void,
  ): void;
  removeEventListener(
    type: string,
    listener: (event: SocketEvent) => void,
  ): void;
};