Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(1166)

Issues Search
    My Issues | Starred     Open | Closed | All

Side by Side Diff: Source/modules/fetch/Body.cpp

Issue 1192913007: Change BodyStreamBuffer to be FetchDataConsumerHandle-based and enable backpressure in Fetch API (Closed) Base URL: https://chromium.googlesource.com/chromium/blink.git@master
Patch Set: (temp) alternative to calling didGetReadable in sync. Created 5 years, 5 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View unified diff | Download patch
« no previous file with comments | « Source/modules/fetch/Body.h ('k') | Source/modules/fetch/BodyStreamBuffer.h » ('j') | Source/modules/fetch/BodyStreamBuffer.h » ('J')
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 // Copyright 2014 The Chromium Authors. All rights reserved. 1 // Copyright 2014 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 #include "config.h" 5 #include "config.h"
6 #include "modules/fetch/Body.h" 6 #include "modules/fetch/Body.h"
7 7
8 #include "bindings/core/v8/ExceptionState.h" 8 #include "bindings/core/v8/ExceptionState.h"
9 #include "bindings/core/v8/ScriptPromiseResolver.h" 9 #include "bindings/core/v8/ScriptPromiseResolver.h"
10 #include "bindings/core/v8/ScriptState.h" 10 #include "bindings/core/v8/ScriptState.h"
11 #include "bindings/core/v8/V8ArrayBuffer.h" 11 #include "bindings/core/v8/V8ArrayBuffer.h"
12 #include "bindings/core/v8/V8ThrowException.h" 12 #include "bindings/core/v8/V8ThrowException.h"
13 #include "core/dom/DOMArrayBuffer.h" 13 #include "core/dom/DOMArrayBuffer.h"
14 #include "core/dom/DOMTypedArray.h" 14 #include "core/dom/DOMTypedArray.h"
15 #include "core/dom/ExceptionCode.h"
15 #include "core/fileapi/Blob.h" 16 #include "core/fileapi/Blob.h"
16 #include "core/fileapi/FileReaderLoader.h"
17 #include "core/fileapi/FileReaderLoaderClient.h"
18 #include "core/frame/UseCounter.h" 17 #include "core/frame/UseCounter.h"
19 #include "core/streams/ReadableByteStream.h" 18 #include "core/streams/ReadableByteStream.h"
20 #include "core/streams/ReadableByteStreamReader.h" 19 #include "core/streams/ReadableByteStreamReader.h"
21 #include "core/streams/UnderlyingSource.h" 20 #include "core/streams/UnderlyingSource.h"
22 #include "modules/fetch/BodyStreamBuffer.h" 21 #include "modules/fetch/BodyStreamBuffer.h"
22 #include "modules/fetch/DataConsumerHandleUtil.h"
23 #include "modules/fetch/FetchBlobDataConsumerHandle.h"
23 24
24 namespace blink { 25 namespace blink {
25 26
26 class Body::BlobHandleReceiver final : public BodyStreamBuffer::BlobHandleCreato rClient {
27 public:
28 explicit BlobHandleReceiver(Body* body)
29 : m_body(body)
30 {
31 }
32 void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
33 {
34 ASSERT(m_body);
35 m_body->readAsyncFromBlob(handle);
36 m_body = nullptr;
37 }
38 void didFail(DOMException* exception) override
39 {
40 ASSERT(m_body);
41 m_body->didBlobHandleReceiveError(exception);
42 m_body = nullptr;
43 }
44 DEFINE_INLINE_VIRTUAL_TRACE()
45 {
46 BodyStreamBuffer::BlobHandleCreatorClient::trace(visitor);
47 visitor->trace(m_body);
48 }
49 private:
50 Member<Body> m_body;
51 };
52
53 // This class is an ActiveDOMObject subclass only for holding the 27 // This class is an ActiveDOMObject subclass only for holding the
yhirano 2015/07/03 04:42:28 We do no longer need ExecutionContext, right?
hiroshige 2015/07/05 07:30:24 Done.
54 // ExecutionContext used in |pullSource|. 28 // ExecutionContext used in |pullSource|.
55 class Body::ReadableStreamSource : public BodyStreamBuffer::Observer, public Und erlyingSource, public FileReaderLoaderClient, public ActiveDOMObject { 29 class Body::ReadableStreamSource : public GarbageCollectedFinalized<Body::Readab leStreamSource>, public UnderlyingSource, public WebDataConsumerHandle::Client, public BodyStreamBuffer::DrainingStreamNotificationClient, public ActiveDOMObjec t {
56 USING_GARBAGE_COLLECTED_MIXIN(ReadableStreamSource); 30 USING_GARBAGE_COLLECTED_MIXIN(ReadableStreamSource);
57 public: 31 public:
58 enum State { 32 enum State {
yhirano 2015/07/03 04:42:28 Now we can remove this enum and rely on the stream
hiroshige 2015/07/05 07:30:24 Done.
59 Initial,
60 Streaming, 33 Streaming,
61 Closed, 34 Closed,
62 Errored, 35 Errored,
63 }; 36 };
64 ReadableStreamSource(ExecutionContext* executionContext, PassRefPtr<BlobData Handle> handle)
65 : ActiveDOMObject(executionContext)
66 , m_blobDataHandle(handle ? handle : BlobDataHandle::create(BlobData::cr eate(), 0))
67 , m_state(Initial)
68 {
69 suspendIfNeeded();
70 }
71 37
72 ReadableStreamSource(ExecutionContext* executionContext, BodyStreamBuffer* b uffer) 38 ReadableStreamSource(ExecutionContext* executionContext, BodyStreamBuffer* b uffer)
73 : ActiveDOMObject(executionContext) 39 : ActiveDOMObject(executionContext)
74 , m_bodyStreamBuffer(buffer) 40 , m_bodyStreamBuffer(buffer)
75 , m_state(Initial) 41 , m_state(Streaming)
42 , m_streamNeedsMore(false)
43 #if ENABLE(ASSERT)
44 , m_drained(false)
45 #endif
76 { 46 {
77 suspendIfNeeded(); 47 suspendIfNeeded();
78 } 48 if (m_bodyStreamBuffer)
79 49 obtainReader();
80 explicit ReadableStreamSource(ExecutionContext* executionContext)
81 : ActiveDOMObject(executionContext)
82 , m_blobDataHandle(BlobDataHandle::create(BlobData::create(), 0))
83 , m_state(Initial)
84 {
85 suspendIfNeeded();
86 } 50 }
87 51
88 ~ReadableStreamSource() override { } 52 ~ReadableStreamSource() override { }
89 53
90 State state() const { return m_state; } 54 State state() const { return m_state; }
91 55
92 void startStream(ReadableByteStream* stream) 56 void startStream(ReadableByteStream* stream)
93 { 57 {
94 m_stream = stream; 58 m_stream = stream;
95 stream->didSourceStart(); 59 stream->didSourceStart();
96 } 60 }
97 // Creates a new BodyStreamBuffer to drain the data. 61 // Creates a new BodyStreamBuffer to drain the data.
98 BodyStreamBuffer* createDrainingStream() 62 PassOwnPtr<DrainingBodyStreamBuffer> createDrainingStream()
99 { 63 {
100 ASSERT(m_state != Initial); 64 if (!m_bodyStreamBuffer)
65 return nullptr;
101 66
102 auto drainingStreamBuffer = new BodyStreamBuffer(new Canceller(this)); 67 ASSERT(!m_drained);
103 if (m_stream->stateInternal() == ReadableByteStream::Closed) { 68 #if ENABLE(ASSERT)
104 drainingStreamBuffer->close(); 69 m_drained = true;
105 return drainingStreamBuffer; 70 #endif
106 }
107 if (m_stream->stateInternal() == ReadableByteStream::Errored) {
108 drainingStreamBuffer->error(exception());
109 return drainingStreamBuffer;
110 }
111 71
112 ASSERT(!m_drainingStreamBuffer); 72 if (m_stream->stateInternal() == ReadableByteStream::Closed)
113 // Take back the data in |m_stream|. 73 m_bodyStreamBuffer = BodyStreamBuffer::createEmpty();
114 Deque<std::pair<RefPtr<DOMArrayBufferView>, size_t>> tmp_queue; 74 if (m_stream->stateInternal() == ReadableByteStream::Errored)
115 ASSERT(m_stream->stateInternal() == ReadableStream::Readable); 75 m_bodyStreamBuffer = BodyStreamBuffer::create(createFetchDataConsume rHandleFromWebHandle(createUnexpectedErrorDataConsumerHandle()));
116 m_stream->readInternal(tmp_queue);
117 while (!tmp_queue.isEmpty()) {
118 std::pair<RefPtr<DOMArrayBufferView>, size_t> data = tmp_queue.takeF irst();
119 drainingStreamBuffer->write(data.first->buffer());
120 }
121 if (m_state == Closed)
122 drainingStreamBuffer->close();
123 76
124 m_drainingStreamBuffer = drainingStreamBuffer; 77 m_reader.clear();
125 return m_drainingStreamBuffer; 78 return DrainingBodyStreamBuffer::create(m_bodyStreamBuffer, this);
126 } 79 }
80
127 DEFINE_INLINE_VIRTUAL_TRACE() 81 DEFINE_INLINE_VIRTUAL_TRACE()
128 { 82 {
129 visitor->trace(m_bodyStreamBuffer); 83 visitor->trace(m_bodyStreamBuffer);
130 visitor->trace(m_drainingStreamBuffer);
131 visitor->trace(m_stream); 84 visitor->trace(m_stream);
132 BodyStreamBuffer::Observer::trace(visitor);
133 UnderlyingSource::trace(visitor); 85 UnderlyingSource::trace(visitor);
86 DrainingStreamNotificationClient::trace(visitor);
134 ActiveDOMObject::trace(visitor); 87 ActiveDOMObject::trace(visitor);
135 } 88 }
136 89
137 void close() 90 void close()
138 { 91 {
139 if (m_state == Closed) { 92 if (m_state == Closed) {
140 // It is possible to call |close| from the source side (such 93 // It is possible to call |close| from the source side (such
141 // as blob loading finish) and from the consumer side (such as 94 // as blob loading finish) and from the consumer side (such as
142 // calling |cancel|). Thus we should ignore it here. 95 // calling |cancel|). Thus we should ignore it here.
143 return; 96 return;
144 } 97 }
145 m_state = Closed; 98 m_state = Closed;
146 if (m_drainingStreamBuffer) 99 m_reader.clear();
147 m_drainingStreamBuffer->close();
148 m_stream->close(); 100 m_stream->close();
149 } 101 }
150 void error() 102 void error()
151 { 103 {
152 m_state = Errored; 104 m_state = Errored;
153 if (m_drainingStreamBuffer) 105 m_reader.clear();
154 m_drainingStreamBuffer->error(exception()); 106 m_stream->error(DOMException::create(NetworkError, "network error"));
155 m_stream->error(exception()); 107 }
108
109 void obtainReader()
110 {
111 m_reader = m_bodyStreamBuffer->handle()->obtainReader(this);
156 } 112 }
157 113
158 private: 114 private:
159 class Canceller : public BodyStreamBuffer::Canceller { 115 void didFetchDataLoadFinishedFromDrainingStream()
160 public: 116 {
161 Canceller(ReadableStreamSource* source) : m_source(source) { } 117 ASSERT(m_bodyStreamBuffer);
yhirano 2015/07/03 04:42:28 ASSERT(m_drained)?
hiroshige 2015/07/05 07:30:24 Done.
162 void cancel() override 118
163 { 119 #if ENABLE(ASSERT)
164 m_source->cancel(); 120 m_drained = false;
121 #endif
122 obtainReader();
123 // We have to call didGetReadable() now to call close()/error() if
124 // necessary.
125 // didGetReadable() would be called asynchronously, but it is too late.
126 didGetReadable();
127 }
128
129 void didGetReadable() override
130 {
131 if (!m_streamNeedsMore) {
132 // Perform zero-length read to call close()/error() early.
133 size_t readSize;
134 WebDataConsumerHandle::Result result = m_reader->read(nullptr, 0, We bDataConsumerHandle::FlagNone, &readSize);
135 switch (result) {
136 case WebDataConsumerHandle::Ok:
137 case WebDataConsumerHandle::ShouldWait:
138 return;
139 case WebDataConsumerHandle::Done:
140 close();
141 return;
142 case WebDataConsumerHandle::Busy:
143 case WebDataConsumerHandle::ResourceExhausted:
144 case WebDataConsumerHandle::UnexpectedError:
145 error();
146 return;
147 }
165 } 148 }
166 149
167 DEFINE_INLINE_VIRTUAL_TRACE() 150 processData();
168 { 151 }
169 visitor->trace(m_source);
170 BodyStreamBuffer::Canceller::trace(visitor);
171 }
172
173 private:
174 Member<ReadableStreamSource> m_source;
175 };
176 152
177 // UnderlyingSource functions. 153 // UnderlyingSource functions.
178 void pullSource() override 154 void pullSource() override
179 { 155 {
180 // Note that one |pull| is called only when |read| is called on the 156 ASSERT(!m_streamNeedsMore);
181 // associated ReadableByteStreamReader because we create a stream with 157 m_streamNeedsMore = true;
182 // StrictStrategy. 158
183 if (m_state == Initial) { 159 ASSERT(!m_drained);
184 m_state = Streaming; 160
185 if (m_bodyStreamBuffer) { 161 processData();
186 m_bodyStreamBuffer->registerObserver(this); 162 }
187 onWrite(); 163
188 if (m_bodyStreamBuffer->hasError()) 164 ScriptPromise cancelSource(ScriptState* scriptState, ScriptValue reason) ove rride
189 return onError(); 165 {
190 if (m_bodyStreamBuffer->isClosed()) 166 close();
191 return onClose(); 167 return ScriptPromise::cast(scriptState, v8::Undefined(scriptState->isola te()));
192 } else { 168 }
193 FileReaderLoader::ReadType readType = FileReaderLoader::ReadAsAr rayBuffer; 169
194 m_loader = adoptPtr(new FileReaderLoader(readType, this)); 170 // Reads data and writes the data to |m_stream|, as long as data are
195 m_loader->start(executionContext(), m_blobDataHandle); 171 // available and the stream has pending reads.
172 void processData()
173 {
174 ASSERT(m_reader);
175 while (m_streamNeedsMore) {
176 const void* buffer;
177 size_t available;
178 WebDataConsumerHandle::Result result = m_reader->beginRead(&buffer, WebDataConsumerHandle::FlagNone, &available);
179 switch (result) {
180 case WebDataConsumerHandle::Ok:
181 m_streamNeedsMore = m_stream->enqueue(DOMUint8Array::create(stat ic_cast<const unsigned char*>(buffer), available));
182 m_reader->endRead(available);
183 break;
184
185 case WebDataConsumerHandle::Done:
186 close();
187 return;
188
189 case WebDataConsumerHandle::ShouldWait:
190 return;
191
192 case WebDataConsumerHandle::Busy:
193 case WebDataConsumerHandle::ResourceExhausted:
194 case WebDataConsumerHandle::UnexpectedError:
195 error();
196 return;
196 } 197 }
197 } 198 }
198 } 199 }
199 200
200 ScriptPromise cancelSource(ScriptState* scriptState, ScriptValue reason) ove rride 201 // Source of data.
201 { 202 Member<BodyStreamBuffer> m_bodyStreamBuffer;
202 cancel(); 203 OwnPtr<FetchDataConsumerHandle::Reader> m_reader;
203 return ScriptPromise::cast(scriptState, v8::Undefined(scriptState->isola te()));
204 }
205 204
206 // BodyStreamBuffer::Observer functions.
207 void onWrite() override
208 {
209 ASSERT(m_state == Streaming);
210 while (RefPtr<DOMArrayBuffer> buf = m_bodyStreamBuffer->read()) {
211 write(buf);
212 }
213 }
214 void onClose() override
215 {
216 ASSERT(m_state == Streaming);
217 close();
218 m_bodyStreamBuffer->unregisterObserver();
219 }
220 void onError() override
221 {
222 ASSERT(m_state == Streaming);
223 error();
224 m_bodyStreamBuffer->unregisterObserver();
225 }
226
227 // FileReaderLoaderClient functions.
228 void didStartLoading() override { }
229 void didReceiveData() override { }
230 void didFinishLoading() override
231 {
232 ASSERT(m_state == Streaming);
233 write(m_loader->arrayBufferResult());
234 close();
235 }
236 void didFail(FileError::ErrorCode) override
237 {
238 ASSERT(m_state == Streaming);
239 error();
240 }
241
242 void write(PassRefPtr<DOMArrayBuffer> buf)
243 {
244 if (m_drainingStreamBuffer) {
245 m_drainingStreamBuffer->write(buf);
246 } else {
247 auto size = buf->byteLength();
248 m_stream->enqueue(DOMUint8Array::create(buf, 0, size));
249 }
250 }
251 void cancel()
252 {
253 if (m_bodyStreamBuffer) {
254 m_bodyStreamBuffer->cancel();
255 // We should not close the stream here, because it is canceller's
256 // responsibility.
257 } else {
258 if (m_loader)
259 m_loader->cancel();
260 close();
261 }
262 }
263
264 DOMException* exception()
265 {
266 if (m_state != Errored)
267 return nullptr;
268 if (m_bodyStreamBuffer) {
269 ASSERT(m_bodyStreamBuffer->exception());
270 return m_bodyStreamBuffer->exception();
271 }
272 return DOMException::create(NetworkError, "network error");
273 }
274
275 // Set when the data container of the Body is a BodyStreamBuffer.
276 Member<BodyStreamBuffer> m_bodyStreamBuffer;
277 // Set when the data container of the Body is a BlobDataHandle.
278 RefPtr<BlobDataHandle> m_blobDataHandle;
279 // Used to read the data from BlobDataHandle.
280 OwnPtr<FileReaderLoader> m_loader;
281 // Created when createDrainingStream is called to drain the data.
282 Member<BodyStreamBuffer> m_drainingStreamBuffer;
283 Member<ReadableByteStream> m_stream; 205 Member<ReadableByteStream> m_stream;
284 State m_state; 206 State m_state;
207 bool m_streamNeedsMore;
208 #if ENABLE(ASSERT)
209 bool m_drained;
210 #endif
285 }; 211 };
286 212
287 ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type) 213 ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type)
288 { 214 {
289 if (bodyUsed()) 215 if (bodyUsed())
290 return ScriptPromise::reject(scriptState, V8ThrowException::createTypeEr ror(scriptState->isolate(), "Already read")); 216 return ScriptPromise::reject(scriptState, V8ThrowException::createTypeEr ror(scriptState->isolate(), "Already read"));
291 217
292 // When the main thread sends a V8::TerminateExecution() signal to a worker 218 // When the main thread sends a V8::TerminateExecution() signal to a worker
293 // thread, any V8 API on the worker thread starts returning an empty 219 // thread, any V8 API on the worker thread starts returning an empty
294 // handle. This can happen in Body::readAsync. To avoid the situation, we 220 // handle. This can happen in Body::readAsync. To avoid the situation, we
295 // first check the ExecutionContext and return immediately if it's already 221 // first check the ExecutionContext and return immediately if it's already
296 // gone (which means that the V8::TerminateExecution() signal has been sent 222 // gone (which means that the V8::TerminateExecution() signal has been sent
297 // to this worker thread). 223 // to this worker thread).
298 ExecutionContext* executionContext = scriptState->executionContext(); 224 ExecutionContext* executionContext = scriptState->executionContext();
299 if (!executionContext) 225 if (!executionContext)
300 return ScriptPromise(); 226 return ScriptPromise();
301 227
302 lockBody(); 228 lockBody();
303 m_responseType = type; 229 m_responseType = type;
304 230
305 ASSERT(!m_resolver); 231 ASSERT(!m_resolver);
306 m_resolver = ScriptPromiseResolver::create(scriptState); 232 m_resolver = ScriptPromiseResolver::create(scriptState);
307 ScriptPromise promise = m_resolver->promise(); 233 ScriptPromise promise = m_resolver->promise();
308 234
309 if (m_stream->stateInternal() == ReadableStream::Closed) { 235 if (m_stream->stateInternal() == ReadableStream::Closed) {
310 // We resolve the resolver manually in order not to use member 236 resolveWithEmptyDataSynchronously();
311 // variables.
312 switch (m_responseType) {
313 case ResponseAsArrayBuffer:
314 m_resolver->resolve(DOMArrayBuffer::create(nullptr, 0));
315 break;
316 case ResponseAsBlob: {
317 OwnPtr<BlobData> blobData = BlobData::create();
318 blobData->setContentType(mimeType());
319 m_resolver->resolve(Blob::create(BlobDataHandle::create(blobData.rel ease(), 0)));
320 break;
321 }
322 case ResponseAsText:
323 m_resolver->resolve(String());
324 break;
325 case ResponseAsFormData:
326 // TODO(yhirano): Implement this.
327 ASSERT_NOT_REACHED();
328 break;
329 case ResponseAsJSON: {
330 ScriptState::Scope scope(m_resolver->scriptState());
331 m_resolver->reject(V8ThrowException::createSyntaxError(m_resolver->s criptState()->isolate(), "Unexpected end of input"));
332 break;
333 }
334 case ResponseUnknown:
335 ASSERT_NOT_REACHED();
336 break;
337 }
338 m_resolver.clear();
339 } else if (m_stream->stateInternal() == ReadableStream::Errored) { 237 } else if (m_stream->stateInternal() == ReadableStream::Errored) {
340 m_resolver->reject(m_stream->storedException()); 238 m_resolver->reject(m_stream->storedException());
341 m_resolver.clear(); 239 m_resolver.clear();
342 } else if (isBodyConsumed()) {
343 m_streamSource->createDrainingStream()->readAllAndCreateBlobHandle(mimeT ype(), new BlobHandleReceiver(this));
344 } else if (buffer()) {
345 buffer()->readAllAndCreateBlobHandle(mimeType(), new BlobHandleReceiver( this));
346 } else { 240 } else {
347 readAsyncFromBlob(blobDataHandle()); 241 readAsyncFromDrainingBodyStreamBuffer(createDrainingStream(), mimeType() );
348 } 242 }
349 return promise; 243 return promise;
350 } 244 }
351 245
352 void Body::readAsyncFromBlob(PassRefPtr<BlobDataHandle> handle) 246 void Body::resolveWithEmptyDataSynchronously()
353 { 247 {
354 FileReaderLoader::ReadType readType = FileReaderLoader::ReadAsText; 248 // We resolve the resolver manually in order not to use member
355 RefPtr<BlobDataHandle> blobHandle = handle; 249 // variables.
356 if (!blobHandle)
357 blobHandle = BlobDataHandle::create(BlobData::create(), 0);
358 switch (m_responseType) { 250 switch (m_responseType) {
359 case ResponseAsArrayBuffer: 251 case ResponseAsArrayBuffer:
360 readType = FileReaderLoader::ReadAsArrayBuffer; 252 m_resolver->resolve(DOMArrayBuffer::create(nullptr, 0));
361 break; 253 break;
254 case ResponseAsBlob: {
255 OwnPtr<BlobData> blobData = BlobData::create();
256 blobData->setContentType(mimeType());
257 m_resolver->resolve(Blob::create(BlobDataHandle::create(blobData.release (), 0)));
258 break;
259 }
260 case ResponseAsText:
261 m_resolver->resolve(String());
262 break;
263 case ResponseAsFormData:
264 // TODO(yhirano): Implement this.
265 ASSERT_NOT_REACHED();
266 break;
267 case ResponseAsJSON: {
268 ScriptState::Scope scope(m_resolver->scriptState());
269 m_resolver->reject(V8ThrowException::createSyntaxError(m_resolver->scrip tState()->isolate(), "Unexpected end of input"));
270 break;
271 }
272 case ResponseUnknown:
273 ASSERT_NOT_REACHED();
274 break;
275 }
276 m_resolver.clear();
277 }
278
279 void Body::readAsyncFromDrainingBodyStreamBuffer(PassOwnPtr<DrainingBodyStreamBu ffer> buffer, const String& mimeType)
280 {
281 if (!buffer) {
282 resolveWithEmptyDataSynchronously();
283 m_streamSource->close();
284 return;
285 }
286
287 FetchDataLoader* fetchDataLoader = nullptr;
288
289 switch (m_responseType) {
290 case ResponseAsArrayBuffer:
291 fetchDataLoader = FetchDataLoader::createLoaderAsArrayBuffer();
292 break;
293
294 case ResponseAsJSON:
295 case ResponseAsText:
296 fetchDataLoader = FetchDataLoader::createLoaderAsString();
297 break;
298
362 case ResponseAsBlob: 299 case ResponseAsBlob:
363 if (blobHandle->size() != kuint64max) { 300 fetchDataLoader = FetchDataLoader::createLoaderAsBlobHandle(mimeType);
364 // If the size of |blobHandle| is set correctly, creates Blob from
365 // it.
366 if (blobHandle->type() != mimeType()) {
367 // A new BlobDataHandle is created to override the Blob's type.
368 m_resolver->resolve(Blob::create(BlobDataHandle::create(blobHand le->uuid(), mimeType(), blobHandle->size())));
369 } else {
370 m_resolver->resolve(Blob::create(blobHandle));
371 }
372 m_stream->close();
373 m_resolver.clear();
374 return;
375 }
376 // If the size is not set, read as ArrayBuffer and create a new blob to
377 // get the size.
378 // FIXME: This workaround is not good for performance.
379 // When we will stop using Blob as a base system of Body to support
380 // stream, this problem should be solved.
381 readType = FileReaderLoader::ReadAsArrayBuffer;
382 break; 301 break;
302
383 case ResponseAsFormData: 303 case ResponseAsFormData:
384 // FIXME: Implement this. 304 // FIXME: Implement this.
385 ASSERT_NOT_REACHED(); 305 ASSERT_NOT_REACHED();
386 break; 306 return;
387 case ResponseAsJSON: 307
388 case ResponseAsText:
389 break;
390 default: 308 default:
391 ASSERT_NOT_REACHED(); 309 ASSERT_NOT_REACHED();
310 return;
392 } 311 }
393 312
394 m_loader = adoptPtr(new FileReaderLoader(readType, this)); 313 buffer->startLoading(fetchDataLoader, this);
395 m_loader->start(m_resolver->scriptState()->executionContext(), blobHandle);
396
397 return;
398 } 314 }
399 315
400 ScriptPromise Body::arrayBuffer(ScriptState* scriptState) 316 ScriptPromise Body::arrayBuffer(ScriptState* scriptState)
401 { 317 {
402 return readAsync(scriptState, ResponseAsArrayBuffer); 318 return readAsync(scriptState, ResponseAsArrayBuffer);
403 } 319 }
404 320
405 ScriptPromise Body::blob(ScriptState* scriptState) 321 ScriptPromise Body::blob(ScriptState* scriptState)
406 { 322 {
407 return readAsync(scriptState, ResponseAsBlob); 323 return readAsync(scriptState, ResponseAsBlob);
(...skipping 29 matching lines...) Expand all
437 { 353 {
438 ASSERT(!bodyUsed()); 354 ASSERT(!bodyUsed());
439 if (option == PassBody) 355 if (option == PassBody)
440 m_bodyUsed = true; 356 m_bodyUsed = true;
441 ASSERT(!m_stream->isLocked()); 357 ASSERT(!m_stream->isLocked());
442 TrackExceptionState exceptionState; 358 TrackExceptionState exceptionState;
443 m_stream->getBytesReader(executionContext(), exceptionState); 359 m_stream->getBytesReader(executionContext(), exceptionState);
444 ASSERT(!exceptionState.hadException()); 360 ASSERT(!exceptionState.hadException());
445 } 361 }
446 362
447 bool Body::isBodyConsumed() const 363 void Body::setBody(BodyStreamBuffer* buffer)
448 { 364 {
449 if (m_streamSource->state() != m_streamSource->Initial) { 365 m_streamSource = new ReadableStreamSource(executionContext(), buffer);
450 // Some data is pulled from the source.
451 return true;
452 }
453 if (m_stream->stateInternal() == ReadableStream::Closed) {
454 // Return true if the blob handle is originally not empty.
455 RefPtr<BlobDataHandle> handle = blobDataHandle();
456 return handle && handle->size();
457 }
458 if (m_stream->stateInternal() == ReadableStream::Errored) {
459 // The stream is errored. That means an effort to read data was made.
460 return true;
461 }
462 return false;
463 }
464
465 void Body::setBody(ReadableStreamSource* source)
466 {
467 m_streamSource = source;
468 m_stream = new ReadableByteStream(m_streamSource, new ReadableByteStream::St rictStrategy); 366 m_stream = new ReadableByteStream(m_streamSource, new ReadableByteStream::St rictStrategy);
469 m_streamSource->startStream(m_stream); 367 m_streamSource->startStream(m_stream);
470 } 368 }
471 369
472 BodyStreamBuffer* Body::createDrainingStream() 370 PassOwnPtr<DrainingBodyStreamBuffer> Body::createDrainingStream()
473 { 371 {
474 return m_streamSource->createDrainingStream(); 372 return m_streamSource->createDrainingStream();
475 } 373 }
476 374
477 void Body::stop() 375 void Body::stop()
yhirano 2015/07/03 04:42:28 If you do nothing you can delete this function.
hiroshige 2015/07/05 07:30:24 Done.
478 { 376 {
479 // Canceling the load will call didFail which will remove the resolver.
480 if (m_loader)
481 m_loader->cancel();
482 } 377 }
483 378
484 bool Body::hasPendingActivity() const 379 bool Body::hasPendingActivity() const
485 { 380 {
486 if (executionContext()->activeDOMObjectsAreStopped()) 381 if (executionContext()->activeDOMObjectsAreStopped())
487 return false; 382 return false;
488 if (m_resolver) 383 if (m_resolver)
489 return true; 384 return true;
490 if (m_stream->isLocked()) 385 if (m_stream->isLocked())
491 return true; 386 return true;
492 return false; 387 return false;
493 } 388 }
494 389
495 Body::ReadableStreamSource* Body::createBodySource(PassRefPtr<BlobDataHandle> ha ndle)
496 {
497 return new ReadableStreamSource(executionContext(), handle);
498 }
499
500 Body::ReadableStreamSource* Body::createBodySource(BodyStreamBuffer* buffer)
501 {
502 return new ReadableStreamSource(executionContext(), buffer);
503 }
504
505 DEFINE_TRACE(Body) 390 DEFINE_TRACE(Body)
506 { 391 {
507 visitor->trace(m_resolver); 392 visitor->trace(m_resolver);
508 visitor->trace(m_stream); 393 visitor->trace(m_stream);
509 visitor->trace(m_streamSource); 394 visitor->trace(m_streamSource);
510 ActiveDOMObject::trace(visitor); 395 ActiveDOMObject::trace(visitor);
396 FetchDataLoader::Client::trace(visitor);
511 } 397 }
512 398
513 Body::Body(ExecutionContext* context) 399 Body::Body(ExecutionContext* context)
514 : ActiveDOMObject(context) 400 : ActiveDOMObject(context)
515 , m_bodyUsed(false) 401 , m_bodyUsed(false)
516 , m_responseType(ResponseType::ResponseUnknown) 402 , m_responseType(ResponseType::ResponseUnknown)
517 , m_streamSource(new ReadableStreamSource(context)) 403 , m_streamSource(new ReadableStreamSource(context, nullptr))
518 , m_stream(new ReadableByteStream(m_streamSource, new ReadableByteStream::St rictStrategy)) 404 , m_stream(new ReadableByteStream(m_streamSource, new ReadableByteStream::St rictStrategy))
519 { 405 {
520 m_streamSource->startStream(m_stream); 406 m_streamSource->startStream(m_stream);
521 } 407 }
522 408
523 void Body::resolveJSON(const String& string) 409 void Body::resolveJSON(const String& string)
524 { 410 {
525 ASSERT(m_responseType == ResponseAsJSON); 411 ASSERT(m_responseType == ResponseAsJSON);
526 ScriptState::Scope scope(m_resolver->scriptState()); 412 ScriptState::Scope scope(m_resolver->scriptState());
527 v8::Isolate* isolate = m_resolver->scriptState()->isolate(); 413 v8::Isolate* isolate = m_resolver->scriptState()->isolate();
528 v8::Local<v8::String> inputString = v8String(isolate, string); 414 v8::Local<v8::String> inputString = v8String(isolate, string);
529 v8::TryCatch trycatch; 415 v8::TryCatch trycatch;
530 v8::Local<v8::Value> parsed; 416 v8::Local<v8::Value> parsed;
531 if (v8Call(v8::JSON::Parse(isolate, inputString), parsed, trycatch)) 417 if (v8Call(v8::JSON::Parse(isolate, inputString), parsed, trycatch))
532 m_resolver->resolve(parsed); 418 m_resolver->resolve(parsed);
533 else 419 else
534 m_resolver->reject(trycatch.Exception()); 420 m_resolver->reject(trycatch.Exception());
535 } 421 }
536 422
537 // FileReaderLoaderClient functions. 423 // FetchDataLoader::Client functions.
538 void Body::didStartLoading() { } 424 void Body::didFetchDataLoadFailed()
539 void Body::didReceiveData() { } 425 {
540 void Body::didFinishLoading() 426 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
427 return;
428
429 if (m_resolver) {
430 if (!m_resolver->executionContext() || m_resolver->executionContext()->a ctiveDOMObjectsAreStopped()) {
431 m_resolver.clear();
432 return;
433 }
434 ScriptState* state = m_resolver->scriptState();
435 ScriptState::Scope scope(state);
436 m_resolver->reject(V8ThrowException::createTypeError(state->isolate(), " Failed to fetch"));
437 m_resolver.clear();
438 }
439 }
440
441 void Body::didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> blobDataHandl e)
442 {
443 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
444 return;
445
446 ASSERT(m_responseType == ResponseAsBlob);
447 m_resolver->resolve(Blob::create(blobDataHandle));
448 m_resolver.clear();
449 }
450
451 void Body::didFetchDataLoadedArrayBuffer(PassRefPtr<DOMArrayBuffer> arrayBuffer)
452 {
453 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
454 return;
455
456 ASSERT(m_responseType == ResponseAsArrayBuffer);
457 m_resolver->resolve(arrayBuffer);
458 m_resolver.clear();
459 }
460
461 void Body::didFetchDataLoadedString(const String& str)
541 { 462 {
542 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped()) 463 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
543 return; 464 return;
544 465
545 switch (m_responseType) { 466 switch (m_responseType) {
546 case ResponseAsArrayBuffer:
547 m_resolver->resolve(m_loader->arrayBufferResult());
548 break;
549 case ResponseAsBlob: {
550 ASSERT(blobDataHandle()->size() == kuint64max);
551 OwnPtr<BlobData> blobData = BlobData::create();
552 RefPtr<DOMArrayBuffer> buffer = m_loader->arrayBufferResult();
553 blobData->appendBytes(buffer->data(), buffer->byteLength());
554 blobData->setContentType(mimeType());
555 const size_t length = blobData->length();
556 m_resolver->resolve(Blob::create(BlobDataHandle::create(blobData.release (), length)));
557 break;
558 }
559 case ResponseAsFormData:
560 ASSERT_NOT_REACHED();
561 break;
562 case ResponseAsJSON: 467 case ResponseAsJSON:
563 resolveJSON(m_loader->stringResult()); 468 resolveJSON(str);
564 break; 469 break;
565 case ResponseAsText: 470 case ResponseAsText:
566 m_resolver->resolve(m_loader->stringResult()); 471 m_resolver->resolve(str);
567 break; 472 break;
568 default: 473 default:
569 ASSERT_NOT_REACHED(); 474 ASSERT_NOT_REACHED();
570 } 475 }
571 m_streamSource->close(); 476
572 m_resolver.clear(); 477 m_resolver.clear();
573 } 478 }
574 479
575 void Body::didFail(FileError::ErrorCode code)
576 {
577 if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
578 return;
579
580 m_streamSource->error();
581 if (m_resolver) {
582 // FIXME: We should reject the promise.
583 m_resolver->resolve("");
584 m_resolver.clear();
585 }
586 }
587
588 void Body::didBlobHandleReceiveError(DOMException* exception)
589 {
590 if (!m_resolver)
591 return;
592 m_streamSource->error();
593 m_resolver->reject(exception);
594 m_resolver.clear();
595 }
596
597 } // namespace blink 480 } // namespace blink
OLDNEW
« no previous file with comments | « Source/modules/fetch/Body.h ('k') | Source/modules/fetch/BodyStreamBuffer.h » ('j') | Source/modules/fetch/BodyStreamBuffer.h » ('J')

Powered by Google App Engine
This is Rietveld 408576698