Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

stream: add isErrored helper #41121

Closed
wants to merge 6 commits into from
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
13 changes: 13 additions & 0 deletions doc/api/stream.md
Expand Up @@ -2225,6 +2225,19 @@ added: v16.8.0

Returns whether the stream has been read from or cancelled.

### `stream.Readable.isErrored(stream)`
ronag marked this conversation as resolved.
Show resolved Hide resolved

<!-- YAML
added: v16.8.0
ronag marked this conversation as resolved.
Show resolved Hide resolved
-->

> Stability: 1 - Experimental

* `stream` {stream.Readable|ReadableStream}
ronag marked this conversation as resolved.
Show resolved Hide resolved
* Returns: `boolean`
ronag marked this conversation as resolved.
Show resolved Hide resolved

Returns whether the stream has been errored.
ronag marked this conversation as resolved.
Show resolved Hide resolved

### `stream.Readable.toWeb(streamReadable)`

<!-- YAML
Expand Down
20 changes: 17 additions & 3 deletions lib/internal/streams/utils.js
Expand Up @@ -7,6 +7,7 @@ const {
} = primordials;

const kDestroyed = Symbol('kDestroyed');
const kIsErrored = Symbol('kIsErrored');
const kIsDisturbed = Symbol('kIsDisturbed');

function isReadableNodeStream(obj, strict = false) {
Expand Down Expand Up @@ -239,16 +240,29 @@ function willEmitClose(stream) {

function isDisturbed(stream) {
return !!(stream && (
stream.readableDidRead ||
stream.readableAborted ||
stream[kIsDisturbed]
stream[kIsDisturbed] ??
(stream.readableDidRead || stream.readableAborted)
));
}

function isErrored(stream) {
return !!(stream && (
stream[kIsErrored] ??
stream.readableErrored ??
stream.writableErrored ??
stream._readableState?.errorEmitted ??
stream._writableState?.errorEmitted ??
stream._readableState?.errored ??
stream._writableState?.errored
));
}

module.exports = {
kDestroyed,
isDisturbed,
isErrored,
kIsDisturbed,
kIsErrored,
isClosed,
isDestroyed,
isDuplexNodeStream,
Expand Down
5 changes: 5 additions & 0 deletions lib/internal/webstreams/readablestream.js
Expand Up @@ -82,6 +82,7 @@ const {

const {
kIsDisturbed,
kIsErrored,
} = require('internal/streams/utils');

const {
Expand Down Expand Up @@ -241,6 +242,10 @@ class ReadableStream {
return this[kState].disturbed;
}

get [kIsErrored]() {
return this[kState].state === 'errored';
}

/**
* @readonly
* @type {boolean}
Expand Down
4 changes: 3 additions & 1 deletion lib/stream.js
Expand Up @@ -36,9 +36,11 @@ const eos = require('internal/streams/end-of-stream');
const internalBuffer = require('internal/buffer');

const promises = require('stream/promises');
const utils = require('internal/streams/utils');

const Stream = module.exports = require('internal/streams/legacy').Stream;
Stream.isDisturbed = require('internal/streams/utils').isDisturbed;
Stream.isDisturbed = utils.isDisturbed;
Stream.isErrored = utils.isErrored;
Stream.Readable = require('internal/streams/readable');
Stream.Writable = require('internal/streams/writable');
Stream.Duplex = require('internal/streams/duplex');
Expand Down
7 changes: 5 additions & 2 deletions test/parallel/test-stream-readable-didRead.js
@@ -1,15 +1,18 @@
'use strict';
const common = require('../common');
const assert = require('assert');
const { isDisturbed, Readable } = require('stream');
const { isDisturbed, isErrored, Readable } = require('stream');

function noop() {}

function check(readable, data, fn) {
assert.strictEqual(readable.readableDidRead, false);
assert.strictEqual(isDisturbed(readable), false);
assert.strictEqual(isErrored(readable), false);
if (data === -1) {
readable.on('error', common.mustCall());
readable.on('error', common.mustCall(() => {
assert.strictEqual(isErrored(readable), true);
}));
readable.on('data', common.mustNotCall());
readable.on('end', common.mustNotCall());
} else {
Expand Down
19 changes: 18 additions & 1 deletion test/parallel/test-whatwg-readablestream.js
Expand Up @@ -2,7 +2,7 @@
'use strict';

const common = require('../common');
const { isDisturbed } = require('stream');
const { isDisturbed, isErrored } = require('stream');
const assert = require('assert');
const {
isPromise,
Expand Down Expand Up @@ -1572,3 +1572,20 @@ class Source {
isDisturbed(stream, true);
})().then(common.mustCall());
}


{
const stream = new ReadableStream({
start(controller) {
controller.error(new Error());
},
pull: common.mustNotCall(),
});

const reader = stream.getReader();
(async () => {
isErrored(stream, false);
await reader.read();
isErrored(stream, true);
})().then(common.mustCall());
}