Byte stream update, including reference implementation #361
- Loading branch information
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -66,6 +66,8 @@ class ReadableByteStreamController { | |
|
|
||
| this._underlyingByteSource = underlyingByteSource; | ||
|
|
||
| this._insideUnderlyingByteSource = false; | ||
|
There was a problem hiding this comment. In ReadableStream these are called, respectively, pullAgain and pulling. Those names seems shorter and accurate enough to maybe use here. |
||
|
|
||
| this._pendingPulls = []; | ||
|
|
||
| this._queue = []; | ||
|
|
@@ -113,9 +115,10 @@ class ReadableByteStreamController { | |
|
|
||
| if (this._pendingPulls.length > 0 && this._pendingPulls[0].bytesFilled > 0) { | ||
| DestroyReadableByteStreamController(this); | ||
|
There was a problem hiding this comment. In all but one case, DestroyReadableByteStreamController is followed by ErrorReadableByteStream. It might be worth thinking about how to combine them into one function. If not possible, then that's fine. There was a problem hiding this comment. Actually two. Maybe you searched for Destroy. There's one with typo (Destory). Fixed by 186a262. We could have a helper to run both, but maybe leads to increase of texts and a little longer time to trace call path in reviewing? There was a problem hiding this comment. I was just kind of hoping there was a way to refactor so that they always happen together, instead of happening together usually except in 2 cases. But I haven't thought about it too hard so probably there is a good reason for it to be separate :) There was a problem hiding this comment. Ya. For now I'll leave them as is. |
||
| const e = new TypeError('Insufficient bytes to fill elements in the given buffer'); | ||
| ErrorReadableByteStream(stream, e); | ||
|
|
||
| throw e; | ||
| } | ||
|
|
||
| CloseReadableByteStream(stream); | ||
|
|
@@ -239,15 +242,19 @@ class ReadableByteStreamController { | |
|
|
||
| ConsumeQueueForReadableByteStreamByobReaderReadIntoRequest(this, reader); | ||
| } else { | ||
| if (!controller._insideUnderlyingByteSource) { | ||
| controller._insideUnderlyingByteSource = true; | ||
| try { | ||
| controller._underlyingByteSource.pullInto(pullDescriptor.buffer, | ||
| pullDescriptor.byteOffset + pullDescriptor.bytesFilled, | ||
| pullDescriptor.byteLength - pullDescriptor.bytesFilled); | ||
| } catch (e) { | ||
| DestroyReadableByteStreamController(this); | ||
| if (stream._state === 'readable') { | ||
| ErrorReadableByteStream(stream, e); | ||
| } | ||
| } | ||
| controller._insideUnderlyingByteSource = false; | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -405,7 +412,7 @@ class ReadableByteStreamByobReader { | |
| } | ||
|
|
||
| if (this._state === 'closed' && this._ownerReadableByteStream === undefined) { | ||
| return Promise.resolve(CreateIterResultObject(new view.constructor(view.buffer, view.byteOffset, 0), true)); | ||
| } | ||
|
|
||
| const promise = new Promise((resolve, reject) => { | ||
|
|
@@ -746,15 +753,19 @@ function ConsumeQueueForReadableByteStreamByobReaderReadIntoRequest(controller, | |
|
|
||
| // TODO: Detach pullDescriptor.buffer if detachRequired is true. | ||
|
|
||
| if (!controller._insideUnderlyingByteSource) { | ||
| controller._insideUnderlyingByteSource = true; | ||
| try { | ||
| controller._underlyingByteSource.pullInto(pullDescriptor.buffer, | ||
| pullDescriptor.byteOffset + pullDescriptor.bytesFilled, | ||
| pullDescriptor.byteLength - pullDescriptor.bytesFilled); | ||
| } catch (e) { | ||
| DestroyReadableByteStreamController(controller); | ||
| if (stream._state === 'readable') { | ||
| ErrorReadableByteStream(stream, e); | ||
| } | ||
| } | ||
| controller._insideUnderlyingByteSource = false; | ||
| } | ||
|
|
||
| return undefined; | ||
|
|
@@ -800,13 +811,17 @@ function MaybeRespondToReadableByteStreamReaderReadRequest(controller, reader) { | |
| return undefined; | ||
| } | ||
|
|
||
| if (!controller._insideUnderlyingByteSource) { | ||
| controller._insideUnderlyingByteSource = true; | ||
| try { | ||
| controller._underlyingByteSource.pull(); | ||
| } catch (e) { | ||
| DestoryReadableByteStreamController(controller); | ||
| if (stream._state === 'readable') { | ||
| ErrorReadableByteStream(stream, e); | ||
| } | ||
| } | ||
| controller._insideUnderlyingByteSource = false; | ||
| } | ||
|
|
||
| return undefined; | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm, all these being on the controller seems a bit weird... we should probably be consistent between RBS and RS.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is attempt to satisfy #312. I've put all the helper stuff (buffering, fragmentation, coalescing) into
ReadableByteStreamControllerexperimentally.ReadableByteStreamandReadableByteStream.*Readerdoesn't have any variable for these work.