Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 2 additions & 0 deletions docs/runtime/streams.mdx
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,8 @@ const stream = new ReadableStream({

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

A direct `ReadableStream` needs a `pull()` function. Bun does not call `start()` on a direct source, so `pull()` is where you write. Without `pull()`, the `ReadableStream` constructor throws a `TypeError`.

### 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`.
Expand Down
3 changes: 3 additions & 0 deletions packages/bun-types/bun.d.ts
Original file line number Diff line number Diff line change
Expand Up @@ -603,6 +603,9 @@ declare module "bun" {
*
* A reader (`getReader()`, `for await`, `pipeTo()`) calls `pull()` again
* for a later read, once the previous call has settled.
*
* `pull` is required: a direct stream does not call `start()`. Without
* `pull`, the `ReadableStream` constructor throws a `TypeError`.
*/
pull: (controller: ReadableStreamDirectController) => void | PromiseLike<void>;
type: "direct";
Expand Down
4 changes: 0 additions & 4 deletions src/jsc/bindings/webcore/streams/BunStreamConsumers.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -1075,10 +1075,6 @@ static JSValue oneShotCallPull(JSC::VM& vm, JSGlobalObject* globalObject, JSDire
{
auto scope = DECLARE_THROW_SCOPE(vm);
JSObject* pullFunction = source->pullFunction();
if (!pullFunction) [[unlikely]] {
throwTypeError(globalObject, scope, "The 'pull' method of a direct ReadableStream's underlying source is not a function"_s);
return {};
}
MarkedArgumentBuffer arguments;
arguments.append(sink);
RELEASE_AND_RETURN(scope, JSC::call(globalObject, pullFunction, JSC::getCallData(pullFunction), source->thisValue(), arguments));
Expand Down
5 changes: 0 additions & 5 deletions src/jsc/bindings/webcore/streams/BunStreamSource.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -731,11 +731,6 @@ static JSValue readDirectStream(JSGlobalObject* globalObject, JSReadableStream*
stream->m_bunMode = BunStreamMode::Default;

JSObject* pull = source->pullFunction();
if (!pull) {
sinkController->end(globalObject);
RETURN_IF_EXCEPTION(scope, {});
return jsUndefined();
}

stream->m_controller.set(vm, stream, sinkController);
stream->m_controllerKind = ControllerKind::NativeSink;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -51,9 +51,10 @@ void JSDirectStreamSource::finishCreation(VM& vm, JSValue underlyingSource, JSOb
{
Base::finishCreation(vm);
ASSERT(inherits(info()));
ASSERT(pull);
auto valueOrNull = [](JSObject* object) { return object ? JSValue(object) : jsNull(); };
Base::internalField(static_cast<uint32_t>(Field::UnderlyingSource)).set(vm, this, underlyingSource);
Base::internalField(static_cast<uint32_t>(Field::Pull)).set(vm, this, valueOrNull(pull));
Base::internalField(static_cast<uint32_t>(Field::Pull)).set(vm, this, pull);
Base::internalField(static_cast<uint32_t>(Field::Cancel)).set(vm, this, valueOrNull(cancel));
Base::internalField(static_cast<uint32_t>(Field::Close)).set(vm, this, valueOrNull(close));
}
Expand Down
3 changes: 2 additions & 1 deletion src/jsc/bindings/webcore/streams/JSDirectStreamSource.h
Original file line number Diff line number Diff line change
Expand Up @@ -22,8 +22,9 @@ class JSDirectStreamSource final : public JSC::JSInternalFieldObjectImpl<4> {
enum class Field : uint32_t {
// The user's underlyingSource object, or an internal source's context cell.
UnderlyingSource = 0,
// Each is a callable or null.
// A callable: the ReadableStream constructor rejects a direct source without pull().
Pull,
// Each is a callable or null.
Cancel,
// Bun's close(reason) lifecycle hook: the stream ended or errored.
Close,
Expand Down
4 changes: 3 additions & 1 deletion src/jsc/bindings/webcore/streams/JSReadableStream.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -335,13 +335,15 @@ template<> JSC::EncodedJSValue JSC_HOST_CALL_ATTRIBUTES JSReadableStreamConstruc

switch (source.type) {
case BunUnderlyingSourceType::Direct: {
if (!source.dict.pull)
return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'pull' property is required when 'type' is \"direct\": a direct stream does not call start()"_s);
Comment thread
robobun marked this conversation as resolved.
Resolved
// Bun's close(reason) hook is not a dictionary member; it is converted like one.
JSValue close = asObject(underlyingSource)->get(lexicalGlobalObject, builtinNames(vm).closePublicName());
RETURN_IF_EXCEPTION(scope, {});
if (!close.isUndefined() && !close.isCallable())
return throwVMTypeError(lexicalGlobalObject, scope, "The underlying source's 'close' property must be a function"_s);
auto* directSource = JSDirectStreamSource::create(vm, JSStreamsRuntime::from(lexicalGlobalObject)->directStreamSourceStructure(defaultGlobalObject(lexicalGlobalObject)), underlyingSource,
source.dict.pull ? source.dict.pull.getObject() : nullptr, source.dict.cancel ? source.dict.cancel.getObject() : nullptr, close.isUndefined() ? nullptr : close.getObject());
source.dict.pull.getObject(), source.dict.cancel ? source.dict.cancel.getObject() : nullptr, close.isUndefined() ? nullptr : close.getObject());
// A direct stream has no controller yet; materializeIfNeeded() builds it on first use.
stream->m_bunMode = BunStreamMode::DirectPending;
stream->m_directSource.set(vm, stream, directSource);
Expand Down
19 changes: 19 additions & 0 deletions test/js/web/streams/streams.test.js
Original file line number Diff line number Diff line change
Expand Up @@ -1416,6 +1416,25 @@ describe("multi-chunk consumers produce exactly the concatenated bytes", () => {
expect(reads.toSorted()).toEqual(["cancel", "close", "pull"]);
expect(() => new ReadableStream({ type: "direct", pull() {}, close: 1 })).toThrow(TypeError);
});

// A direct stream does not run start(), so a source without pull() can never write or close.
it.each([
["no methods", { type: "direct" }],
["pull: undefined", { type: "direct", pull: undefined }],
["start() only", { type: "direct", start() {} }],
["cancel() only", { type: "direct", cancel() {} }],
])("a source without pull() does not construct: %s", (_label, source) => {
let thrown;
try {
new ReadableStream(source);
} catch (error) {
thrown = error;
}
expect({ name: thrown?.name, message: thrown?.message }).toEqual({
name: "TypeError",
message: `The underlying source's 'pull' property is required when 'type' is "direct": a direct stream does not call start()`,
});
});
});

// A type:"direct" pull() is re-invoked per read as a demand signal, but never while a
Expand Down
Loading