OLD | NEW |
---|---|
1 // Copyright 2016 The Chromium Authors. All rights reserved. | 1 // Copyright 2016 The Chromium Authors. All rights reserved. |
2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
4 | 4 |
5 // Implementation of WritableStream for Blink. See | 5 // Implementation of WritableStream for Blink. See |
6 // https://streams.spec.whatwg.org/#ws. The implementation closely follows the | 6 // https://streams.spec.whatwg.org/#ws. The implementation closely follows the |
7 // standard, except where required for performance or integration with Blink. In | 7 // standard, except where required for performance or integration with Blink. In |
8 // particular, classes, methods and abstract operations are implemented in the | 8 // particular, classes, methods and abstract operations are implemented in the |
9 // same order as in the standard, to simplify side-by-side reading. | 9 // same order as in the standard, to simplify side-by-side reading. |
10 | 10 |
11 (function(global, binding, v8) { | 11 (function(global, binding, v8) { |
12 'use strict'; | 12 'use strict'; |
13 | 13 |
14 // Private symbols. These correspond to the internal slots in the standard. | 14 // Private symbols. These correspond to the internal slots in the standard. |
15 // "[[X]]" in the standard is spelt _X in this implementation. | 15 // "[[X]]" in the standard is spelt _X in this implementation. |
16 const _pendingWriteRequest = v8.createPrivateSymbol('[[pendingWriteRequest]]') ; | 16 |
17 const _pendingCloseRequest = v8.createPrivateSymbol('[[pendingCloseRequest]]') ; | 17 // TODO(ricea): Optimise [[closeRequest]] and [[inFlightCloseRequest]] into a |
18 const _pendingAbortRequest = v8.createPrivateSymbol('[[pendingAbortRequest]]') ; | 18 // single slot + a flag to say which one is set at the moment. |
19 const _state = v8.createPrivateSymbol('[[state]]'); | 19 const _closeRequest = v8.createPrivateSymbol('[[closeRequest]]'); |
20 const _inFlightWriteRequest = v8.createPrivateSymbol('[[inFlightWriteRequest]] '); | |
21 const _inFlightCloseRequest = v8.createPrivateSymbol('[[inFlightCloseRequest]] '); | |
22 const _pendingAbortRequest = | |
23 v8.createPrivateSymbol('[[pendingAbortRequest]]'); | |
24 // Flags and state are combined into a single integer field for efficiency. | |
25 const _stateAndFlags = v8.createPrivateSymbol('[[state]] and flags'); | |
20 const _storedError = v8.createPrivateSymbol('[[storedError]]'); | 26 const _storedError = v8.createPrivateSymbol('[[storedError]]'); |
21 const _writer = v8.createPrivateSymbol('[[writer]]'); | |
22 const _writableStreamController = | 27 const _writableStreamController = |
23 v8.createPrivateSymbol('[[writableStreamController]]'); | 28 v8.createPrivateSymbol('[[writableStreamController]]'); |
29 const _writer = v8.createPrivateSymbol('[[writer]]'); | |
24 const _writeRequests = v8.createPrivateSymbol('[[writeRequests]]'); | 30 const _writeRequests = v8.createPrivateSymbol('[[writeRequests]]'); |
25 const _closedPromise = v8.createPrivateSymbol('[[closedPromise]]'); | 31 const _closedPromise = v8.createPrivateSymbol('[[closedPromise]]'); |
26 const _ownerWritableStream = | 32 const _ownerWritableStream = |
27 v8.createPrivateSymbol('[[ownerWritableStream]]'); | 33 v8.createPrivateSymbol('[[ownerWritableStream]]'); |
28 const _readyPromise = v8.createPrivateSymbol('[[readyPromise]]'); | 34 const _readyPromise = v8.createPrivateSymbol('[[readyPromise]]'); |
29 const _controlledWritableStream = | 35 const _controlledWritableStream = |
30 v8.createPrivateSymbol('[[controlledWritableStream]]'); | 36 v8.createPrivateSymbol('[[controlledWritableStream]]'); |
31 const _queue = v8.createPrivateSymbol('[[queue]]'); | 37 const _queue = v8.createPrivateSymbol('[[queue]]'); |
32 const _queueSize = v8.createPrivateSymbol('[[queueSize]]'); | 38 const _queueTotalSize = v8.createPrivateSymbol('[[queueTotalSize]]'); |
39 const _started = v8.createPrivateSymbol('[[started]]'); | |
33 const _strategyHWM = v8.createPrivateSymbol('[[strategyHWM]]'); | 40 const _strategyHWM = v8.createPrivateSymbol('[[strategyHWM]]'); |
34 const _strategySize = v8.createPrivateSymbol('[[strategySize]]'); | 41 const _strategySize = v8.createPrivateSymbol('[[strategySize]]'); |
35 const _underlyingSink = v8.createPrivateSymbol('[[underlyingSink]]'); | 42 const _underlyingSink = v8.createPrivateSymbol('[[underlyingSink]]'); |
36 | 43 |
37 // _defaultControllerFlags combines WritableStreamDefaultController's internal | 44 // Numeric encodings of states |
38 // slots [[started]], [[writing]], and [[inClose]] into a single bitmask for | 45 const WRITABLE = 0; |
39 // efficiency. | 46 const CLOSED = 1; |
40 const _defaultControllerFlags = | 47 const ERRORED = 2; |
41 v8.createPrivateSymbol('[[defaultControllerFlags]]'); | |
42 const FLAG_STARTED = 0b1; | |
43 const FLAG_WRITING = 0b10; | |
44 const FLAG_INCLOSE = 0b100; | |
45 | 48 |
46 // For efficiency, WritableStream [[state]] contains numeric values. | 49 // Mask to extract or assign states to _stateAndFlags |
47 const WRITABLE = 0; | 50 const STATE_MASK = 0xF; |
48 const CLOSING = 1; | 51 |
49 const CLOSED = 2; | 52 const BACKPRESSURE_FLAG = 0x10; |
50 const ERRORED = 3; | |
51 | 53 |
52 // Javascript functions. It is important to use these copies, as the ones on | 54 // Javascript functions. It is important to use these copies, as the ones on |
53 // the global object may have been overwritten. See "V8 Extras Design Doc", | 55 // the global object may have been overwritten. See "V8 Extras Design Doc", |
54 // section "Security Considerations". | 56 // section "Security Considerations". |
55 // https://docs.google.com/document/d/1AT5-T0aHGp7Lt29vPWFr2-qG8r3l9CByyvKwEuA 8Ec0/edit#heading=h.9yixony1a18r | 57 // https://docs.google.com/document/d/1AT5-T0aHGp7Lt29vPWFr2-qG8r3l9CByyvKwEuA 8Ec0/edit#heading=h.9yixony1a18r |
56 const undefined = global.undefined; | 58 const undefined = global.undefined; |
57 | 59 |
58 const defineProperty = global.Object.defineProperty; | 60 const defineProperty = global.Object.defineProperty; |
59 const hasOwnProperty = v8.uncurryThis(global.Object.hasOwnProperty); | 61 const hasOwnProperty = v8.uncurryThis(global.Object.hasOwnProperty); |
60 | 62 |
61 const Function_apply = v8.uncurryThis(global.Function.prototype.apply); | 63 const Function_apply = v8.uncurryThis(global.Function.prototype.apply); |
64 const Function_call = v8.uncurryThis(global.Function.prototype.call); | |
62 | 65 |
63 const TypeError = global.TypeError; | 66 const TypeError = global.TypeError; |
64 const RangeError = global.RangeError; | 67 const RangeError = global.RangeError; |
65 | 68 |
66 const Boolean = global.Boolean; | 69 const Boolean = global.Boolean; |
67 const Number = global.Number; | 70 const Number = global.Number; |
68 const Number_isNaN = Number.isNaN; | 71 const Number_isNaN = Number.isNaN; |
69 const Number_isFinite = Number.isFinite; | 72 const Number_isFinite = Number.isFinite; |
70 | 73 |
71 const Promise = global.Promise; | 74 const Promise = global.Promise; |
72 const thenPromise = v8.uncurryThis(Promise.prototype.then); | 75 const thenPromise = v8.uncurryThis(Promise.prototype.then); |
73 const Promise_resolve = v8.simpleBind(Promise.resolve, Promise); | 76 const Promise_resolve = v8.simpleBind(Promise.resolve, Promise); |
74 const Promise_reject = v8.simpleBind(Promise.reject, Promise); | 77 const Promise_reject = v8.simpleBind(Promise.reject, Promise); |
75 | 78 |
76 // User-visible strings. | 79 // User-visible strings. |
77 const streamErrors = binding.streamErrors; | 80 const streamErrors = binding.streamErrors; |
78 const errAbortLockedStream = 'Cannot abort a writable stream that is locked to a writer'; | 81 const errAbortLockedStream = 'Cannot abort a writable stream that is locked to a writer'; |
79 const errStreamAborted = 'The stream has been aborted'; | 82 const errStreamAborted = 'The stream has been aborted'; |
83 const errStreamAborting = 'The stream is in the process of being aborted'; | |
80 const errWriterLockReleasedPrefix = 'This writable stream writer has been rele ased and cannot be '; | 84 const errWriterLockReleasedPrefix = 'This writable stream writer has been rele ased and cannot be '; |
81 const errCloseCloseRequestedStream = | 85 const errCloseCloseRequestedStream = |
82 'Cannot close a writable stream that has already been requested to be clos ed'; | 86 'Cannot close a writable stream that has already been requested to be clos ed'; |
83 const errWriteCloseRequestedStream = | 87 const errWriteCloseRequestedStream = |
84 'Cannot write to a writable stream that is due to be closed'; | 88 'Cannot write to a writable stream that is due to be closed'; |
85 const templateErrorCannotActionOnStateStream = | 89 const templateErrorCannotActionOnStateStream = |
86 (action, state) => `Cannot ${action} a ${state} writable stream`; | 90 (action, state) => `Cannot ${action} a ${state} writable stream`; |
87 const errReleasedWriterClosedPromise = | 91 const errReleasedWriterClosedPromise = |
88 'This writable stream writer has been released and cannot be used to monit or the stream\'s state'; | 92 'This writable stream writer has been released and cannot be used to monit or the stream\'s state'; |
89 const templateErrorIsNotAFunction = f => `${f} is not a function`; | 93 const templateErrorIsNotAFunction = f => `${f} is not a function`; |
(...skipping 10 matching lines...) Expand all Loading... | |
100 } | 104 } |
101 | 105 |
102 const stateNames = {[CLOSED]: 'closed', [ERRORED]: 'errored'}; | 106 const stateNames = {[CLOSED]: 'closed', [ERRORED]: 'errored'}; |
103 function createCannotActionOnStateStreamError(action, state) { | 107 function createCannotActionOnStateStreamError(action, state) { |
104 TEMP_ASSERT(stateNames[state] !== undefined, | 108 TEMP_ASSERT(stateNames[state] !== undefined, |
105 `name for state ${state} exists in stateNames`); | 109 `name for state ${state} exists in stateNames`); |
106 return new TypeError( | 110 return new TypeError( |
107 templateErrorCannotActionOnStateStream(action, stateNames[state])); | 111 templateErrorCannotActionOnStateStream(action, stateNames[state])); |
108 } | 112 } |
109 | 113 |
110 function setDefaultControllerFlag(controller, flag, value) { | |
111 let flags = controller[_defaultControllerFlags]; | |
112 if (value) { | |
113 flags = flags | flag; | |
114 } else { | |
115 flags = flags & ~flag; | |
116 } | |
117 controller[_defaultControllerFlags] = flags; | |
118 } | |
119 | |
120 function getDefaultControllerStartedFlag(controller) { | |
121 return Boolean(controller[_defaultControllerFlags] & FLAG_STARTED); | |
122 } | |
123 | |
124 function setDefaultControllerStartedFlag(controller, value) { | |
125 setDefaultControllerFlag(controller, FLAG_STARTED, value); | |
126 } | |
127 | |
128 function getDefaultControllerWritingFlag(controller) { | |
129 return Boolean(controller[_defaultControllerFlags] & FLAG_WRITING); | |
130 } | |
131 | |
132 function setDefaultControllerWritingFlag(controller, value) { | |
133 setDefaultControllerFlag(controller, FLAG_WRITING, value); | |
134 } | |
135 | |
136 function getDefaultControllerInCloseFlag(controller) { | |
137 return Boolean(controller[_defaultControllerFlags] & FLAG_INCLOSE); | |
138 } | |
139 | |
140 function setDefaultControllerInCloseFlag(controller, value) { | |
141 setDefaultControllerFlag(controller, FLAG_INCLOSE, value); | |
142 } | |
143 | |
144 function rejectPromises(queue, e) { | 114 function rejectPromises(queue, e) { |
145 queue.forEach(promise => v8.rejectPromise(promise, e)); | 115 queue.forEach(promise => v8.rejectPromise(promise, e)); |
146 } | 116 } |
147 | 117 |
148 // https://tc39.github.io/ecma262/#sec-ispropertykey | 118 // https://tc39.github.io/ecma262/#sec-ispropertykey |
149 // TODO(ricea): Remove this when the asserts using it are removed. | 119 // TODO(ricea): Remove this when the asserts using it are removed. |
150 function IsPropertyKey(argument) { | 120 function IsPropertyKey(argument) { |
151 return typeof argument === 'string' || typeof argument === 'symbol'; | 121 return typeof argument === 'string' || typeof argument === 'symbol'; |
152 } | 122 } |
153 | 123 |
154 // TODO(ricea): Remove all asserts once the implementation has stabilised. | 124 // TODO(ricea): Remove all asserts once the implementation has stabilised. |
155 function TEMP_ASSERT(predicate, message) { | 125 function TEMP_ASSERT(predicate, message) { |
156 if (predicate) { | 126 if (predicate) { |
157 return; | 127 return; |
158 } | 128 } |
159 v8.log(`Assertion failed: ${message}\n`); | 129 v8.log(`Assertion failed: ${message}\n`); |
160 v8.logStackTrace(); | 130 v8.logStackTrace(); |
161 class WritableStreamInternalError extends Error { | 131 class WritableStreamInternalError extends Error { |
162 constructor(message) { | 132 constructor(message) { |
163 super(message); | 133 super(message); |
164 } | 134 } |
165 } | 135 } |
166 throw new WritableStreamInternalError(message); | 136 throw new WritableStreamInternalError(message); |
167 } | 137 } |
168 | 138 |
169 class WritableStream { | 139 class WritableStream { |
170 constructor(underlyingSink = {}, { size, highWaterMark = 1 } = {}) { | 140 constructor(underlyingSink = {}, { size, highWaterMark = 1 } = {}) { |
171 this[_state] = WRITABLE; | 141 this[_stateAndFlags] = WRITABLE; |
172 this[_storedError] = undefined; | 142 this[_storedError] = undefined; |
173 this[_writer] = undefined; | 143 this[_writer] = undefined; |
174 this[_writableStreamController] = undefined; | 144 this[_writableStreamController] = undefined; |
175 this[_pendingWriteRequest] = undefined; | 145 this[_inFlightWriteRequest] = undefined; |
176 this[_pendingCloseRequest] = undefined; | 146 this[_closeRequest] = undefined; |
147 this[_inFlightCloseRequest] = undefined; | |
177 this[_pendingAbortRequest] = undefined; | 148 this[_pendingAbortRequest] = undefined; |
178 this[_writeRequests] = new binding.SimpleQueue(); | 149 this[_writeRequests] = new binding.SimpleQueue(); |
179 const type = underlyingSink.type; | 150 const type = underlyingSink.type; |
180 if (type !== undefined) { | 151 if (type !== undefined) { |
181 throw new RangeError(streamErrors.invalidType); | 152 throw new RangeError(streamErrors.invalidType); |
182 } | 153 } |
183 this[_writableStreamController] = | 154 this[_writableStreamController] = |
184 new WritableStreamDefaultController(this, underlyingSink, size, | 155 new WritableStreamDefaultController(this, underlyingSink, size, |
185 highWaterMark); | 156 highWaterMark); |
157 WritableStreamDefaultControllerStartSteps(this[_writableStreamController]) ; | |
186 } | 158 } |
187 | 159 |
188 get locked() { | 160 get locked() { |
189 if (!IsWritableStream(this)) { | 161 if (!IsWritableStream(this)) { |
190 throw new TypeError(streamErrors.illegalInvocation); | 162 throw new TypeError(streamErrors.illegalInvocation); |
191 } | 163 } |
192 return IsWritableStreamLocked(this); | 164 return IsWritableStreamLocked(this); |
193 } | 165 } |
194 | 166 |
195 abort(reason) { | 167 abort(reason) { |
(...skipping 24 matching lines...) Expand all Loading... | |
220 return hasOwnProperty(x, _writableStreamController); | 192 return hasOwnProperty(x, _writableStreamController); |
221 } | 193 } |
222 | 194 |
223 function IsWritableStreamLocked(stream) { | 195 function IsWritableStreamLocked(stream) { |
224 TEMP_ASSERT(IsWritableStream(stream), | 196 TEMP_ASSERT(IsWritableStream(stream), |
225 '! IsWritableStream(stream) is true.'); | 197 '! IsWritableStream(stream) is true.'); |
226 return stream[_writer] !== undefined; | 198 return stream[_writer] !== undefined; |
227 } | 199 } |
228 | 200 |
229 function WritableStreamAbort(stream, reason) { | 201 function WritableStreamAbort(stream, reason) { |
230 const state = stream[_state]; | 202 const state = stream[_stateAndFlags] & STATE_MASK; |
231 if (state === CLOSED) { | 203 if (state === CLOSED) { |
232 return Promise_resolve(undefined); | 204 return Promise_resolve(undefined); |
233 } | 205 } |
234 if (state === ERRORED) { | 206 if (state === ERRORED) { |
235 return Promise_reject(stream[_storedError]); | 207 return Promise_reject(stream[_storedError]); |
236 } | 208 } |
237 TEMP_ASSERT(state === WRITABLE || state === CLOSING, | 209 TEMP_ASSERT(state === WRITABLE, |
238 'state is "writable" or "closing".'); | 210 'state is "writable".'); |
239 const error = new TypeError(errStreamAborted); | 211 const error = new TypeError(errStreamAborting); |
240 WritableStreamError(stream, error); | 212 if (stream[_pendingAbortRequest] !== undefined) { |
213 return Promise_reject(error); | |
214 } | |
241 | 215 |
242 const controller = stream[_writableStreamController]; | 216 const controller = stream[_writableStreamController]; |
243 TEMP_ASSERT(controller !== undefined, | 217 TEMP_ASSERT(controller !== undefined, |
244 'controller is not undefined'); | 218 'controller is not undefined'); |
219 if (!WritableStreamHasOperationMarkedInFlight(stream) && | |
220 controller[_started]) { | |
221 WritableStreamFinishAbort(stream); | |
222 return WritableStreamDefaultControllerAbortSteps(controller, reason); | |
223 } | |
224 const writer = stream[_writer]; | |
225 if (writer !== undefined) { | |
226 WritableStreamDefaultWriterEnsureReadyPromiseRejected(writer, error); | |
227 } | |
228 const promise = v8.createPromise(); | |
229 stream[_pendingAbortRequest] = {promise, reason}; | |
230 return promise; | |
231 } | |
245 | 232 |
246 const isWriting = getDefaultControllerWritingFlag(controller); | 233 function WritableStreamError(stream, error) { |
247 if (isWriting || getDefaultControllerInCloseFlag(controller)) { | 234 stream[_stateAndFlags] = (stream[_stateAndFlags] & ~STATE_MASK) | ERRORED; |
248 const promise = v8.createPromise(); | 235 stream[_storedError] = error; |
249 stream[_pendingAbortRequest] = promise; | 236 WritableStreamDefaultControllerErrorSteps(stream[_writableStreamController]) ; |
237 if (stream[_pendingAbortRequest] === undefined) { | |
238 const writer = stream[_writer]; | |
239 if (writer !== undefined) { | |
240 WritableStreamDefaultWriterEnsureReadyPromiseRejected(writer, error); | |
241 } | |
242 } | |
243 if (!WritableStreamHasOperationMarkedInFlight(stream)) { | |
244 WritableStreamRejectPromisesInReactionToError(stream); | |
245 } | |
246 } | |
250 | 247 |
251 if (isWriting) { | 248 function WritableStreamFinishAbort(stream) { |
252 return thenPromise(promise, () => { | 249 const error = new TypeError(errStreamAborted); |
253 return WritableStreamDefaultControllerAbort(controller, reason); | 250 WritableStreamError(stream, error); |
254 }); | |
255 } | |
256 return promise; | |
257 } | |
258 | |
259 return WritableStreamDefaultControllerAbort(controller, reason); | |
260 } | 251 } |
261 | 252 |
262 // Writable Stream Abstract Operations Used by Controllers | 253 // Writable Stream Abstract Operations Used by Controllers |
263 | 254 |
264 function WritableStreamAddWriteRequest(stream) { | 255 function WritableStreamAddWriteRequest(stream) { |
265 TEMP_ASSERT(IsWritableStreamLocked(stream), | 256 TEMP_ASSERT(IsWritableStreamLocked(stream), |
266 '! IsWritableStreamLocked(writer) is true.'); | 257 '! IsWritableStreamLocked(writer) is true.'); |
267 TEMP_ASSERT(stream[_state] === WRITABLE, | 258 TEMP_ASSERT((stream[_stateAndFlags] & STATE_MASK) === WRITABLE, |
268 'stream.[[state]] is "writable".'); | 259 'stream.[[state]] is "writable".'); |
269 const promise = v8.createPromise(); | 260 const promise = v8.createPromise(); |
270 stream[_writeRequests].push(promise); | 261 stream[_writeRequests].push(promise); |
271 return promise; | 262 return promise; |
272 } | 263 } |
273 | 264 |
274 function WritableStreamError(stream, e) { | 265 function WritableStreamFinishInFlightWrite(stream) { |
275 const oldState = stream[_state]; | 266 TEMP_ASSERT(stream[_inFlightWriteRequest] !== undefined, |
276 TEMP_ASSERT(oldState === WRITABLE || oldState === CLOSING, | 267 '_stream_.[[inFlightWriteRequest]] is not *undefined*.'); |
277 'oldState is "writable" or "closing".'); | 268 v8.resolvePromise(stream[_inFlightWriteRequest], undefined); |
269 stream[_inFlightWriteRequest] = undefined; | |
270 const state = stream[_stateAndFlags] & STATE_MASK; | |
271 if (state === ERRORED) { | |
272 WritableStreamFinishInFlightWriteInErroredState(stream); | |
273 return; | |
274 } | |
275 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); | |
276 WritableStreamHandleAbortRequestIfPending(stream); | |
277 } | |
278 | 278 |
279 stream[_state] = ERRORED; | 279 function WritableStreamFinishInFlightWriteInErroredState(stream) { |
280 stream[_storedError] = e; | 280 WritableStreamRejectAbortRequestIfPending(stream); |
281 WritableStreamRejectPromisesInReactionToError(stream); | |
282 } | |
281 | 283 |
282 const controller = stream[_writableStreamController]; | 284 function WritableStreamFinishInFlightWriteWithError(stream, error) { |
283 if (controller === undefined || | 285 TEMP_ASSERT(stream[_inFlightWriteRequest] !== undefined, |
284 (!getDefaultControllerWritingFlag(controller) && | 286 '_stream_.[[inFlightWriteRequest]] is not *undefined*.'); |
285 !getDefaultControllerInCloseFlag(controller))) { | 287 v8.rejectPromise(stream[_inFlightWriteRequest], error); |
286 WritableStreamRejectPromisesInReactionToError(stream); | 288 stream[_inFlightWriteRequest] = undefined; |
289 const state = stream[_stateAndFlags] & STATE_MASK; | |
290 if (state === ERRORED) { | |
291 WritableStreamFinishInFlightWriteInErroredState(stream); | |
292 return; | |
287 } | 293 } |
294 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); | |
295 WritableStreamError(stream, error); | |
296 WritableStreamRejectAbortRequestIfPending(stream); | |
297 } | |
288 | 298 |
299 function WritableStreamFinishInFlightClose(stream) { | |
300 TEMP_ASSERT(stream[_inFlightCloseRequest] !== undefined, | |
301 '_stream_.[[inFlightCloseRequest]] is not *undefined*.'); | |
302 v8.resolvePromise(stream[_inFlightCloseRequest], undefined); | |
303 stream[_inFlightCloseRequest] = undefined; | |
304 const state = stream[_stateAndFlags] & STATE_MASK; | |
305 if (state === ERRORED) { | |
306 WritableStreamFinishInFlightCloseInErroredState(stream); | |
307 return; | |
308 } | |
309 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); | |
310 stream[_stateAndFlags] = (stream[_stateAndFlags] & ~STATE_MASK) | CLOSED; | |
289 const writer = stream[_writer]; | 311 const writer = stream[_writer]; |
290 if (writer !== undefined) { | 312 if (writer !== undefined) { |
291 if (oldState === WRITABLE && | 313 v8.resolvePromise(writer[_closedPromise], undefined); |
292 WritableStreamDefaultControllerGetBackpressure(controller) === true) { | 314 } |
293 v8.rejectPromise(writer[_readyPromise], e); | 315 if (stream[_pendingAbortRequest] !== undefined) { |
294 } else { | 316 v8.resolvePromise(stream[_pendingAbortRequest].promise, undefined); |
295 writer[_readyPromise] = Promise_reject(e); | 317 stream[_pendingAbortRequest] = undefined; |
296 } | |
297 v8.markPromiseAsHandled(writer[_readyPromise]); | |
298 } | 318 } |
299 } | 319 } |
300 | 320 |
301 function WritableStreamFinishClose(stream) { | 321 function WritableStreamFinishInFlightCloseInErroredState(stream) { |
302 const state = stream[_state]; | 322 WritableStreamRejectAbortRequestIfPending(stream); |
303 TEMP_ASSERT(state === CLOSING || state === ERRORED, | 323 WritableStreamRejectClosedPromiseInReactionToError(stream); |
304 'state is "closing" or "errored"'); | 324 } |
305 | 325 |
326 function WritableStreamFinishInFlightCloseWithError(stream, error) { | |
327 TEMP_ASSERT(stream[_inFlightCloseRequest] !== undefined, | |
328 '_stream_.[[inFlightCloseRequest]] is not *undefined*.'); | |
329 v8.rejectPromise(stream[_inFlightCloseRequest], error); | |
330 stream[_inFlightCloseRequest] = undefined; | |
331 const state = stream[_stateAndFlags] & STATE_MASK; | |
332 if (state === ERRORED) { | |
333 WritableStreamFinishInFlightCloseInErroredState(stream); | |
334 return; | |
335 } | |
336 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); | |
337 WritableStreamError(stream, error); | |
338 WritableStreamRejectAbortRequestIfPending(stream); | |
339 } | |
340 | |
341 function WritableStreamCloseQueuedOrInFlight(stream) { | |
342 return stream[_closeRequest] !== undefined || | |
343 stream[_inFlightCloseRequest] !== undefined; | |
344 } | |
345 | |
346 function WritableStreamHandleAbortRequestIfPending(stream) { | |
347 if (stream[_pendingAbortRequest] === undefined) { | |
348 return; | |
349 } | |
350 WritableStreamFinishAbort(stream); | |
351 const abortRequest = stream[_pendingAbortRequest]; | |
352 stream[_pendingAbortRequest] = undefined; | |
353 const promise = | |
354 WritableStreamDefaultControllerAbortSteps(stream[_writableStreamControll er], | |
355 abortRequest.reason); | |
356 thenPromise(promise, | |
357 result => v8.resolvePromise(abortRequest.promise, result), | |
358 reason => v8.rejectPromise(abortRequest.promise, reason)); | |
359 } | |
360 | |
361 function WritableStreamHasOperationMarkedInFlight(stream) { | |
362 return stream[_inFlightWriteRequest] !== undefined || | |
363 stream[_inFlightCloseRequest] !== undefined; | |
364 } | |
365 | |
366 function WritableStreamMarkCloseRequestInFlight(stream) { | |
367 TEMP_ASSERT(stream[_inFlightCloseRequest] === undefined, | |
368 '_stream_.[[inFlightCloseRequest]] is *undefined*.'); | |
369 TEMP_ASSERT(stream[_closeRequest] !== undefined, | |
370 '_stream_.[[closeRequest]] is not *undefined*.'); | |
371 stream[_inFlightCloseRequest] = stream[_closeRequest]; | |
372 stream[_closeRequest] = undefined; | |
373 } | |
374 | |
375 function WritableStreamMarkFirstWriteRequestInFlight(stream) { | |
376 TEMP_ASSERT(stream[_inFlightWriteRequest] === undefined, | |
377 '_stream_.[[inFlightWriteRequest]] is *undefined*.'); | |
378 TEMP_ASSERT(stream[_writeRequests].length !== 0, | |
379 '_stream_.[[writeRequests]] is not empty.'); | |
380 const writeRequest = stream[_writeRequests].shift(); | |
381 stream[_inFlightWriteRequest] = writeRequest; | |
382 } | |
383 | |
384 function WritableStreamRejectClosedPromiseInReactionToError(stream) { | |
306 const writer = stream[_writer]; | 385 const writer = stream[_writer]; |
307 if (state === CLOSING) { | 386 if (writer !== undefined) { |
308 if (writer !== undefined) { | |
309 v8.resolvePromise(writer[_closedPromise], undefined); | |
310 } | |
311 stream[_state] = CLOSED; | |
312 } else if (writer !== undefined) { | |
313 TEMP_ASSERT(state === ERRORED, 'state is "errored"'); | |
314 v8.rejectPromise(writer[_closedPromise], stream[_storedError]); | 387 v8.rejectPromise(writer[_closedPromise], stream[_storedError]); |
315 v8.markPromiseAsHandled(writer[_closedPromise]); | 388 v8.markPromiseAsHandled(writer[_closedPromise]); |
316 } | 389 } |
390 } | |
317 | 391 |
392 function WritableStreamRejectAbortRequestIfPending(stream) { | |
318 if (stream[_pendingAbortRequest] !== undefined) { | 393 if (stream[_pendingAbortRequest] !== undefined) { |
319 v8.resolvePromise(stream[_pendingAbortRequest], undefined); | 394 v8.rejectPromise(stream[_pendingAbortRequest].promise, |
395 stream[_storedError]); | |
320 stream[_pendingAbortRequest] = undefined; | 396 stream[_pendingAbortRequest] = undefined; |
321 } | 397 } |
322 } | 398 } |
323 | 399 |
324 function WritableStreamRejectPromisesInReactionToError(stream) { | 400 function WritableStreamRejectPromisesInReactionToError(stream) { |
325 TEMP_ASSERT(stream[_state] === ERRORED, 'stream.[[state]] is "errored"'); | |
326 TEMP_ASSERT(stream[_pendingWriteRequest] === undefined, | |
327 'stream.[[pendingWriteRequest]] is undefined'); | |
328 | |
329 const storedError = stream[_storedError]; | 401 const storedError = stream[_storedError]; |
330 rejectPromises(stream[_writeRequests], storedError); | 402 rejectPromises(stream[_writeRequests], storedError); |
331 stream[_writeRequests] = new binding.SimpleQueue(); | 403 stream[_writeRequests] = new binding.SimpleQueue(); |
332 | 404 |
333 if (stream[_pendingCloseRequest] !== undefined) { | 405 if (stream[_closeRequest] !== undefined) { |
334 TEMP_ASSERT( | 406 TEMP_ASSERT(stream[_inFlightCloseRequest] === undefined, |
335 getDefaultControllerInCloseFlag(stream[_writableStreamController]) === | 407 '_stream_.[[inFlightCloseRequest]] is *undefined*.'); |
336 false, 'stream.[[writableStreamController]].[[inClose]] === false'); | 408 v8.rejectPromise(stream[_closeRequest], storedError); |
337 v8.rejectPromise(stream[_pendingCloseRequest], storedError); | 409 stream[_closeRequest] = undefined; |
338 stream[_pendingCloseRequest] = undefined; | |
339 } | 410 } |
340 | 411 |
412 WritableStreamRejectClosedPromiseInReactionToError(stream); | |
413 } | |
414 | |
415 function WritableStreamUpdateBackpressure(stream, backpressure) { | |
416 TEMP_ASSERT((stream[_stateAndFlags] & STATE_MASK) === WRITABLE, | |
417 'stream.[[state]] is "writable".'); | |
418 TEMP_ASSERT(!WritableStreamCloseQueuedOrInFlight(stream), | |
419 'WritableStreamCloseQueuedOrInFlight(_stream_) is *false*.'); | |
341 const writer = stream[_writer]; | 420 const writer = stream[_writer]; |
342 if (writer !== undefined) { | 421 if (writer !== undefined && |
343 v8.rejectPromise(writer[_closedPromise], storedError); | 422 backpressure !== Boolean(stream[_stateAndFlags] & BACKPRESSURE_FLAG)) { |
344 v8.markPromiseAsHandled(writer[_closedPromise]); | 423 if (backpressure) { |
424 writer[_readyPromise] = v8.createPromise(); | |
425 } else { | |
426 TEMP_ASSERT(!backpressure, '_backpressure_ is *false*.'); | |
427 v8.resolvePromise(writer[_readyPromise], undefined); | |
428 } | |
429 } | |
430 if (backpressure) { | |
431 stream[_stateAndFlags] |= BACKPRESSURE_FLAG; | |
432 } else { | |
433 stream[_stateAndFlags] &= ~BACKPRESSURE_FLAG; | |
345 } | 434 } |
346 } | 435 } |
347 | 436 |
348 function WritableStreamUpdateBackpressure(stream, backpressure) { | |
349 TEMP_ASSERT(stream[_state] === WRITABLE, | |
350 'stream.[[state]] is "writable".'); | |
351 const writer = stream[_writer]; | |
352 if (writer === undefined) { | |
353 return; | |
354 } | |
355 if (backpressure) { | |
356 writer[_readyPromise] = v8.createPromise(); | |
357 } else { | |
358 TEMP_ASSERT(backpressure === false, | |
359 'backpressure is false.'); | |
360 v8.resolvePromise(writer[_readyPromise], undefined); | |
361 } | |
362 } | |
363 | |
364 // Functions to expose internals for ReadableStream.pipeTo. These are not | 437 // Functions to expose internals for ReadableStream.pipeTo. These are not |
365 // part of the standard. | 438 // part of the standard. |
366 function isWritableStreamErrored(stream) { | 439 function isWritableStreamErrored(stream) { |
367 TEMP_ASSERT( | 440 TEMP_ASSERT( |
368 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); | 441 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); |
369 return stream[_state] === ERRORED; | 442 return (stream[_stateAndFlags] & STATE_MASK) === ERRORED; |
370 } | 443 } |
371 | 444 |
372 function isWritableStreamClosingOrClosed(stream) { | 445 function isWritableStreamClosingOrClosed(stream) { |
373 TEMP_ASSERT( | 446 TEMP_ASSERT( |
374 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); | 447 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); |
375 return stream[_state] === CLOSING || stream[_state] === CLOSED; | 448 return WritableStreamCloseQueuedOrInFlight(stream) || |
449 (stream[_stateAndFlags] & STATE_MASK) === CLOSED; | |
376 } | 450 } |
377 | 451 |
378 function getWritableStreamStoredError(stream) { | 452 function getWritableStreamStoredError(stream) { |
379 TEMP_ASSERT( | 453 TEMP_ASSERT( |
380 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); | 454 IsWritableStream(stream), '! IsWritableStream(stream) is true.'); |
381 return stream[_storedError]; | 455 return stream[_storedError]; |
382 } | 456 } |
383 | 457 |
384 class WritableStreamDefaultWriter { | 458 class WritableStreamDefaultWriter { |
385 constructor(stream) { | 459 constructor(stream) { |
386 if (!IsWritableStream(stream)) { | 460 if (!IsWritableStream(stream)) { |
387 throw new TypeError(streamErrors.illegalConstructor); | 461 throw new TypeError(streamErrors.illegalConstructor); |
388 } | 462 } |
389 if (IsWritableStreamLocked(stream)) { | 463 if (IsWritableStreamLocked(stream)) { |
390 throw new TypeError(streamErrors.illegalConstructor); | 464 throw new TypeError(streamErrors.illegalConstructor); |
391 } | 465 } |
392 this[_ownerWritableStream] = stream; | 466 this[_ownerWritableStream] = stream; |
393 stream[_writer] = this; | 467 stream[_writer] = this; |
394 const state = stream[_state]; | 468 const state = stream[_stateAndFlags] & STATE_MASK; |
395 if (state === WRITABLE || state === CLOSING) { | 469 if (state === WRITABLE) { |
470 if (stream[_pendingAbortRequest] !== undefined) { | |
471 const error = new TypeError(errStreamAborting); | |
472 this[_readyPromise] = Promise_reject(error); | |
473 v8.markPromiseAsHandled(this[_readyPromise]); | |
474 } else if (!WritableStreamCloseQueuedOrInFlight(stream) && | |
475 stream[_stateAndFlags] & BACKPRESSURE_FLAG) { | |
476 this[_readyPromise] = v8.createPromise(); | |
477 } else { | |
478 this[_readyPromise] = Promise_resolve(undefined); | |
479 } | |
396 this[_closedPromise] = v8.createPromise(); | 480 this[_closedPromise] = v8.createPromise(); |
397 } else if (state === CLOSED) { | 481 } else if (state === CLOSED) { |
482 this[_readyPromise] = Promise_resolve(undefined); | |
398 this[_closedPromise] = Promise_resolve(undefined); | 483 this[_closedPromise] = Promise_resolve(undefined); |
399 } else { | 484 } else { |
400 TEMP_ASSERT(state === ERRORED, | 485 TEMP_ASSERT(state === ERRORED, '_state_ is `"errored"`.'); |
401 'state is "errored".'); | 486 const storedError = stream[_storedError]; |
402 this[_closedPromise] = Promise_reject(stream[_storedError]); | 487 this[_readyPromise] = Promise_reject(storedError); |
488 v8.markPromiseAsHandled(this[_readyPromise]); | |
489 this[_closedPromise] = Promise_reject(storedError); | |
403 v8.markPromiseAsHandled(this[_closedPromise]); | 490 v8.markPromiseAsHandled(this[_closedPromise]); |
404 } | 491 } |
405 if (state === WRITABLE && | |
406 WritableStreamDefaultControllerGetBackpressure( | |
407 stream[_writableStreamController])) { | |
408 this[_readyPromise] = v8.createPromise(); | |
409 } else { | |
410 this[_readyPromise] = Promise_resolve(undefined); | |
411 } | |
412 } | 492 } |
413 | 493 |
414 get closed() { | 494 get closed() { |
415 if (!IsWritableStreamDefaultWriter(this)) { | 495 if (!IsWritableStreamDefaultWriter(this)) { |
416 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); | 496 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); |
417 } | 497 } |
418 return this[_closedPromise]; | 498 return this[_closedPromise]; |
419 } | 499 } |
420 | 500 |
421 get desiredSize() { | 501 get desiredSize() { |
(...skipping 24 matching lines...) Expand all Loading... | |
446 } | 526 } |
447 | 527 |
448 close() { | 528 close() { |
449 if (!IsWritableStreamDefaultWriter(this)) { | 529 if (!IsWritableStreamDefaultWriter(this)) { |
450 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); | 530 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); |
451 } | 531 } |
452 const stream = this[_ownerWritableStream]; | 532 const stream = this[_ownerWritableStream]; |
453 if (stream === undefined) { | 533 if (stream === undefined) { |
454 return Promise_reject(createWriterLockReleasedError(verbClosed)); | 534 return Promise_reject(createWriterLockReleasedError(verbClosed)); |
455 } | 535 } |
456 if (stream[_state] === CLOSING) { | 536 if (WritableStreamCloseQueuedOrInFlight(stream)) { |
457 return Promise_reject(new TypeError(errCloseCloseRequestedStream)); | 537 return Promise_reject(new TypeError(errCloseCloseRequestedStream)); |
458 } | 538 } |
459 return WritableStreamDefaultWriterClose(this); | 539 return WritableStreamDefaultWriterClose(this); |
460 } | 540 } |
461 | 541 |
462 releaseLock() { | 542 releaseLock() { |
463 if (!IsWritableStreamDefaultWriter(this)) { | 543 if (!IsWritableStreamDefaultWriter(this)) { |
464 throw new TypeError(streamErrors.illegalInvocation); | 544 throw new TypeError(streamErrors.illegalInvocation); |
465 } | 545 } |
466 const stream = this[_ownerWritableStream]; | 546 const stream = this[_ownerWritableStream]; |
467 if (stream === undefined) { | 547 if (stream === undefined) { |
468 return; | 548 return; |
469 } | 549 } |
470 TEMP_ASSERT(stream[_writer] !== undefined, | 550 TEMP_ASSERT(stream[_writer] !== undefined, |
471 'stream.[[writer]] is not undefined.'); | 551 'stream.[[writer]] is not undefined.'); |
472 WritableStreamDefaultWriterRelease(this); | 552 WritableStreamDefaultWriterRelease(this); |
473 } | 553 } |
474 | 554 |
475 write(chunk) { | 555 write(chunk) { |
476 if (!IsWritableStreamDefaultWriter(this)) { | 556 if (!IsWritableStreamDefaultWriter(this)) { |
477 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); | 557 return Promise_reject(new TypeError(streamErrors.illegalInvocation)); |
478 } | 558 } |
479 const stream = this[_ownerWritableStream]; | 559 if (this[_ownerWritableStream] === undefined) { |
480 if (stream === undefined) { | |
481 return Promise_reject(createWriterLockReleasedError(verbWrittenTo)); | 560 return Promise_reject(createWriterLockReleasedError(verbWrittenTo)); |
482 } | 561 } |
483 if (stream[_state] === CLOSING) { | |
484 return Promise_reject(new TypeError(errWriteCloseRequestedStream)); | |
485 } | |
486 return WritableStreamDefaultWriterWrite(this, chunk); | 562 return WritableStreamDefaultWriterWrite(this, chunk); |
487 } | 563 } |
488 } | 564 } |
489 | 565 |
490 // Writable Stream Writer Abstract Operations | 566 // Writable Stream Writer Abstract Operations |
491 | 567 |
492 function IsWritableStreamDefaultWriter(x) { | 568 function IsWritableStreamDefaultWriter(x) { |
493 return hasOwnProperty(x, _ownerWritableStream); | 569 return hasOwnProperty(x, _ownerWritableStream); |
494 } | 570 } |
495 | 571 |
496 function WritableStreamDefaultWriterAbort(writer, reason) { | 572 function WritableStreamDefaultWriterAbort(writer, reason) { |
497 const stream = writer[_ownerWritableStream]; | 573 const stream = writer[_ownerWritableStream]; |
498 TEMP_ASSERT(stream !== undefined, | 574 TEMP_ASSERT(stream !== undefined, |
499 'stream is not undefined.'); | 575 'stream is not undefined.'); |
500 return WritableStreamAbort(stream, reason); | 576 return WritableStreamAbort(stream, reason); |
501 } | 577 } |
502 | 578 |
503 function WritableStreamDefaultWriterClose(writer) { | 579 function WritableStreamDefaultWriterClose(writer) { |
504 const stream = writer[_ownerWritableStream]; | 580 const stream = writer[_ownerWritableStream]; |
505 TEMP_ASSERT(stream !== undefined, | 581 TEMP_ASSERT(stream !== undefined, 'stream is not undefined.'); |
506 'stream is not undefined.'); | 582 const state = stream[_stateAndFlags] & STATE_MASK; |
507 const state = stream[_state]; | |
508 if (state === CLOSED || state === ERRORED) { | 583 if (state === CLOSED || state === ERRORED) { |
509 return Promise_reject( | 584 return Promise_reject( |
510 createCannotActionOnStateStreamError('close', state)); | 585 createCannotActionOnStateStreamError('close', state)); |
511 } | 586 } |
512 TEMP_ASSERT(state === WRITABLE, | 587 if (stream[_pendingAbortRequest] !== undefined) { |
513 'state is "writable".'); | 588 return Promise_reject(new TypeError(errStreamAborting)); |
514 stream[_pendingCloseRequest] = v8.createPromise(); | 589 } |
515 if (WritableStreamDefaultControllerGetBackpressure( | 590 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); |
516 stream[_writableStreamController])) { | 591 TEMP_ASSERT(!WritableStreamCloseQueuedOrInFlight(stream), |
592 '! WritableStreamCloseQueuedOrInFlight(_stream_) is *false*.'); | |
593 const promise = v8.createPromise(); | |
594 stream[_closeRequest] = promise; | |
595 if (stream[_stateAndFlags] & BACKPRESSURE_FLAG) { | |
517 v8.resolvePromise(writer[_readyPromise], undefined); | 596 v8.resolvePromise(writer[_readyPromise], undefined); |
518 } | 597 } |
519 stream[_state] = CLOSING; | |
520 WritableStreamDefaultControllerClose(stream[_writableStreamController]); | 598 WritableStreamDefaultControllerClose(stream[_writableStreamController]); |
521 return stream[_pendingCloseRequest]; | 599 return promise; |
522 } | 600 } |
523 | 601 |
524 function WritableStreamDefaultWriterCloseWithErrorPropagation(writer) { | 602 function WritableStreamDefaultWriterCloseWithErrorPropagation(writer) { |
525 const stream = writer[_ownerWritableStream]; | 603 const stream = writer[_ownerWritableStream]; |
526 TEMP_ASSERT(stream !== undefined, 'stream is not undefined.'); | 604 TEMP_ASSERT(stream !== undefined, 'stream is not undefined.'); |
527 const state = stream[_state]; | 605 const state = stream[_stateAndFlags] & STATE_MASK; |
528 if (state === CLOSING || state === CLOSED) { | 606 if (WritableStreamCloseQueuedOrInFlight(stream) || state === CLOSED) { |
529 return Promise_resolve(undefined); | 607 return Promise_resolve(undefined); |
530 } | 608 } |
531 if (state === ERRORED) { | 609 if (state === ERRORED) { |
532 return Promise_reject(stream[_storedError]); | 610 return Promise_reject(stream[_storedError]); |
533 } | 611 } |
534 TEMP_ASSERT(state === WRITABLE, 'state is "writable".'); | 612 TEMP_ASSERT(state === WRITABLE, 'state is "writable".'); |
535 return WritableStreamDefaultWriterClose(writer); | 613 return WritableStreamDefaultWriterClose(writer); |
536 } | 614 } |
537 | 615 |
616 function WritableStreamDefaultWriterEnsureReadyPromiseRejected(writer, | |
617 error) { | |
618 if (v8.promiseState(writer[_readyPromise]) === v8.kPROMISE_PENDING) { | |
619 v8.rejectPromise(writer[_readyPromise], error); | |
620 } else { | |
621 writer[_readyPromise] = Promise_reject(error); | |
622 } | |
623 v8.markPromiseAsHandled(writer[_readyPromise]); | |
624 } | |
625 | |
538 function WritableStreamDefaultWriterGetDesiredSize(writer) { | 626 function WritableStreamDefaultWriterGetDesiredSize(writer) { |
539 const stream = writer[_ownerWritableStream]; | 627 const stream = writer[_ownerWritableStream]; |
540 const state = stream[_state]; | 628 const state = stream[_stateAndFlags] & STATE_MASK; |
541 if (state === ERRORED) { | 629 if (state === ERRORED || stream[_pendingAbortRequest] !== undefined) { |
542 return null; | 630 return null; |
543 } | 631 } |
544 if (state === CLOSED) { | 632 if (state === CLOSED) { |
545 return 0; | 633 return 0; |
546 } | 634 } |
547 return WritableStreamDefaultControllerGetDesiredSize( | 635 return WritableStreamDefaultControllerGetDesiredSize( |
548 stream[_writableStreamController]); | 636 stream[_writableStreamController]); |
549 } | 637 } |
550 | 638 |
551 function WritableStreamDefaultWriterRelease(writer) { | 639 function WritableStreamDefaultWriterRelease(writer) { |
552 const stream = writer[_ownerWritableStream]; | 640 const stream = writer[_ownerWritableStream]; |
553 TEMP_ASSERT(stream !== undefined, | 641 TEMP_ASSERT(stream !== undefined, |
554 'stream is not undefined.'); | 642 'stream is not undefined.'); |
555 TEMP_ASSERT(stream[_writer] === writer, | 643 TEMP_ASSERT(stream[_writer] === writer, |
556 'stream.[[writer]] is writer.'); | 644 'stream.[[writer]] is writer.'); |
557 const releasedError = new TypeError(errReleasedWriterClosedPromise); | 645 const releasedError = new TypeError(errReleasedWriterClosedPromise); |
558 const state = stream[_state]; | 646 const state = stream[_stateAndFlags] & STATE_MASK; |
559 if (state === WRITABLE || state === CLOSING || | 647 WritableStreamDefaultWriterEnsureReadyPromiseRejected(writer, |
560 stream[_pendingAbortRequest] !== undefined) { | 648 releasedError); |
649 if (state === WRITABLE || | |
650 WritableStreamHasOperationMarkedInFlight(stream)) { | |
561 v8.rejectPromise(writer[_closedPromise], releasedError); | 651 v8.rejectPromise(writer[_closedPromise], releasedError); |
562 } else { | 652 } else { |
563 writer[_closedPromise] = Promise_reject(releasedError); | 653 writer[_closedPromise] = Promise_reject(releasedError); |
564 } | 654 } |
565 v8.markPromiseAsHandled(writer[_closedPromise]); | 655 v8.markPromiseAsHandled(writer[_closedPromise]); |
566 | |
567 if (state === WRITABLE && | |
568 WritableStreamDefaultControllerGetBackpressure( | |
569 stream[_writableStreamController])) { | |
570 v8.rejectPromise(writer[_readyPromise], releasedError); | |
571 } else { | |
572 writer[_readyPromise] = Promise_reject(releasedError); | |
573 } | |
574 v8.markPromiseAsHandled(writer[_readyPromise]); | |
575 | |
576 stream[_writer] = undefined; | 656 stream[_writer] = undefined; |
577 writer[_ownerWritableStream] = undefined; | 657 writer[_ownerWritableStream] = undefined; |
578 } | 658 } |
579 | 659 |
580 function WritableStreamDefaultWriterWrite(writer, chunk) { | 660 function WritableStreamDefaultWriterWrite(writer, chunk) { |
581 const stream = writer[_ownerWritableStream]; | 661 const stream = writer[_ownerWritableStream]; |
582 TEMP_ASSERT(stream !== undefined, | 662 TEMP_ASSERT(stream !== undefined, 'stream is not undefined.'); |
583 'stream is not undefined.'); | 663 const controller = stream[_writableStreamController]; |
584 const state = stream[_state]; | 664 const chunkSize = |
585 if (state === CLOSED || state === ERRORED) { | 665 WritableStreamDefaultControllerGetChunkSize(controller, chunk); |
666 if (stream !== writer[_ownerWritableStream]) { | |
667 return Promise_reject(createWriterLockReleasedError(verbWrittenTo)); | |
668 } | |
669 const state = stream[_stateAndFlags] & STATE_MASK; | |
670 if (state === ERRORED) { | |
671 return Promise_reject(stream[_storedError]); | |
672 } | |
673 if (WritableStreamCloseQueuedOrInFlight(stream)) { | |
674 return Promise_reject(new TypeError( | |
675 templateErrorCannotActionOnStateStream('write to', 'closing'))); | |
676 } | |
677 if (state === CLOSED) { | |
586 return Promise_reject( | 678 return Promise_reject( |
587 createCannotActionOnStateStreamError('write to', state)); | 679 createCannotActionOnStateStreamError('write to', CLOSED)); |
588 } | 680 } |
589 TEMP_ASSERT(state === WRITABLE, | 681 if (stream[_pendingAbortRequest] !== undefined) { |
590 'state is "writable".'); | 682 return Promise_reject(new TypeError(errStreamAborting)); |
683 } | |
591 const promise = WritableStreamAddWriteRequest(stream); | 684 const promise = WritableStreamAddWriteRequest(stream); |
592 WritableStreamDefaultControllerWrite(stream[_writableStreamController], | 685 WritableStreamDefaultControllerWrite(controller, chunk, chunkSize); |
593 chunk); | |
594 return promise; | 686 return promise; |
595 } | 687 } |
596 | 688 |
597 // Functions to expose internals for ReadableStream.pipeTo. These do not | 689 // Functions to expose internals for ReadableStream.pipeTo. These do not |
598 // appear in the standard. | 690 // appear in the standard. |
599 function getWritableStreamDefaultWriterClosedPromise(writer) { | 691 function getWritableStreamDefaultWriterClosedPromise(writer) { |
600 TEMP_ASSERT( | 692 TEMP_ASSERT( |
601 IsWritableStreamDefaultWriter(writer), | 693 IsWritableStreamDefaultWriter(writer), |
602 'writer is a WritableStreamDefaultWriter.'); | 694 'writer is a WritableStreamDefaultWriter.'); |
603 return writer[_closedPromise]; | 695 return writer[_closedPromise]; |
604 } | 696 } |
605 | 697 |
606 function getWritableStreamDefaultWriterReadyPromise(writer) { | 698 function getWritableStreamDefaultWriterReadyPromise(writer) { |
607 TEMP_ASSERT( | 699 TEMP_ASSERT( |
608 IsWritableStreamDefaultWriter(writer), | 700 IsWritableStreamDefaultWriter(writer), |
609 'writer is a WritableStreamDefaultWriter.'); | 701 'writer is a WritableStreamDefaultWriter.'); |
610 return writer[_readyPromise]; | 702 return writer[_readyPromise]; |
611 } | 703 } |
612 | 704 |
613 class WritableStreamDefaultController { | 705 class WritableStreamDefaultController { |
614 constructor(stream, underlyingSink, size, highWaterMark) { | 706 constructor(stream, underlyingSink, size, highWaterMark) { |
615 if (!IsWritableStream(stream)) { | 707 if (!IsWritableStream(stream)) { |
616 throw new TypeError(streamErrors.illegalConstructor); | 708 throw new TypeError(streamErrors.illegalConstructor); |
617 } | 709 } |
618 if (stream[_writableStreamController] !== undefined) { | 710 if (stream[_writableStreamController] !== undefined) { |
619 throw new TypeError(streamErrors.illegalConstructor); | 711 throw new TypeError(streamErrors.illegalConstructor); |
620 } | 712 } |
621 this[_controlledWritableStream] = stream; | 713 this[_controlledWritableStream] = stream; |
622 this[_underlyingSink] = underlyingSink; | 714 this[_underlyingSink] = underlyingSink; |
623 this[_queue] = new binding.SimpleQueue(); | 715 // These are just initialised to avoid triggering the assert() in |
624 this[_queueSize] = 0; | 716 // ResetQueue. They are overwritten by ResetQueue(). |
625 this[_defaultControllerFlags] = 0; | 717 this[_queue] = undefined; |
718 this[_queueTotalSize] = undefined; | |
719 ResetQueue(this); | |
720 this[_started] = false; | |
626 const normalizedStrategy = | 721 const normalizedStrategy = |
627 ValidateAndNormalizeQueuingStrategy(size, highWaterMark); | 722 ValidateAndNormalizeQueuingStrategy(size, highWaterMark); |
628 this[_strategySize] = normalizedStrategy.size; | 723 this[_strategySize] = normalizedStrategy.size; |
629 this[_strategyHWM] = normalizedStrategy.highWaterMark; | 724 this[_strategyHWM] = normalizedStrategy.highWaterMark; |
630 const backpressure = WritableStreamDefaultControllerGetBackpressure(this); | 725 const backpressure = WritableStreamDefaultControllerGetBackpressure(this); |
631 if (backpressure) { | 726 WritableStreamUpdateBackpressure(stream, backpressure); |
632 WritableStreamUpdateBackpressure(stream, backpressure); | |
633 } | |
634 const controller = this; | |
635 const startResult = InvokeOrNoop(underlyingSink, 'start', [this]); | |
636 const onFulfilled = () => { | |
637 setDefaultControllerStartedFlag(controller, true); | |
638 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); | |
639 }; | |
640 const onRejected = r => { | |
641 WritableStreamDefaultControllerErrorIfNeeded(controller, r); | |
642 }; | |
643 thenPromise(Promise_resolve(startResult), onFulfilled, onRejected); | |
644 } | 727 } |
645 | 728 |
646 error(e) { | 729 error(e) { |
647 if (!IsWritableStreamDefaultController(this)) { | 730 if (!IsWritableStreamDefaultController(this)) { |
648 throw new TypeError(streamErrors.illegalInvocation); | 731 throw new TypeError(streamErrors.illegalInvocation); |
649 } | 732 } |
650 const state = this[_controlledWritableStream][_state]; | 733 const state = this[_controlledWritableStream][_stateAndFlags] & STATE_MASK ; |
651 if (state === CLOSED || state === ERRORED) { | 734 if (state === CLOSED || state === ERRORED) { |
652 throw createCannotActionOnStateStreamError('error', state); | 735 throw createCannotActionOnStateStreamError('error', state); |
653 } | 736 } |
654 WritableStreamDefaultControllerError(this, e); | 737 WritableStreamDefaultControllerError(this, e); |
655 } | 738 } |
656 } | 739 } |
657 | 740 |
741 // Writable Stream Default Controller Internal Methods | |
742 | |
743 // TODO(ricea): Virtual dispatch via V8 Private Symbols seems to be difficult | |
744 // or impossible, so use static dispatch for now. This will have to be fixed | |
745 // when adding a byte controller. | |
746 function WritableStreamDefaultControllerAbortSteps(controller, reason) { | |
747 const sinkAbortPromise = | |
748 PromiseInvokeOrNoop(controller[_underlyingSink], 'abort', [reason]); | |
749 return thenPromise(sinkAbortPromise, () => undefined); | |
750 } | |
751 | |
752 function WritableStreamDefaultControllerErrorSteps(controller) { | |
753 ResetQueue(controller); | |
754 } | |
755 | |
756 function WritableStreamDefaultControllerStartSteps(controller) { | |
757 const startResult = | |
758 InvokeOrNoop(controller[_underlyingSink], 'start', [controller]); | |
759 const stream = controller[_controlledWritableStream]; | |
760 const startPromise = Promise_resolve(startResult); | |
761 thenPromise( | |
762 startPromise, | |
763 () => { | |
764 controller[_started] = true; | |
765 if ((stream[_stateAndFlags] & STATE_MASK) === ERRORED) { | |
766 WritableStreamRejectAbortRequestIfPending(stream); | |
767 } else { | |
768 WritableStreamHandleAbortRequestIfPending(stream); | |
769 } | |
770 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); | |
771 }, | |
772 r => { | |
773 TEMP_ASSERT( | |
774 (stream[_stateAndFlags] & STATE_MASK) === WRITABLE || | |
775 (stream[_stateAndFlags] & STATE_MASK) === ERRORED, | |
776 '_stream_.[[state]] is `"writable"` or `"errored"`.'); | |
777 WritableStreamDefaultControllerErrorIfNeeded(controller, r); | |
778 WritableStreamRejectAbortRequestIfPending(stream); | |
779 }); | |
780 } | |
781 | |
658 // Writable Stream Default Controller Abstract Operations | 782 // Writable Stream Default Controller Abstract Operations |
659 | 783 |
660 function IsWritableStreamDefaultController(x) { | 784 function IsWritableStreamDefaultController(x) { |
661 return hasOwnProperty(x, _underlyingSink); | 785 return hasOwnProperty(x, _underlyingSink); |
662 } | 786 } |
663 | 787 |
664 function WritableStreamDefaultControllerAbort(controller, reason) { | |
665 controller[_queue] = new binding.SimpleQueue(); | |
666 controller[_queueSize] = 0; | |
667 const sinkAbortPromise = | |
668 PromiseInvokeOrNoop(controller[_underlyingSink], 'abort', [reason]); | |
669 return thenPromise(sinkAbortPromise, () => undefined); | |
670 } | |
671 | |
672 function WritableStreamDefaultControllerClose(controller) { | 788 function WritableStreamDefaultControllerClose(controller) { |
673 EnqueueValueWithSizeForController(controller, 'close', 0); | 789 EnqueueValueWithSize(controller, 'close', 0); |
674 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); | 790 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); |
675 } | 791 } |
676 | 792 |
677 function WritableStreamDefaultControllerGetDesiredSize(controller) { | 793 function WritableStreamDefaultControllerGetChunkSize(controller, chunk) { |
678 const queueSize = GetTotalQueueSizeForController(controller); | 794 const strategySize = controller[_strategySize]; |
679 return controller[_strategyHWM] - queueSize; | 795 if (strategySize === undefined) { |
796 return 1; | |
797 } | |
798 let value; | |
799 try { | |
800 value = Function_call(strategySize, undefined, chunk); | |
801 } catch (e) { | |
802 WritableStreamDefaultControllerErrorIfNeeded(controller, e); | |
803 return 1; | |
804 } | |
805 return value; | |
680 } | 806 } |
681 | 807 |
682 function WritableStreamDefaultControllerWrite(controller, chunk) { | 808 function WritableStreamDefaultControllerGetDesiredSize(controller) { |
683 const stream = controller[_controlledWritableStream]; | 809 return controller[_strategyHWM] - controller[_queueTotalSize]; |
684 TEMP_ASSERT(stream[_state] === WRITABLE, | 810 } |
685 'stream.[[state]] is "writable".'); | 811 |
686 let chunkSize = 1; | 812 function WritableStreamDefaultControllerWrite(controller, chunk, chunkSize) { |
687 const strategySize = controller[_strategySize]; | |
688 if (strategySize !== undefined) { | |
689 try { | |
690 chunkSize = strategySize(chunk); | |
691 } catch (e) { | |
692 WritableStreamDefaultControllerErrorIfNeeded(controller, e); | |
693 return; | |
694 } | |
695 } | |
696 const writeRecord = {chunk}; | 813 const writeRecord = {chunk}; |
697 const lastBackpressure = | |
698 WritableStreamDefaultControllerGetBackpressure(controller); | |
699 try { | 814 try { |
700 EnqueueValueWithSizeForController(controller, writeRecord, chunkSize); | 815 EnqueueValueWithSize(controller, writeRecord, chunkSize); |
701 } catch (e) { | 816 } catch (e) { |
702 WritableStreamDefaultControllerErrorIfNeeded(controller, e); | 817 WritableStreamDefaultControllerErrorIfNeeded(controller, e); |
703 return; | 818 return; |
704 } | 819 } |
705 if (stream[_state] === WRITABLE) { | 820 const stream = controller[_controlledWritableStream]; |
821 if (!WritableStreamCloseQueuedOrInFlight(stream)) { | |
706 const backpressure = | 822 const backpressure = |
707 WritableStreamDefaultControllerGetBackpressure(controller); | 823 WritableStreamDefaultControllerGetBackpressure(controller); |
708 if (lastBackpressure !== backpressure) { | 824 WritableStreamUpdateBackpressure(stream, backpressure); |
709 WritableStreamUpdateBackpressure(stream, backpressure); | |
710 } | |
711 } | 825 } |
712 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); | 826 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); |
713 } | 827 } |
714 | 828 |
715 function WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { | 829 function WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller) { |
716 const state = controller[_controlledWritableStream][_state]; | 830 const stream = controller[_controlledWritableStream]; |
831 const state = stream[_stateAndFlags] & STATE_MASK; | |
717 if (state === CLOSED || state === ERRORED) { | 832 if (state === CLOSED || state === ERRORED) { |
718 return; | 833 return; |
719 } | 834 } |
720 if (!getDefaultControllerStartedFlag(controller)) { | 835 if (!controller[_started]) { |
721 return; | 836 return; |
722 } | 837 } |
723 if (getDefaultControllerWritingFlag(controller)) { | 838 if (stream[_inFlightWriteRequest] !== undefined) { |
724 return; | 839 return; |
725 } | 840 } |
726 if (controller[_queue].length === 0) { | 841 if (controller[_queue].length === 0) { |
727 return; | 842 return; |
728 } | 843 } |
729 const writeRecord = PeekQueueValue(controller[_queue]); | 844 const writeRecord = PeekQueueValue(controller); |
730 if (writeRecord === 'close') { | 845 if (writeRecord === 'close') { |
731 WritableStreamDefaultControllerProcessClose(controller); | 846 WritableStreamDefaultControllerProcessClose(controller); |
732 } else { | 847 } else { |
733 WritableStreamDefaultControllerProcessWrite(controller, | 848 WritableStreamDefaultControllerProcessWrite(controller, |
734 writeRecord.chunk); | 849 writeRecord.chunk); |
735 } | 850 } |
736 } | 851 } |
737 | 852 |
738 function WritableStreamDefaultControllerErrorIfNeeded(controller, e) { | 853 function WritableStreamDefaultControllerErrorIfNeeded(controller, error) { |
739 const state = controller[_controlledWritableStream][_state]; | 854 const state = controller[_controlledWritableStream][_stateAndFlags] & STATE_ MASK; |
740 if (state === WRITABLE || state === CLOSING) { | 855 if (state === WRITABLE) { |
741 WritableStreamDefaultControllerError(controller, e); | 856 WritableStreamDefaultControllerError(controller, error); |
742 } | 857 } |
743 } | 858 } |
744 | 859 |
745 function WritableStreamDefaultControllerProcessClose(controller) { | 860 function WritableStreamDefaultControllerProcessClose(controller) { |
746 const stream = controller[_controlledWritableStream]; | 861 const stream = controller[_controlledWritableStream]; |
747 TEMP_ASSERT(stream[_state] === CLOSING, | 862 WritableStreamMarkCloseRequestInFlight(stream); |
748 'stream.[[state]] is "closing".'); | 863 DequeueValue(controller); |
749 DequeueValueForController(controller); | |
750 TEMP_ASSERT(controller[_queue].length === 0, | 864 TEMP_ASSERT(controller[_queue].length === 0, |
751 'controller.[[queue]] is empty.'); | 865 'controller.[[queue]] is empty.'); |
752 setDefaultControllerInCloseFlag(controller, true); | |
753 const sinkClosePromise = PromiseInvokeOrNoop(controller[_underlyingSink], | 866 const sinkClosePromise = PromiseInvokeOrNoop(controller[_underlyingSink], |
754 'close', [controller]); | 867 'close', [controller]); |
755 thenPromise(sinkClosePromise, | 868 thenPromise( |
756 () => { | 869 sinkClosePromise, |
757 TEMP_ASSERT(getDefaultControllerInCloseFlag(controller) | 870 () => WritableStreamFinishInFlightClose(stream), |
758 === true, | 871 reason => WritableStreamFinishInFlightCloseWithError(stream, reason) |
759 'controller.[[inClose]] is true'); | 872 ); |
760 setDefaultControllerInCloseFlag(controller, false); | |
761 | |
762 if (stream[_state] !== CLOSING && | |
763 stream[_state] !== ERRORED) { | |
764 return; | |
765 } | |
766 | |
767 TEMP_ASSERT(stream[_pendingCloseRequest] !== undefined); | |
768 v8.resolvePromise(stream[_pendingCloseRequest], undefined); | |
769 stream[_pendingCloseRequest] = undefined; | |
770 | |
771 WritableStreamFinishClose(stream); | |
772 }, | |
773 r => { | |
774 TEMP_ASSERT(getDefaultControllerInCloseFlag(controller) | |
775 === true, | |
776 'controller.[[inClose]] is true'); | |
777 setDefaultControllerInCloseFlag(controller, false); | |
778 | |
779 TEMP_ASSERT(stream[_pendingCloseRequest] !== undefined); | |
780 v8.rejectPromise(stream[_pendingCloseRequest], r); | |
781 stream[_pendingCloseRequest] = undefined; | |
782 | |
783 if (stream[_pendingAbortRequest] !== undefined) { | |
784 v8.rejectPromise(stream[_pendingAbortRequest], r); | |
785 stream[_pendingAbortRequest] = undefined; | |
786 } | |
787 | |
788 WritableStreamDefaultControllerErrorIfNeeded(controller, r); | |
789 } | |
790 ); | |
791 } | 873 } |
792 | 874 |
793 function WritableStreamDefaultControllerProcessWrite(controller, chunk) { | 875 function WritableStreamDefaultControllerProcessWrite(controller, chunk) { |
794 setDefaultControllerWritingFlag(controller, true); | |
795 | |
796 const stream = controller[_controlledWritableStream]; | 876 const stream = controller[_controlledWritableStream]; |
797 | 877 WritableStreamMarkFirstWriteRequestInFlight(stream); |
798 TEMP_ASSERT(stream[_pendingWriteRequest] === undefined, | |
799 'stream.[[pendingWriteRequest]] is undefined'); | |
800 TEMP_ASSERT(stream[_writeRequests].length > 0, | |
801 'stream.[[writeRequests]] is not empty'); | |
802 stream[_pendingWriteRequest] = stream[_writeRequests].shift(); | |
803 | |
804 const sinkWritePromise = PromiseInvokeOrNoop(controller[_underlyingSink], | 878 const sinkWritePromise = PromiseInvokeOrNoop(controller[_underlyingSink], |
805 'write', [chunk, controller]); | 879 'write', [chunk, controller]); |
806 thenPromise( | 880 thenPromise( |
807 sinkWritePromise, | 881 sinkWritePromise, |
808 () => { | 882 () => { |
809 TEMP_ASSERT(getDefaultControllerWritingFlag(controller) === true, | 883 WritableStreamFinishInFlightWrite(stream); |
810 'controller.[[writing]] is true'); | 884 const state = stream[_stateAndFlags] & STATE_MASK; |
811 setDefaultControllerWritingFlag(controller, false); | |
812 | |
813 TEMP_ASSERT(stream[_pendingWriteRequest] !== undefined, | |
814 'stream.[[pendingWriteRequest]] is not undefined'); | |
815 v8.resolvePromise(stream[_pendingWriteRequest], undefined); | |
816 stream[_pendingWriteRequest] = undefined; | |
817 | |
818 const state = stream[_state]; | |
819 if (state === ERRORED) { | 885 if (state === ERRORED) { |
820 WritableStreamRejectPromisesInReactionToError(stream); | |
821 | |
822 if (stream[_pendingAbortRequest] !== undefined) { | |
823 v8.resolvePromise(stream[_pendingAbortRequest], undefined); | |
824 stream[_pendingAbortRequest] = undefined; | |
825 } | |
826 return; | 886 return; |
827 } | 887 } |
828 | 888 TEMP_ASSERT(state === WRITABLE, '_state_ is `"writable"`.'); |
829 const lastBackpressure = | 889 DequeueValue(controller); |
830 WritableStreamDefaultControllerGetBackpressure(controller); | 890 if (!WritableStreamCloseQueuedOrInFlight(stream)) { |
831 DequeueValueForController(controller); | |
832 | |
833 if (state !== CLOSING) { | |
834 const backpressure = | 891 const backpressure = |
835 WritableStreamDefaultControllerGetBackpressure(controller); | 892 WritableStreamDefaultControllerGetBackpressure(controller); |
836 if (lastBackpressure !== backpressure) { | 893 WritableStreamUpdateBackpressure(stream, backpressure); |
837 WritableStreamUpdateBackpressure( | |
838 controller[_controlledWritableStream], backpressure); | |
839 } | |
840 } | 894 } |
841 | |
842 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); | 895 WritableStreamDefaultControllerAdvanceQueueIfNeeded(controller); |
843 }, | 896 }, |
844 r => { | 897 reason => { |
845 TEMP_ASSERT(getDefaultControllerWritingFlag(controller) === true, | 898 const wasErrored = (stream[_stateAndFlags] & STATE_MASK) === ERRORED; |
846 'controller.[[writing]] is true'); | 899 WritableStreamFinishInFlightWriteWithError(stream, reason); |
847 setDefaultControllerWritingFlag(controller, false); | 900 TEMP_ASSERT((stream[_stateAndFlags] & STATE_MASK) === ERRORED, |
848 | 901 '_stream_.[[state]] is `"errored"`.'); |
849 TEMP_ASSERT(stream[_pendingWriteRequest] !== undefined, | 902 if (!wasErrored) { |
850 'stream.[[pendingWriteRequest]] is not undefined'); | 903 ResetQueue(controller); |
tyoshino (SeeGerritForStatus)
2017/04/12 04:21:13
In the spec, only [[queue]] is cleared here. Was t
Adam Rice
2017/04/13 05:54:40
I think I intended to update the spec, but it look
tyoshino (SeeGerritForStatus)
2017/04/13 07:38:07
OK. Thanks
| |
851 v8.rejectPromise(stream[_pendingWriteRequest], r); | |
852 stream[_pendingWriteRequest] = undefined; | |
853 | |
854 if (stream[_state] === ERRORED) { | |
855 stream[_storedError] = r; | |
856 WritableStreamRejectPromisesInReactionToError(stream); | |
857 } | 904 } |
858 | 905 }); |
859 if (stream[_pendingAbortRequest] !== undefined) { | |
860 v8.rejectPromise(stream[_pendingAbortRequest], r); | |
861 stream[_pendingAbortRequest] = undefined; | |
862 } | |
863 | |
864 WritableStreamDefaultControllerErrorIfNeeded(controller, r); | |
865 } | |
866 ); | |
867 } | 906 } |
868 | 907 |
869 function WritableStreamDefaultControllerGetBackpressure(controller) { | 908 function WritableStreamDefaultControllerGetBackpressure(controller) { |
870 const desiredSize = | 909 const desiredSize = |
871 WritableStreamDefaultControllerGetDesiredSize(controller); | 910 WritableStreamDefaultControllerGetDesiredSize(controller); |
872 return desiredSize <= 0; | 911 return desiredSize <= 0; |
873 } | 912 } |
874 | 913 |
875 function WritableStreamDefaultControllerError(controller, e) { | 914 function WritableStreamDefaultControllerError(controller, error) { |
876 const stream = controller[_controlledWritableStream]; | 915 const stream = controller[_controlledWritableStream]; |
877 const state = stream[_state]; | 916 TEMP_ASSERT((stream[_stateAndFlags] & STATE_MASK) === WRITABLE, |
878 TEMP_ASSERT(state === WRITABLE || state === CLOSING, | 917 '_stream_.[[state]] is `"writable"`.'); |
879 'stream.[[state]] is "writable" or "closing".'); | 918 WritableStreamError(stream, error); |
880 WritableStreamError(stream, e); | |
881 controller[_queue] = new binding.SimpleQueue(); | |
882 controller[_queueSize] = 0; | |
883 } | 919 } |
884 | 920 |
885 // Queue-with-Sizes Operations | 921 // Queue-with-Sizes Operations |
886 // | 922 // |
887 // TODO(ricea): Share these operations with ReadableStream.js. | 923 // TODO(ricea): Share these operations with ReadableStream.js. |
888 function DequeueValueForController(controller) { | 924 function DequeueValue(container) { |
889 TEMP_ASSERT(controller[_queue].length !== 0, | 925 TEMP_ASSERT( |
890 'queue is not empty.'); | 926 hasOwnProperty(container, _queue) && |
891 const result = controller[_queue].shift(); | 927 hasOwnProperty(container, _queueTotalSize), |
892 controller[_queueSize] -= result.size; | 928 'Assert: _container_ has [[queue]] and [[queueTotalSize]] internal ' + |
893 return result.value; | 929 'slots.'); |
930 TEMP_ASSERT(container[_queue].length !== 0, | |
931 '_container_.[[queue]] is not empty.'); | |
932 const pair = container[_queue].shift(); | |
933 container[_queueTotalSize] -= pair.size; | |
934 if (container[_queueTotalSize] < 0) { | |
935 container[_queueTotalSize] = 0; | |
936 } | |
937 return pair.value; | |
894 } | 938 } |
895 | 939 |
896 function EnqueueValueWithSizeForController(controller, value, size) { | 940 function EnqueueValueWithSize(container, value, size) { |
941 TEMP_ASSERT( | |
942 hasOwnProperty(container, _queue) && | |
943 hasOwnProperty(container, _queueTotalSize), | |
944 'Assert: _container_ has [[queue]] and [[queueTotalSize]] internal ' + | |
945 'slots.'); | |
897 size = Number(size); | 946 size = Number(size); |
898 if (!IsFiniteNonNegativeNumber(size)) { | 947 if (!IsFiniteNonNegativeNumber(size)) { |
899 throw new RangeError(streamErrors.invalidSize); | 948 throw new RangeError(streamErrors.invalidSize); |
900 } | 949 } |
901 | 950 |
902 controller[_queueSize] += size; | 951 container[_queue].push({value, size}); |
903 controller[_queue].push({value, size}); | 952 container[_queueTotalSize] += size; |
904 } | 953 } |
905 | 954 |
906 function GetTotalQueueSizeForController(controller) { | 955 function PeekQueueValue(container) { |
907 return controller[_queueSize]; | 956 TEMP_ASSERT( |
957 hasOwnProperty(container, _queue) && | |
958 hasOwnProperty(container, _queueTotalSize), | |
959 'Assert: _container_ has [[queue]] and [[queueTotalSize]] internal ' + | |
960 'slots.'); | |
961 TEMP_ASSERT(container[_queue].length !== 0, | |
962 '_container_.[[queue]] is not empty.'); | |
963 const pair = container[_queue].peek(); | |
964 return pair.value; | |
908 } | 965 } |
909 | 966 |
910 function PeekQueueValue(queue) { | 967 function ResetQueue(container) { |
911 TEMP_ASSERT(queue.length !== 0, | 968 TEMP_ASSERT( |
912 'queue is not empty.'); | 969 hasOwnProperty(container, _queue) && |
913 return queue.peek().value; | 970 hasOwnProperty(container, _queueTotalSize), |
971 'Assert: _container_ has [[queue]] and [[queueTotalSize]] internal ' + | |
972 'slots.'); | |
973 container[_queue] = new binding.SimpleQueue(); | |
974 container[_queueTotalSize] = 0; | |
914 } | 975 } |
915 | 976 |
916 // Miscellaneous Operations | 977 // Miscellaneous Operations |
917 | 978 |
918 // This differs from "CallOrNoop" in the ReadableStream implementation in | 979 // This differs from "CallOrNoop" in the ReadableStream implementation in |
919 // that it takes the arguments as an array, so that multiple arguments can be | 980 // that it takes the arguments as an array, so that multiple arguments can be |
920 // passed. | 981 // passed. |
921 // | 982 // |
922 // TODO(ricea): Consolidate with ReadableStream implementation. | 983 // TODO(ricea): Consolidate with ReadableStream implementation. |
923 function InvokeOrNoop(O, P, args) { | 984 function InvokeOrNoop(O, P, args) { |
(...skipping 68 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... | |
992 getWritableStreamDefaultWriterClosedPromise; | 1053 getWritableStreamDefaultWriterClosedPromise; |
993 binding.WritableStreamDefaultWriterGetDesiredSize = | 1054 binding.WritableStreamDefaultWriterGetDesiredSize = |
994 WritableStreamDefaultWriterGetDesiredSize; | 1055 WritableStreamDefaultWriterGetDesiredSize; |
995 binding.getWritableStreamDefaultWriterReadyPromise = | 1056 binding.getWritableStreamDefaultWriterReadyPromise = |
996 getWritableStreamDefaultWriterReadyPromise; | 1057 getWritableStreamDefaultWriterReadyPromise; |
997 binding.WritableStreamDefaultWriterRelease = | 1058 binding.WritableStreamDefaultWriterRelease = |
998 WritableStreamDefaultWriterRelease; | 1059 WritableStreamDefaultWriterRelease; |
999 binding.WritableStreamDefaultWriterWrite = WritableStreamDefaultWriterWrite; | 1060 binding.WritableStreamDefaultWriterWrite = WritableStreamDefaultWriterWrite; |
1000 binding.getWritableStreamStoredError = getWritableStreamStoredError; | 1061 binding.getWritableStreamStoredError = getWritableStreamStoredError; |
1001 }); | 1062 }); |
OLD | NEW |