Add IsReadableStreamDisturbed predicate

See the suggestion in the OP of #378.

Author: tyoshino

index.bs

@@ -387,6 +387,7 @@ Instances of <code>ReadableStream</code> are created with the internal slots des
   1. Set *this*@[[state]] to "readable".
   1. Set *this*@[[started]], *this*@[[closeRequested]], *this*@[[pullAgain]], and *this*@[[pulling]] to *false*.
   1. Set *this*@[[reader]] and *this*@[[storedError]] to *undefined*.
+  1. Set *this*@[[disturbed]] to *false*.
   1. Set *this*@[[controller]] to Construct(`ReadableStreamController`, «*this*»).
   1. Let _normalizedStrategy_ be ValidateAndNormalizeQueuingStrategy(_size_, _highWaterMark_).
   1. Set *this*@[[strategySize]] to _normalizedStrategy_.[[size]] and *this*@[[strategyHWM]] to _normalizedStrategy_.[[highWaterMark]].
@@ -855,6 +856,7 @@ reader</a> for a given stream.
 <emu-alg>
   1. If _stream_@[[state]] is "closed", return a new promise resolved with *undefined*.
   1. If _stream_@[[state]] is "errored", return a new promise rejected with _stream_@[[storedError]].
+  1. Set _stream_@[[disturbed]] to *true*.
   1. Set _stream_@[[queue]] to a new empty List.
   1. Perform FinishClosingReadableStream(_stream_).
   1. Let _sourceCancelPromise_ be PromiseInvokeOrNoop(_stream_@[[underlyingSource]], "cancel", «‍_reason_»).
@@ -971,6 +973,16 @@ an assert).
   1. Return *true*.
 </emu-alg>
 
+<h4 id="is-readable-stream-disturbed" aoid="IsReadableStreamDisturbed">IsReadableStreamDisturbed ( stream )</h4>
+
+This abstract operation is meant to be called from other specifications that may wish to query whether or not a
+readable stream is ever attemped to be read or cancelled.
+
+<emu-alg>
+  1. Assert: IsReadableStream(_stream_) is *true*.
+  1. Return _stream_@[[disturbed]].
+</emu-alg>
+
 <h4 id="is-readable-stream-locked" aoid="IsReadableStreamLocked">IsReadableStreamLocked ( stream )</h4>
 
 This abstract operation is meant to be called from other specifications that may wish to query whether or not a
@@ -997,6 +1009,7 @@ readable stream is <a>locked to a reader</a>.
   1. If _reader_@[[state]] is "errored", return a new promise rejected with _reader_@[[storedError]].
   1. Assert: _reader_@[[ownerReadableStream]] is not *undefined*.
   1. Assert: _reader_@[[ownerReadableStream]]@[[state]] is "readable".
+  1. Set _reader_@[[ownerReadableStream]]@[[disturbed]] to *true*.
   1. If _reader_@[[ownerReadableStream]]@[[queue]] is not empty,
     1. Let _chunk_ be DequeueValue(_reader_@[[ownerReadableStream]]@[[queue]]).
     1. If _reader_@[[ownerReadableStream]]@[[closeRequested]] is *true* and _reader_@[[ownerReadableStream]]@[[queue]] is now empty, perform FinishClosingReadableStream(_reader_@[[ownerReadableStream]]).

reference-implementation/lib/readable-stream.js

@@ -16,6 +16,8 @@ export default class ReadableStream {
     this._reader = undefined;
     this._storedError = undefined;
 
+    this._disturbed = false;
+
     const normalizedStrategy = ValidateAndNormalizeQueuingStrategy(size, highWaterMark);
     this._strategySize = normalizedStrategy.size;
     this._strategyHWM = normalizedStrategy.highWaterMark;
@@ -336,6 +338,8 @@ function CancelReadableStream(stream, reason) {
     return Promise.reject(stream._storedError);
   }
 
+  stream._disturbed = true;
+
   stream._queue = [];
   FinishClosingReadableStream(stream);
 
@@ -439,6 +443,12 @@ function IsReadableStream(x) {
   return true;
 }
 
+export function IsReadableStreamDisturbed(stream) {
+  assert(IsReadableStream(stream) === true, 'IsReadableStreamDisturbed should only be used on known readable streams');
+
+  return stream._disturbed;
+}
+
 function IsReadableStreamLocked(stream) {
   assert(IsReadableStream(stream) === true, 'IsReadableStreamLocked should only be used on known readable streams');
 
@@ -485,6 +495,8 @@ function ReadFromReadableStreamReader(reader) {
   assert(reader._ownerReadableStream !== undefined);
   assert(reader._ownerReadableStream._state === 'readable');
 
+  reader._ownerReadableStream._disturbed = true;
+
   if (reader._ownerReadableStream._queue.length > 0) {
     const chunk = DequeueValue(reader._ownerReadableStream._queue);
 

reference-implementation/run-tests.js

@@ -2,12 +2,14 @@ const glob = require('glob');
 const path = require('path');
 
 import ReadableStream from './lib/readable-stream';
+import { IsReadableStreamDisturbed } from './lib/readable-stream';
 import WritableStream from './lib/writable-stream';
 import ByteLengthQueuingStrategy from './lib/byte-length-queuing-strategy';
 import CountQueuingStrategy from './lib/count-queuing-strategy';
 import TransformStream from './lib/transform-stream';
 
 global.ReadableStream = ReadableStream;
+global.IsReadableStreamDisturbed = IsReadableStreamDisturbed;
 global.WritableStream = WritableStream;
 global.ByteLengthQueuingStrategy = ByteLengthQueuingStrategy;
 global.CountQueuingStrategy = CountQueuingStrategy;

reference-implementation/test/readable-stream.js

@@ -771,3 +771,33 @@ test('ReadableStream integration test: adapting an async pull source', t => {
     t.end();
   });
 });
+
+test('IsReadableStreamDisturbed returns true for a stream on which read() has been called', t => {
+  const rs = new ReadableStream();
+
+  t.equal(IsReadableStreamDisturbed(rs), false, 'rs should not be disturbed on construction');
+
+  const reader = rs.getReader();
+  t.equal(IsReadableStreamDisturbed(rs), false,
+               'getReader() call has no effect on whether a stream is disturbed or not');
+
+  reader.read();
+  t.equal(IsReadableStreamDisturbed(rs), true, 'rs should be disturbed after read() call');
+
+  t.end();
+});
+
+test('IsReadableStreamDisturbed returns true for a stream on which cancel() has been called', t => {
+  const rs = new ReadableStream();
+
+  t.equal(IsReadableStreamDisturbed(rs), false, 'rs should not be disturbed on construction');
+
+  const reader = rs.getReader();
+  t.equal(IsReadableStreamDisturbed(rs), false,
+               'getReader() call has no effect on whether a stream is disturbed or not');
+
+  reader.cancel();
+  t.equal(IsReadableStreamDisturbed(rs), true, 'rs should be disturbed after cancel() call');
+
+  t.end();
+});