Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
31 commits
Select commit Hold shift + click to select a range
97aa4dc
Bun.serve direct streams: flush on controller.close(), end the respon…
Jarred-Sumner Sep 9, 2026
8a47ae5
Sink controllers call their close/ready handler directly; an async di…
Jarred-Sumner Sep 9, 2026
a93c239
One-line comment on directStreamOnClose
robobun Sep 9, 2026
fc40ef8
Native stream sources notify their C++ adapter on close directly; dro…
Jarred-Sumner Sep 9, 2026
fc8979b
FileSink::close_with_error: hold a ref across end_from_stream
robobun Sep 9, 2026
458d4a6
Direct streams: pull() runs once for readers too; close(error) fails …
Jarred-Sumner Sep 9, 2026
7cd62d8
close(error) rejects the sink owner's promise; server contract cells …
Jarred-Sumner Sep 9, 2026
4dee616
close(error) framing test: the response is aborted, not terminated
Jarred-Sumner Sep 9, 2026
b4227a5
[autofix.ci] apply automated fixes
autofix-ci[bot] Sep 9, 2026
7567030
Direct stream edge-case tests: matrix shapes, pipes, generators, node…
Jarred-Sumner Sep 9, 2026
093df49
A pull() that rejects after the sink already detached has nothing to …
Jarred-Sumner Sep 9, 2026
985a3a8
abort-while-parked test: keep writing until backpressure so the abort…
Jarred-Sumner Sep 9, 2026
ed20a6a
Quiet the expected body-error logs in the serve contract and edge-cas…
Jarred-Sumner Sep 9, 2026
ffe8284
Bun.serve: a body that fails after the status line is committed goes …
Jarred-Sumner Sep 9, 2026
0d4dbcf
Direct stream controller: write()/flush() after close or peer disconn…
Jarred-Sumner Sep 9, 2026
2190402
[autofix.ci] apply automated fixes
autofix-ci[bot] Sep 9, 2026
30c33a8
test: read the Bun.write output before the temp dir is disposed; allo…
robobun Sep 9, 2026
6dc9020
Update two serve tests to the write()-after-close → 0 contract
Jarred-Sumner Sep 9, 2026
42772f3
CI follow-ups: Bake SSR closes its direct stream with the error; HTTP…
Jarred-Sumner Sep 9, 2026
89c3c18
serve-stream-body-error: error()'s own body failing after commit is p…
Jarred-Sumner Sep 9, 2026
4540402
[autofix.ci] apply automated fixes
autofix-ci[bot] Sep 9, 2026
35b93cf
serve.test: a body that throws in pull() now reaches error() (whose R…
Jarred-Sumner Sep 9, 2026
f4f932d
[autofix.ci] apply automated fixes
autofix-ci[bot] Sep 9, 2026
71df837
Review follow-ups: end(value) stays a clean close on the reader and o…
Jarred-Sumner Sep 9, 2026
bd51ca9
serve: keep error() bypassed for a body that fails after the status l…
Jarred-Sumner Sep 9, 2026
819d83f
Direct-stream controller: write()/flush() return 0 only when the dest…
Jarred-Sumner Sep 9, 2026
1b48359
Direct streams: a reader keeps calling pull() on demand; only whole-b…
Jarred-Sumner Sep 9, 2026
11fe484
Review follow-ups: a sync pull() that calls close(error) rejects the …
Jarred-Sumner Sep 9, 2026
0839195
[autofix.ci] apply automated fixes
autofix-ci[bot] Sep 9, 2026
c671562
FileSink::pipe_stream: release the stream Strong when the pump fails …
Jarred-Sumner Sep 9, 2026
c70a379
HTTPServerWritable::fail() drops the buffered tail instead of parking…
Jarred-Sumner Sep 9, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion docs/runtime/streams.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -64,15 +64,24 @@ const stream = new ReadableStream({
pull(controller) {
controller.write("hello");
controller.write("world");
controller.close();
},
});
```

When using a direct `ReadableStream`, the destination handles all chunk queueing. The destination receives the bytes you pass to `controller.write()`. When the stream is read from JavaScript, Bun buffers the writes and delivers them as `Uint8Array` chunks (strings are UTF-8 encoded).

### Ending the stream

Call `controller.close()` (or `controller.end()`) once everything is written. Every destination flushes what it has buffered and finishes at that point. `controller.close(error)` fails the stream instead: nothing more is flushed and the consumer rejects with `error`.

A destination that takes the whole body (`Bun.serve`, `fetch`, `Bun.write`, `Bun.spawn` stdin, `.text()`, and so on) calls `pull()` once. If it returns a promise, the stream stays open while that promise is pending and ends when it resolves, as if `pull()` had called `controller.close()` last; if it rejects, the stream errors with that reason. A `pull()` that returns without a promise and without closing keeps the stream open until something calls `controller.close()`, so you can hold on to the controller and write from events.

When the stream is read through a reader (`getReader()`, `for await`, `pipeTo()`, `tee()`), `pull()` is a demand signal instead: it is called again for a later read, never while a previous call's promise is still pending, until `controller.close()`.

### Handling backpressure

`controller.write()` returns the number of bytes written, or a **pending `Promise<number>`** when the destination's internal buffer is full (for example, a slow HTTP client). The chunk is accepted either way. The promise resolves once the destination has drained, so `await`ing the result is enough:
`controller.write()` returns the number of bytes written, or a **pending `Promise<number>`** when the destination's internal buffer is full (for example, a slow HTTP client). The chunk is accepted either way. Once the destination has gone away (for example the HTTP client disconnected), `write()` returns `0`. The promise resolves once the destination has drained, so `await`ing the result is enough:

```ts
const stream = new ReadableStream({
Expand Down
2 changes: 1 addition & 1 deletion mordant-baseline.toml
Original file line number Diff line number Diff line change
Expand Up @@ -140,7 +140,7 @@
"unused_pub:src/io/ParentDeathWatchdog.rs" = 1
"unused_pub:src/io/PipeWriter.rs" = 2
"unused_pub:src/jsc/bindgen.rs" = 6
"unused_pub:src/jsc/host_fn.rs" = 1
"unused_pub:src/jsc/host_fn.rs" = 2
Comment thread
Jarred-Sumner marked this conversation as resolved.
Resolved
"unused_pub:src/jsc/lib.rs" = 1
"unused_pub:src/jsc/webcore_types.rs" = 2
"unused_pub:src/libarchive/lib.rs" = 1
Expand Down
13 changes: 13 additions & 0 deletions packages/bun-types/bun.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -579,6 +579,19 @@ declare module "bun" {

interface DirectUnderlyingSource<R = any> {
cancel?: UnderlyingSourceCancelCallback;
/**
* Write the stream's data with `controller.write()` (await it for
* backpressure) and finish with `controller.close()`.
*
* A destination that takes the whole body (`Bun.serve`, `Bun.write`,
* `.text()`, ...) calls `pull()` once. If it returns a promise, the stream
* stays open while it is pending, ends when it resolves, and errors if it
* rejects. If it returns synchronously without closing, the stream stays
* open until `controller.close()` is called.
*
* A reader (`getReader()`, `for await`, `pipeTo()`) calls `pull()` again
* for a later read, once the previous call has settled.
*/
pull: (controller: ReadableStreamDirectController) => void | PromiseLike<void>;
type: "direct";
}
Expand Down
8 changes: 8 additions & 0 deletions packages/bun-types/globals.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -727,6 +727,11 @@ interface ReadableStreamDefaultController<R = any> {
}

interface ReadableStreamDirectController {
/**
* Finish the stream. With no argument this flushes buffered bytes and ends,
* like {@link end}. With an `error`, the stream fails: buffered bytes are
* dropped and the consumer rejects with `error`.
*/
close(error?: Error): void;
/**
* Write a chunk directly to the destination.
Expand All @@ -741,6 +746,9 @@ interface ReadableStreamDirectController {
*
* The promise resolves once the destination has drained.
* `await controller.flush(true)` is equivalent.
*
* Returns `0` once the destination has gone away (for example, the HTTP
* client disconnected).
*/
write(data: Bun.BufferSource | ArrayBuffer | string): number | Promise<number>;
end(): number | Promise<number>;
Expand Down
Loading
Loading