DOCSEARCHbyAlgolia
Submit Docs

resumable-stream

repository·main·Indexed 17 days ago

https://github.com/vercel/resumable-stream

A library for wrapping string streams (such as SSE) to enable client resumption and multi-client following in serverless environments. It uses a Redis-based pubsub mechanism to coordinate between producers and consumers, specifically designed for environments lacking sticky load balancing. The library provides an idempotent API via `resumableStream` as well as explicit methods like `createNewResumableStream` and `resumeExistingStream` for granular lifecycle control. It supports standard redis, ioredis, and generic Redis-compatible clients like Upstash or Valkey.

Tokens
6.4K
Snippets
23
Records
30
Agent score
62%

What's inside resumable-stream

  1. What is resumable-stream and when to use it

    main

    resumable-stream is a library designed to wrap streams of strings (such as SSE web responses) to allow clients to resume them after a connection loss or to allow multiple clients to follow the same stream.

    It is specifically designed for serverless environments that lack sticky load balancing. The library uses a Redis-based pubsub mechanism to coordinate between producers and consumers. It is optimized for low latency and minimal Redis usage; in the common case where recovery is not needed, it only performs a single INCR and SUBSCRIBE per stream.

  2. Use ResumableStreamContext to manage streams

    main

    The ResumableStreamContext interface provides methods to create, check, and resume streams using a unique streamId. It allows you to handle long-running or interrupted string streams by either starting a new one or picking up where a previous one left off.

    Key methods include:

    • createNewResumableStream: Starts a fresh stream.
    • resumableStream: An idempotent method that either creates a new stream or resumes an existing one.
    • resumeExistingStream: Specifically targets resuming an existing stream.
    • hasExistingStream: Checks the current status of a streamId.

  3. How resumable-stream manages producers and consumers

    main

    The library uses a pubsub mechanism to coordinate stream access:

    1. Producer Creation: The first time a streamId is invoked, a standard stream is created. This instance acts as the producer. The producer is responsible for completing the stream even if the original reader disconnects.
    2. Listening: The producer starts listening on the pubsub for additional consumers.
    3. Consumer Connection: When a second client requests the same streamId, it publishes a message to the pubsub to alert the producer.
    4. Data Flow: The producer receives the alert, publishes any buffered messages to the new consumer, and then continues publishing new chunks of the stream via pubsub.

  4. Configure resumable-stream with ioredis

    main

    If your project uses ioredis instead of the standard redis package, import the context creator from the resumable-stream/ioredis entry point. This automatically configures the library to use ioredis as the default client.

    import { createResumableStreamContext } from "resumable-stream/ioredis";

const streamContext = createResumableStreamContext({
  waitUntil: after,
  // Optionally pass in your own Redis publisher and subscriber
});

  5. Configure resumable-stream with custom Redis clients (Upstash, Valkey, etc.)

    main

    To use a different Redis-compatible client (like Upstash or Valkey), import from resumable-stream/generic.

    Important: When using the generic interface, you must provide both a publisher and a subscriber implementation. The library will throw an error if they are missing. You must implement the Publisher and Subscriber interfaces.

    import { createResumableStreamContext } from "resumable-stream/generic";
import type { Publisher, Subscriber } from "resumable-stream/generic";

// Example: Create adapters for your Redis client
const publisher: Publisher = {
  connect: async () => { /* ... */ },
  publish: async (channel, message) => { /* ... */ },
  set: async (key, value, options) => { /* ... */ },
  get: async (key) => { /* ... */ },
  incr: async (key) => { /* ... */ },
};

const subscriber: Subscriber = {
  connect: async () => { /* ... */ },
  subscribe: async (channel, callback) => { /* ... */ },
  unsubscribe: async (channel) => { /* ... */ },
};

const streamContext = createResumableStreamContext({
  waitUntil: after,
  publisher,
  subscriber,
});

  6. Use waitUntil to manage process lifecycle in serverless environments

    main

    The waitUntil option is used to prevent the execution environment from suspending before asynchronous tasks are complete.

    • In Serverless/Edge environments: Pass a function that accepts a promise to ensure the runtime stays alive until that promise resolves.
    • In standard server environments: Pass null, as you do not need to worry about the function being suspended mid-execution.
    // Example for serverless environments
waitUntil: (promise) => {
  // logic to keep the process alive until promise resolves
}

  7. Use the Publisher interface

    main

    The Publisher interface provides a Redis-like API for publishing messages and managing keys. It is designed to be compatible with clients from both the redis and ioredis packages. You can use it to perform standard operations like setting values, incrementing counters, and publishing to channels.

    // Example of using Publisher methods
await publisher.set('my-key', 'my-value');
await publisher.incr('my-counter');
await publisher.publish('my-channel', 'hello world');
const val = await publisher.get('my-key');

  8. Use the Subscriber interface

    main

    The Subscriber interface defines a Redis-like subscriber compatible with clients from both the redis and ioredis packages. It provides methods to establish a connection, subscribe to specific channels with a callback, and unsubscribe from channels.

    Methods

    • connect(): Establishes the connection to the underlying data store. Returns a Promise<unknown>.
    • subscribe(channel, callback): Subscribes to a specific channel (string). When a message is received, the provided callback is executed. Returns a Promise<number | void>.
    • unsubscribe(channel): Stops listening to the specified channel (string). Returns a Promise<unknown>.
    // Example usage pattern for a Subscriber
const subscriber: Subscriber = getSubscriber();

await subscriber.connect();

const subscriptionId = await subscriber.subscribe('my-channel', (message) => {
  console.log('Received message:', message);
});

// Later, to stop listening
await subscriber.unsubscribe('my-channel');

  9. Use the Idempotent API for stream management

    main

    The Idempotent API provides a single method, resumableStream, that handles both the creation of a new stream and the resumption of an existing one based on the provided streamId.

    If the stream does not exist, it becomes the producer. If it does exist, it joins as a consumer. You can optionally provide a resumeAt index to start receiving content from a specific point.

    import { createResumableStreamContext } from "resumable-stream";
import { after } from "next/server";

const streamContext = createResumableStreamContext({
  waitUntil: after,
  // Optionally pass in your own Redis publisher and subscriber
});

export async function GET(req: NextRequest, { params }: { params: Promise<{ streamId: string }> }) {
  const { streamId } = await params;
  const resumeAt = req.nextUrl.searchParams.get("resumeAt");
  const stream = await streamContext.resumableStream(
    streamId,
    makeTestStream,
    resumeAt ? parseInt(resumeAt) : undefined
  );
  if (!stream) {
    return new Response("Stream is already done", {
      status: 422,
    });
  }
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
    },
  });
}

  10. Create a resumable stream context with createResumableStreamContext()

    main

    Use createResumableStreamContext(options) to initialize a global context for managing resumable streams. This context serves as the factory for all subsequent streams in your application. Once the context is created, you can instantiate individual streams by calling the .resumableStream() method on the returned context object.

    // Example of creating a context and then a stream
const context = createResumableStreamContext(options);
const stream = context.resumableStream(streamOptions);

  11. Resume a stream with resumeExistingStream()

    main

    Use resumeExistingStream to pick up a stream that was previously initialized via createNewResumableStream.

    Parameters:

    • streamId (string): Unique identifier.
    • skipCharacters (number, optional): Number of characters to skip.

    Returns:

    • Promise<ReadableStream<string>>: The resumed stream.
    • null: The stream with the given streamId is already DONE.
    • undefined: No stream exists with the given streamId.

  12. Create a new stream with createNewResumableStream()

    main

    Use createNewResumableStream to initialize a new stream. The makeStream function is only executed if a stream with the provided streamId is not already in progress.

    Parameters:

    • streamId (string): A unique identifier for the stream.
    • makeStream (() => ReadableStream<string>): A factory function that returns the stream of strings.
    • skipCharacters (number, optional): The number of characters to skip from the start.

    Returns:

    • A Promise resolving to a ReadableStream<string> or null if the stream with that ID is already marked as DONE (streams expire after 24 hours by default).