Use ReadableStream with Durable Object and Workers
- On this page

Use ReadableStream with Durable Object and Workers

Stream ReadableStream from Durable Objects.

Last updated Apr 21, 2026 This example demonstrates:

A Worker receives a request, and forwards it to a Durable Object my-id.
The Durable Object streams an incrementing number every second, until it receives AbortSignal.
The Worker reads and logs the values from the stream.
The Worker then cancels the stream after 5 values.

import { DurableObject } from "cloudflare:workers";

// Send incremented counter value every second
async function* dataSource(signal) {
    let counter = 0;
    while (!signal.aborted) {
        yield counter++;
        await new Promise((resolve) => setTimeout(resolve, 1_000));
    }

    console.log("Data source cancelled");
}

export class MyDurableObject extends DurableObject {
    async fetch(request) {
        const abortController = new AbortController();

        const stream = new ReadableStream({
            async start(controller) {
                if (request.signal.aborted) {
                    controller.close();
                    abortController.abort();
                    return;
                }

                for await (const value of dataSource(abortController.signal)) {
                    controller.enqueue(new TextEncoder().encode(String(value)));
                }
            },
            cancel() {
                console.log("Stream cancelled");
                abortController.abort();
            },
        });

        const headers = new Headers({
            "Content-Type": "application/octet-stream",
        });

        return new Response(stream, { headers });
    }
}

export default {
    async fetch(request, env, ctx) {
        const stub = env.MY_DURABLE_OBJECT.getByName("foo");
        const response = await stub.fetch(request, { ...request });
        if (!response.ok || !response.body) {
            return new Response("Invalid response", { status: 500 });
        }

        const reader = response.body
            .pipeThrough(new TextDecoderStream())
            .getReader();

        let data = [];
        let i = 0;
        while (true) {
            // Cancel the stream after 5 messages
            if (i > 5) {
                reader.cancel();
                break;
            }
            const { value, done } = await reader.read();

            if (value) {
                console.log(`Got value ${value}`);
                data = [...data, value];
            }

            if (done) {
                break;
            }
            i++;
        }

        return Response.json(data);
    },
};

import { DurableObject } from 'cloudflare:workers';

// Send incremented counter value every second
async function* dataSource(signal: AbortSignal) {
    let counter = 0;
    while (!signal.aborted) {
        yield counter++;
        await new Promise((resolve) => setTimeout(resolve, 1_000));
    }

    console.log('Data source cancelled');
}

export class MyDurableObject extends DurableObject<Env> {
    async fetch(request: Request): Promise<Response> {
        const abortController = new AbortController();

        const stream = new ReadableStream({
            async start(controller) {
                if (request.signal.aborted) {
                    controller.close();
                    abortController.abort();
                    return;
                }

                for await (const value of dataSource(abortController.signal)) {
                    controller.enqueue(new TextEncoder().encode(String(value)));
                }
            },
            cancel() {
                console.log('Stream cancelled');
                abortController.abort();
            },
        });

        const headers = new Headers({
            'Content-Type': 'application/octet-stream',
        });

        return new Response(stream, { headers });
    }

}

export default {
    async fetch(request, env, ctx): Promise<Response> {
        const stub = env.MY_DURABLE_OBJECT.getByName("foo");
        const response = await stub.fetch(request, { ...request });
        if (!response.ok || !response.body) {
            return new Response('Invalid response', { status: 500 });
        }

        const reader = response.body.pipeThrough(new TextDecoderStream()).getReader();

        let data = [] as string[];
        let i = 0;
        while (true) {
            // Cancel the stream after 5 messages
            if (i > 5) {
                reader.cancel();
                break;
            }
            const { value, done } = await reader.read();

            if (value) {
                console.log(`Got value ${value}`);
                data = [...data, value];
            }

            if (done) {
                break;
            }
            i++;
        }

        return Response.json(data);
    },

} satisfies ExportedHandler<Env>;

Note

In a setup where a Durable Object returns a readable stream to a Worker, if the Worker cancels the Durable Object's readable stream, the cancellation propagates to the Durable Object.

Was this helpful?

YesNo

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/durable-objects/examples/readable-stream/#page","headline":"Use ReadableStream with Durable Object and Workers","description":"Stream ReadableStream from Durable Objects.","url":"https://developers.cloudflare.com/durable-objects/examples/readable-stream/","inLanguage":"en","image":"https://developers.cloudflare.com/durable-objects/examples/readable-stream/og.png?v=45f9ab05c42f2e38","dateModified":"2026-04-21","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"}}

description: Pass metadata to a Durable Object using RpcTarget. title: Use RpcTarget class to handle Durable Object metadata image: https://developers.cloudflare.com/durable-objects/examples/reference-do-name-using-init/og.png?v=7abae1644e934869

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/durable-objects/llms.txt
Use this file to discover all available pages before exploring further.

Use ReadableStream with Durable Object a

Use ReadableStream with Durable Object and Workers

On this page