1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
|
// Copyright 2019 The Chromium Authors
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
#ifndef THIRD_PARTY_BLINK_RENDERER_CORE_STREAMS_READABLE_STREAM_H_
#define THIRD_PARTY_BLINK_RENDERER_CORE_STREAMS_READABLE_STREAM_H_
#include <stdint.h>
#include <memory>
#include "third_party/blink/renderer/bindings/core/v8/async_iterable.h"
#include "third_party/blink/renderer/bindings/core/v8/script_value.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_async_iterator_readable_stream.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_readable_stream_iterator_options.h"
#include "third_party/blink/renderer/bindings/core/v8/v8_typedefs.h"
#include "third_party/blink/renderer/core/core_export.h"
#include "third_party/blink/renderer/core/streams/readable_stream_default_reader.h"
#include "third_party/blink/renderer/core/streams/transferable_streams.h"
#include "third_party/blink/renderer/platform/bindings/script_wrappable.h"
#include "third_party/blink/renderer/platform/bindings/trace_wrapper_v8_reference.h"
#include "third_party/blink/renderer/platform/heap/garbage_collected.h"
#include "third_party/blink/renderer/platform/heap/visitor.h"
#include "v8/include/v8.h"
namespace blink {
class ExceptionState;
class MessagePort;
class PipeOptions;
class ReadableByteStreamController;
class ReadableStreamBYOBReader;
class ReadableStreamController;
class ReadableStreamDefaultController;
class ReadableStreamGetReaderOptions;
class ReadableStreamTransferringOptimizer;
class ReadableWritablePair;
class ReadIntoRequest;
class ReadRequest;
class ScriptState;
class StrategySizeAlgorithm;
class StreamAlgorithm;
class StreamPipeOptions;
class StreamStartAlgorithm;
class UnderlyingByteSourceBase;
class UnderlyingSourceBase;
class WritableStream;
// C++ implementation of ReadableStream.
// See https://streams.spec.whatwg.org/#rs-model for background.
class CORE_EXPORT ReadableStream
: public ScriptWrappable,
public ValueAsyncIterable<ReadableStream,
ReadableStreamIteratorOptions*> {
DEFINE_WRAPPERTYPEINFO();
public:
enum State : uint8_t { kReadable, kClosed, kErrored };
// Zero-argument form of the constructor called from JavaScript.
static ReadableStream* Create(ScriptState*, ExceptionState&);
// One-argument constructor called from JavaScript.
static ReadableStream* Create(ScriptState*,
ScriptValue underlying_source,
ExceptionState&);
// Two-argument constructor called from JavaScript.
static ReadableStream* Create(ScriptState* script_state,
ScriptValue underlying_source,
ScriptValue strategy,
ExceptionState& exception_state);
// Entry point to create a ReadableStream from other C++ APIs.
static ReadableStream* CreateWithCountQueueingStrategy(
ScriptState* script_state,
UnderlyingSourceBase* underlying_source,
size_t high_water_mark);
// Specifying true for `allow_per_chunk_transferring` implies the following:
// 1. Each chunk has never been exposed to scripts.
// 2. Each chunk is transferable.
static ReadableStream* CreateWithCountQueueingStrategy(
ScriptState* script_state,
UnderlyingSourceBase* underlying_source,
size_t high_water_mark,
AllowPerChunkTransferring allow_per_chunk_transferring,
std::unique_ptr<ReadableStreamTransferringOptimizer> optimizer);
// CreateReadableStream():
// https://streams.spec.whatwg.org/#create-readable-stream
static ReadableStream* Create(ScriptState*,
StreamStartAlgorithm* start_algorithm,
StreamAlgorithm* pull_algorithm,
StreamAlgorithm* cancel_algorithm,
double high_water_mark,
StrategySizeAlgorithm* size_algorithm,
ExceptionState&);
// https://streams.spec.whatwg.org/#abstract-opdef-createreadablebytestream
static ReadableStream* CreateByteStream(ScriptState*,
StreamStartAlgorithm* start_algorithm,
StreamAlgorithm* pull_algorithm,
StreamAlgorithm* cancel_algorithm,
ExceptionState&);
// Entry point to create a ReadableByteStream from other C++ APIs.
// CreateReadableByteStream():
// https://streams.spec.whatwg.org/#abstract-opdef-createreadablebytestream
static ReadableStream* CreateByteStream(
ScriptState*,
UnderlyingByteSourceBase* underlying_byte_source);
static void InitByteStream(ScriptState*,
ReadableStream*,
UnderlyingByteSourceBase* underlying_byte_source,
ExceptionState&);
static void InitByteStream(ScriptState*,
ReadableStream*,
ReadableByteStreamController*,
StreamStartAlgorithm* start_algorithm,
StreamAlgorithm* pull_algorithm,
StreamAlgorithm* cancel_algorithm,
ExceptionState&);
ReadableStream();
~ReadableStream() override;
// See CreateWithCountQueueingStrategy() comment above for how to use
// `allow_per_chunk_transferring`.
void InitWithCountQueueingStrategy(
ScriptState*,
UnderlyingSourceBase*,
size_t high_water_mark,
AllowPerChunkTransferring allow_per_chunk_transferring,
std::unique_ptr<ReadableStreamTransferringOptimizer>,
ExceptionState&);
// https://streams.spec.whatwg.org/#rs-constructor
bool locked() const;
ScriptPromise<IDLUndefined> cancel(ScriptState*, ExceptionState&);
// https://streams.spec.whatwg.org/#rs-cancel
ScriptPromise<IDLUndefined> cancel(ScriptState*,
ScriptValue reason,
ExceptionState&);
V8ReadableStreamReader* getReader(ScriptState* script_state,
ExceptionState& exception_state);
// https://streams.spec.whatwg.org/#rs-get-reader
V8ReadableStreamReader* getReader(
ScriptState* script_state,
const ReadableStreamGetReaderOptions* options,
ExceptionState& exception_state);
ReadableStreamDefaultReader* GetDefaultReaderForTesting(ScriptState*,
ExceptionState&);
ReadableStreamBYOBReader* GetBYOBReaderForTesting(ScriptState*,
ExceptionState&);
ReadableStream* pipeThrough(ScriptState*,
ReadableWritablePair* transform,
ExceptionState&);
// https://streams.spec.whatwg.org/#rs-pipe-through
ReadableStream* pipeThrough(ScriptState*,
ReadableWritablePair* transform,
const StreamPipeOptions* options,
ExceptionState&);
ScriptPromise<IDLUndefined> pipeTo(ScriptState*,
WritableStream* destination,
ExceptionState&);
// https://streams.spec.whatwg.org/#rs-pipe-to
ScriptPromise<IDLUndefined> pipeTo(ScriptState*,
WritableStream* destination,
const StreamPipeOptions* options,
ExceptionState&);
// https://streams.spec.whatwg.org/#rs-tee
HeapVector<Member<ReadableStream>> tee(ScriptState*, ExceptionState&);
void Tee(ScriptState*,
ReadableStream** branch1,
ReadableStream** branch2,
bool clone_for_branch2,
ExceptionState&);
void ByteStreamTee(ScriptState*,
ReadableStream** branch1,
ReadableStream** branch2,
ExceptionState&);
bool IsLocked() const { return IsLocked(this); }
bool IsDisturbed() const { return IsDisturbed(this); }
bool IsReadable() const { return IsReadable(this); }
bool IsClosed() const { return IsClosed(this); }
bool IsErrored() const { return IsErrored(this); }
void LockAndDisturb(ScriptState*);
// https://streams.spec.whatwg.org/#readablestream-close
void CloseStream(ScriptState*, ExceptionState&);
void Serialize(ScriptState*, MessagePort* port, ExceptionState&);
static ReadableStream* Deserialize(
ScriptState*,
MessagePort* port,
std::unique_ptr<ReadableStreamTransferringOptimizer> optimizer,
ExceptionState&);
//
// Readable stream abstract operations
//
// https://streams.spec.whatwg.org/#is-readable-stream-disturbed
static bool IsDisturbed(const ReadableStream* stream) {
return stream->is_disturbed_;
}
// https://streams.spec.whatwg.org/#is-readable-stream-locked
static bool IsLocked(const ReadableStream* stream) {
return stream->reader_ != nullptr;
}
// https://streams.spec.whatwg.org/#readable-stream-pipe-to
static ScriptPromise<IDLUndefined> PipeTo(ScriptState*,
ReadableStream*,
WritableStream*,
PipeOptions*,
ExceptionState&);
// https://streams.spec.whatwg.org/#acquire-readable-stream-reader
static ReadableStreamDefaultReader* AcquireDefaultReader(ScriptState*,
ReadableStream*,
ExceptionState&);
// https://streams.spec.whatwg.org/#acquire-readable-stream-byob-reader
static ReadableStreamBYOBReader* AcquireBYOBReader(ScriptState*,
ReadableStream*,
ExceptionState&);
// https://streams.spec.whatwg.org/#readable-stream-cancel
static ScriptPromise<IDLUndefined> Cancel(ScriptState*,
ReadableStream*,
v8::Local<v8::Value> reason);
//
// Functions exported for use by TransformStream. Not part of the standard.
//
static bool IsReadable(const ReadableStream* stream) {
return stream->state_ == kReadable;
}
static bool IsClosed(const ReadableStream* stream) {
return stream->state_ == kClosed;
}
static bool IsErrored(const ReadableStream* stream) {
return stream->state_ == kErrored;
}
ReadableStreamController* GetController() {
return readable_stream_controller_.Get();
}
v8::Local<v8::Value> GetStoredError(v8::Isolate*) const;
std::unique_ptr<ReadableStreamTransferringOptimizer>
TakeTransferringOptimizer();
void SetAllowPerChunkTransferringForTesting(AllowPerChunkTransferring value) {
allow_per_chunk_transferring_ = value;
}
void Trace(Visitor*) const override;
private:
friend class ByteStreamTeeEngine;
friend class PipeToEngine;
friend class ReadableByteStreamController;
friend class ReadableStreamBYOBReader;
friend class ReadableStreamDefaultController;
friend class ReadableStreamDefaultReader;
friend class ReadableStreamGenericReader;
friend class TeeEngine;
class PullAlgorithm;
class CancelAlgorithm;
class ReadHandleImpl;
class IterationSource;
class IterationReadRequest;
// https://streams.spec.whatwg.org/#rs-constructor
void InitInternal(ScriptState*,
ScriptValue raw_underlying_source,
ScriptValue raw_strategy,
bool created_by_ua,
ExceptionState&);
// https://streams.spec.whatwg.org/#initialize-readable-stream
static void Initialize(ReadableStream*);
static void AddReadIntoRequest(ScriptState*,
ReadableStream*,
ReadIntoRequest*);
// https://streams.spec.whatwg.org/#readable-stream-add-read-request
static void AddReadRequest(ScriptState*, ReadableStream*, ReadRequest*);
// https://streams.spec.whatwg.org/#readable-stream-close
static void Close(ScriptState*, ReadableStream*);
// https://streams.spec.whatwg.org/#readable-stream-error
static void Error(ScriptState*, ReadableStream*, v8::Local<v8::Value> e);
// https://streams.spec.whatwg.org/#readable-stream-fulfill-read-into-request
static void FulfillReadIntoRequest(ScriptState*,
ReadableStream*,
DOMArrayBufferView* chunk,
bool done,
ExceptionState&);
// https://streams.spec.whatwg.org/#readable-stream-fulfill-read-request
static void FulfillReadRequest(ScriptState*,
ReadableStream*,
v8::Local<v8::Value> chunk,
bool done,
ExceptionState&);
// https://streams.spec.whatwg.org/#readable-stream-get-num-read-into-requests
static int GetNumReadIntoRequests(const ReadableStream*);
// https://streams.spec.whatwg.org/#readable-stream-get-num-read-requests
static int GetNumReadRequests(const ReadableStream*);
// https://streams.spec.whatwg.org/#readable-stream-has-byob-reader
static bool HasBYOBReader(const ReadableStream*);
// https://streams.spec.whatwg.org/#readable-stream-has-default-reader
static bool HasDefaultReader(const ReadableStream*);
// Calls Tee() on |readable|, converts the two branches to a JavaScript array
// and returns them.
static HeapVector<Member<ReadableStream>> CallTeeAndReturnBranchArray(
ScriptState* script_state,
ReadableStream* readable,
bool clone_for_branch2,
ExceptionState& exception_state);
bool is_disturbed_ = false;
// When set to true, each chunk can be transferred instead of cloned on
// transferring the stream.
AllowPerChunkTransferring allow_per_chunk_transferring_{false};
State state_ = kReadable;
Member<ReadableStreamController> readable_stream_controller_;
Member<ReadableStreamGenericReader> reader_;
TraceWrapperV8Reference<v8::Value> stored_error_;
std::unique_ptr<ReadableStreamTransferringOptimizer> transferring_optimizer_;
// ValueAsyncIterable<ReadableStream> overrides:
using IterationSourceBase =
ValueAsyncIterable<ReadableStream,
ReadableStreamIteratorOptions*>::IterationSource;
IterationSourceBase* CreateIterationSource(
ScriptState* script_state,
IterationSourceBase::Kind kind,
ReadableStreamIteratorOptions* options,
ExceptionState& exception_state) override;
};
} // namespace blink
#endif // THIRD_PARTY_BLINK_RENDERER_CORE_STREAMS_READABLE_STREAM_H_
|
