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

Issues Search
    My Issues | Starred     Open | Closed | All

Unified Diff: Source/modules/fetch/Body.cpp

Issue 1171913003: **** [WIP] Blink-side: Implement FetchBlobDataConsumerHandle **** (Closed) Base URL: https://chromium.googlesource.com/chromium/blink.git@master
Patch Set: Clean up. Created 5 years, 6 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 side-by-side diff with in-line comments
Download patch
« no previous file with comments | « Source/modules/fetch/Body.h ('k') | Source/modules/fetch/BodyStreamBuffer.h » ('j') | no next file with comments »
Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
Index: Source/modules/fetch/Body.cpp
diff --git a/Source/modules/fetch/Body.cpp b/Source/modules/fetch/Body.cpp
index 0be313e412339d2df44a9ff3303435e9b7786011..74a3fa718de147bd806bf5a8cc2b25a12b76981a 100644
--- a/Source/modules/fetch/Body.cpp
+++ b/Source/modules/fetch/Body.cpp
@@ -12,47 +12,21 @@
#include "bindings/core/v8/V8ThrowException.h"
#include "core/dom/DOMArrayBuffer.h"
#include "core/dom/DOMTypedArray.h"
+#include "core/dom/ExceptionCode.h"
#include "core/fileapi/Blob.h"
-#include "core/fileapi/FileReaderLoader.h"
-#include "core/fileapi/FileReaderLoaderClient.h"
#include "core/frame/UseCounter.h"
#include "core/streams/ReadableByteStream.h"
#include "core/streams/ReadableByteStreamReader.h"
#include "core/streams/UnderlyingSource.h"
#include "modules/fetch/BodyStreamBuffer.h"
+#include "modules/fetch/DataConsumerHandleUtil.h"
+#include "modules/fetch/FetchBlobDataConsumerHandle.h"
namespace blink {
-class Body::BlobHandleReceiver final : public BodyStreamBuffer::BlobHandleCreatorClient {
-public:
- explicit BlobHandleReceiver(Body* body)
- : m_body(body)
- {
- }
- void didCreateBlobHandle(PassRefPtr<BlobDataHandle> handle) override
- {
- ASSERT(m_body);
- m_body->readAsyncFromBlob(handle);
- m_body = nullptr;
- }
- void didFail(DOMException* exception) override
- {
- ASSERT(m_body);
- m_body->didBlobHandleReceiveError(exception);
- m_body = nullptr;
- }
- DEFINE_INLINE_VIRTUAL_TRACE()
- {
- BodyStreamBuffer::BlobHandleCreatorClient::trace(visitor);
- visitor->trace(m_body);
- }
-private:
- Member<Body> m_body;
-};
-
// This class is an ActiveDOMObject subclass only for holding the
// ExecutionContext used in |pullSource|.
-class Body::ReadableStreamSource : public BodyStreamBuffer::Observer, public UnderlyingSource, public FileReaderLoaderClient, public ActiveDOMObject {
+class Body::ReadableStreamSource : public GarbageCollectedFinalized<Body::ReadableStreamSource>, public UnderlyingSource, public WebDataConsumerHandle::Client, public ActiveDOMObject {
USING_GARBAGE_COLLECTED_MIXIN(ReadableStreamSource);
public:
enum State {
@@ -61,26 +35,21 @@ public:
Closed,
Errored,
};
- ReadableStreamSource(ExecutionContext* executionContext, PassRefPtr<BlobDataHandle> handle)
- : ActiveDOMObject(executionContext)
- , m_blobDataHandle(handle ? handle : BlobDataHandle::create(BlobData::create(), 0))
- , m_state(Initial)
- {
- suspendIfNeeded();
- }
- ReadableStreamSource(ExecutionContext* executionContext, BodyStreamBuffer* buffer)
+ ReadableStreamSource(ExecutionContext* executionContext, BodyStreamBuffer2* buffer)
: ActiveDOMObject(executionContext)
- , m_bodyStreamBuffer(buffer)
+ , m_bodyStreamBuffer(buffer ? buffer : BodyStreamBuffer2::create(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle())))
, m_state(Initial)
+ , m_streamNeedsMore(false)
{
suspendIfNeeded();
}
explicit ReadableStreamSource(ExecutionContext* executionContext)
: ActiveDOMObject(executionContext)
- , m_blobDataHandle(BlobDataHandle::create(BlobData::create(), 0))
+ , m_bodyStreamBuffer(BodyStreamBuffer2::create(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle())))
, m_state(Initial)
+ , m_streamNeedsMore(false)
{
suspendIfNeeded();
}
@@ -95,41 +64,30 @@ public:
stream->didSourceStart();
}
// Creates a new BodyStreamBuffer to drain the data.
- BodyStreamBuffer* createDrainingStream()
+ BodyStreamBuffer2* createDrainingStream()
{
ASSERT(m_state != Initial);
- auto drainingStreamBuffer = new BodyStreamBuffer(new Canceller(this));
if (m_stream->stateInternal() == ReadableByteStream::Closed) {
- drainingStreamBuffer->close();
- return drainingStreamBuffer;
+ return BodyStreamBuffer2::create(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle()));
}
if (m_stream->stateInternal() == ReadableByteStream::Errored) {
- drainingStreamBuffer->error(exception());
- return drainingStreamBuffer;
+ return BodyStreamBuffer2::create(createFetchDataConsumerHandleFromWebHandle(createUnexpectedErrorDataConsumerHandle()));
}
- ASSERT(!m_drainingStreamBuffer);
- // Take back the data in |m_stream|.
- Deque<std::pair<RefPtr<DOMArrayBufferView>, size_t>> tmp_queue;
- ASSERT(m_stream->stateInternal() == ReadableStream::Readable);
- m_stream->readInternal(tmp_queue);
- while (!tmp_queue.isEmpty()) {
- std::pair<RefPtr<DOMArrayBufferView>, size_t> data = tmp_queue.takeFirst();
- drainingStreamBuffer->write(data.first->buffer());
- }
+ ASSERT(!m_drained);
+ m_drained = true;
+ m_reader.clear();
if (m_state == Closed)
- drainingStreamBuffer->close();
+ return BodyStreamBuffer2::create(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle()));
- m_drainingStreamBuffer = drainingStreamBuffer;
- return m_drainingStreamBuffer;
+ return m_bodyStreamBuffer;
}
+
DEFINE_INLINE_VIRTUAL_TRACE()
{
visitor->trace(m_bodyStreamBuffer);
- visitor->trace(m_drainingStreamBuffer);
visitor->trace(m_stream);
- BodyStreamBuffer::Observer::trace(visitor);
UnderlyingSource::trace(visitor);
ActiveDOMObject::trace(visitor);
}
@@ -143,145 +101,87 @@ public:
return;
}
m_state = Closed;
- if (m_drainingStreamBuffer)
- m_drainingStreamBuffer->close();
+ m_reader.clear();
m_stream->close();
}
void error()
{
m_state = Errored;
- if (m_drainingStreamBuffer)
- m_drainingStreamBuffer->error(exception());
- m_stream->error(exception());
+ m_reader.clear();
+ m_stream->error(DOMException::create(NetworkError, "network error"));
}
private:
- class Canceller : public BodyStreamBuffer::Canceller {
- public:
- Canceller(ReadableStreamSource* source) : m_source(source) { }
- void cancel() override
- {
- m_source->cancel();
- }
-
- DEFINE_INLINE_VIRTUAL_TRACE()
- {
- visitor->trace(m_source);
- BodyStreamBuffer::Canceller::trace(visitor);
- }
-
- private:
- Member<ReadableStreamSource> m_source;
- };
+ void didGetReadable() override
+ {
+ processData();
+ }
// UnderlyingSource functions.
void pullSource() override
{
- // Note that one |pull| is called only when |read| is called on the
- // associated ReadableByteStreamReader because we create a stream with
- // StrictStrategy.
+ m_streamNeedsMore = true;
+
if (m_state == Initial) {
m_state = Streaming;
- if (m_bodyStreamBuffer) {
- m_bodyStreamBuffer->registerObserver(this);
- onWrite();
- if (m_bodyStreamBuffer->hasError())
- return onError();
- if (m_bodyStreamBuffer->isClosed())
- return onClose();
- } else {
- FileReaderLoader::ReadType readType = FileReaderLoader::ReadAsArrayBuffer;
- m_loader = adoptPtr(new FileReaderLoader(readType, this));
- m_loader->start(executionContext(), m_blobDataHandle);
- }
+ ASSERT(!m_drained);
+ ASSERT(!m_reader);
+ ASSERT(m_bodyStreamBuffer);
+ ASSERT(m_bodyStreamBuffer->handle());
+ m_reader = m_bodyStreamBuffer->handle()->obtainReader(this);
}
- }
- ScriptPromise cancelSource(ScriptState* scriptState, ScriptValue reason) override
- {
- cancel();
- return ScriptPromise::cast(scriptState, v8::Undefined(scriptState->isolate()));
- }
+ if (m_drained)
+ return;
- // BodyStreamBuffer::Observer functions.
- void onWrite() override
- {
- ASSERT(m_state == Streaming);
- while (RefPtr<DOMArrayBuffer> buf = m_bodyStreamBuffer->read()) {
- write(buf);
- }
- }
- void onClose() override
- {
- ASSERT(m_state == Streaming);
- close();
- m_bodyStreamBuffer->unregisterObserver();
- }
- void onError() override
- {
- ASSERT(m_state == Streaming);
- error();
- m_bodyStreamBuffer->unregisterObserver();
+ processData();
}
- // FileReaderLoaderClient functions.
- void didStartLoading() override { }
- void didReceiveData() override { }
- void didFinishLoading() override
+ ScriptPromise cancelSource(ScriptState* scriptState, ScriptValue reason) override
{
- ASSERT(m_state == Streaming);
- write(m_loader->arrayBufferResult());
close();
- }
- void didFail(FileError::ErrorCode) override
- {
- ASSERT(m_state == Streaming);
- error();
+ return ScriptPromise::cast(scriptState, v8::Undefined(scriptState->isolate()));
}
- void write(PassRefPtr<DOMArrayBuffer> buf)
+ // Reads data and writes the data to |m_stream|, as long as data are
+ // available and the stream has pending reads.
+ void processData()
{
- if (m_drainingStreamBuffer) {
- m_drainingStreamBuffer->write(buf);
- } else {
- auto size = buf->byteLength();
- m_stream->enqueue(DOMUint8Array::create(buf, 0, size));
- }
- }
- void cancel()
- {
- if (m_bodyStreamBuffer) {
- m_bodyStreamBuffer->cancel();
- // We should not close the stream here, because it is canceller's
- // responsibility.
- } else {
- if (m_loader)
- m_loader->cancel();
- close();
+ ASSERT(m_reader);
+ while (m_streamNeedsMore) {
+ const void* buffer;
+ size_t available;
+ WebDataConsumerHandle::Result result = m_reader->beginRead(&buffer, WebDataConsumerHandle::FlagNone, &available);
+ switch (result) {
+ case WebDataConsumerHandle::Ok:
+ m_streamNeedsMore = m_stream->enqueue(DOMUint8Array::create(static_cast<const unsigned char*>(buffer), available));
+ m_reader->endRead(available);
+ break;
+
+ case WebDataConsumerHandle::Done:
+ close();
+ return;
+
+ case WebDataConsumerHandle::ShouldWait:
+ return;
+
+ case WebDataConsumerHandle::Busy:
+ case WebDataConsumerHandle::ResourceExhausted:
+ case WebDataConsumerHandle::UnexpectedError:
+ error();
+ return;
+ }
}
}
- DOMException* exception()
- {
- if (m_state != Errored)
- return nullptr;
- if (m_bodyStreamBuffer) {
- ASSERT(m_bodyStreamBuffer->exception());
- return m_bodyStreamBuffer->exception();
- }
- return DOMException::create(NetworkError, "network error");
- }
+ // Source of data.
+ Member<BodyStreamBuffer2> m_bodyStreamBuffer;
+ OwnPtr<FetchDataConsumerHandle::Reader> m_reader;
- // Set when the data container of the Body is a BodyStreamBuffer.
- Member<BodyStreamBuffer> m_bodyStreamBuffer;
- // Set when the data container of the Body is a BlobDataHandle.
- RefPtr<BlobDataHandle> m_blobDataHandle;
- // Used to read the data from BlobDataHandle.
- OwnPtr<FileReaderLoader> m_loader;
- // Created when createDrainingStream is called to drain the data.
- Member<BodyStreamBuffer> m_drainingStreamBuffer;
Member<ReadableByteStream> m_stream;
State m_state;
+ bool m_drained;
+ bool m_streamNeedsMore;
};
ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type)
@@ -340,63 +240,51 @@ ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type)
m_resolver->reject(m_stream->storedException());
m_resolver.clear();
} else if (isBodyConsumed()) {
- m_streamSource->createDrainingStream()->readAllAndCreateBlobHandle(mimeType(), new BlobHandleReceiver(this));
+ readAsyncFromFetchDataConsumerHandle(m_streamSource->createDrainingStream()->handle(), mimeType());
} else if (buffer()) {
- buffer()->readAllAndCreateBlobHandle(mimeType(), new BlobHandleReceiver(this));
+ readAsyncFromFetchDataConsumerHandle(buffer()->handle(), mimeType());
} else {
- readAsyncFromBlob(blobDataHandle());
+ readAsyncFromFetchDataConsumerHandle(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle()).get(), mimeType());
}
return promise;
}
-void Body::readAsyncFromBlob(PassRefPtr<BlobDataHandle> handle)
+void Body::readAsyncFromFetchDataConsumerHandle(FetchDataConsumerHandle* handle, const String& mimeType)
{
- FileReaderLoader::ReadType readType = FileReaderLoader::ReadAsText;
- RefPtr<BlobDataHandle> blobHandle = handle;
- if (!blobHandle)
- blobHandle = BlobDataHandle::create(BlobData::create(), 0);
+ ASSERT(!m_fetchDataLoader);
+
switch (m_responseType) {
case ResponseAsArrayBuffer:
- readType = FileReaderLoader::ReadAsArrayBuffer;
+ m_fetchDataLoader = FetchDataLoader::createLoaderAsArrayBuffer();
break;
+
+ case ResponseAsJSON:
+ case ResponseAsText:
+ m_fetchDataLoader = FetchDataLoader::createLoaderAsString();
+ break;
+
case ResponseAsBlob:
- if (blobHandle->size() != kuint64max) {
- // If the size of |blobHandle| is set correctly, creates Blob from
- // it.
- if (blobHandle->type() != mimeType()) {
- // A new BlobDataHandle is created to override the Blob's type.
- m_resolver->resolve(Blob::create(BlobDataHandle::create(blobHandle->uuid(), mimeType(), blobHandle->size())));
- } else {
- m_resolver->resolve(Blob::create(blobHandle));
- }
- m_stream->close();
- m_resolver.clear();
- return;
- }
- // If the size is not set, read as ArrayBuffer and create a new blob to
- // get the size.
- // FIXME: This workaround is not good for performance.
- // When we will stop using Blob as a base system of Body to support
- // stream, this problem should be solved.
- readType = FileReaderLoader::ReadAsArrayBuffer;
+ m_fetchDataLoader = FetchDataLoader::createLoaderAsBlobHandle(mimeType);
break;
+
case ResponseAsFormData:
// FIXME: Implement this.
ASSERT_NOT_REACHED();
- break;
- case ResponseAsJSON:
- case ResponseAsText:
- break;
+ return;
+
default:
ASSERT_NOT_REACHED();
+ return;
}
- m_loader = adoptPtr(new FileReaderLoader(readType, this));
- m_loader->start(m_resolver->scriptState()->executionContext(), blobHandle);
-
- return;
+ if (handle)
+ m_fetchDataLoader->start(handle, this);
+ else
+ m_fetchDataLoader->start(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle()).get(), this);
}
+
+
ScriptPromise Body::arrayBuffer(ScriptState* scriptState)
{
return readAsync(scriptState, ResponseAsArrayBuffer);
@@ -452,8 +340,8 @@ bool Body::isBodyConsumed() const
}
if (m_stream->stateInternal() == ReadableStream::Closed) {
// Return true if the blob handle is originally not empty.
- RefPtr<BlobDataHandle> handle = blobDataHandle();
- return handle && handle->size();
+ // TODO: delete isBodyConsumed() itself.
+ return true;
}
if (m_stream->stateInternal() == ReadableStream::Errored) {
// The stream is errored. That means an effort to read data was made.
@@ -469,16 +357,17 @@ void Body::setBody(ReadableStreamSource* source)
m_streamSource->startStream(m_stream);
}
-BodyStreamBuffer* Body::createDrainingStream()
+BodyStreamBuffer2* Body::createDrainingStream()
{
return m_streamSource->createDrainingStream();
}
void Body::stop()
{
- // Canceling the load will call didFail which will remove the resolver.
- if (m_loader)
- m_loader->cancel();
+ if (m_fetchDataLoader) {
+ m_fetchDataLoader->cancel();
+ m_fetchDataLoader.clear();
+ }
}
bool Body::hasPendingActivity() const
@@ -492,22 +381,19 @@ bool Body::hasPendingActivity() const
return false;
}
-Body::ReadableStreamSource* Body::createBodySource(PassRefPtr<BlobDataHandle> handle)
-{
- return new ReadableStreamSource(executionContext(), handle);
-}
-
-Body::ReadableStreamSource* Body::createBodySource(BodyStreamBuffer* buffer)
+Body::ReadableStreamSource* Body::createBodySource(BodyStreamBuffer2* buffer)
{
return new ReadableStreamSource(executionContext(), buffer);
}
DEFINE_TRACE(Body)
{
+ visitor->trace(m_fetchDataLoader);
visitor->trace(m_resolver);
visitor->trace(m_stream);
visitor->trace(m_streamSource);
ActiveDOMObject::trace(visitor);
+ FetchDataLoader::Client::trace(visitor);
}
Body::Body(ExecutionContext* context)
@@ -534,63 +420,70 @@ void Body::resolveJSON(const String& string)
m_resolver->reject(trycatch.Exception());
}
-// FileReaderLoaderClient functions.
-void Body::didStartLoading() { }
-void Body::didReceiveData() { }
-void Body::didFinishLoading()
+// FetchDataLoader::Client functions.
+void Body::didFetchDataLoadFailed()
{
+ ASSERT(m_fetchDataLoader);
+ m_fetchDataLoader.clear();
+
if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
return;
- switch (m_responseType) {
- case ResponseAsArrayBuffer:
- m_resolver->resolve(m_loader->arrayBufferResult());
- break;
- case ResponseAsBlob: {
- ASSERT(blobDataHandle()->size() == kuint64max);
- OwnPtr<BlobData> blobData = BlobData::create();
- RefPtr<DOMArrayBuffer> buffer = m_loader->arrayBufferResult();
- blobData->appendBytes(buffer->data(), buffer->byteLength());
- blobData->setContentType(mimeType());
- const size_t length = blobData->length();
- m_resolver->resolve(Blob::create(BlobDataHandle::create(blobData.release(), length)));
- break;
- }
- case ResponseAsFormData:
- ASSERT_NOT_REACHED();
- break;
- case ResponseAsJSON:
- resolveJSON(m_loader->stringResult());
- break;
- case ResponseAsText:
- m_resolver->resolve(m_loader->stringResult());
- break;
- default:
- ASSERT_NOT_REACHED();
+ m_streamSource->error();
+ if (m_resolver) {
+ m_resolver->reject("");
+ m_resolver.clear();
}
+}
+
+void Body::didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> blobDataHandle)
+{
+ ASSERT(m_fetchDataLoader);
+ m_fetchDataLoader.clear();
+
+ if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
+ return;
+
+ ASSERT(m_responseType == ResponseAsBlob);
+ m_resolver->resolve(Blob::create(blobDataHandle));
m_streamSource->close();
m_resolver.clear();
}
-void Body::didFail(FileError::ErrorCode code)
+void Body::didFetchDataLoadedArrayBuffer(PassRefPtr<DOMArrayBuffer> arrayBuffer)
{
+ ASSERT(m_fetchDataLoader);
+ m_fetchDataLoader.clear();
+
if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
return;
- m_streamSource->error();
- if (m_resolver) {
- // FIXME: We should reject the promise.
- m_resolver->resolve("");
- m_resolver.clear();
- }
+ ASSERT(m_responseType == ResponseAsArrayBuffer);
+ m_resolver->resolve(arrayBuffer);
+ m_streamSource->close();
+ m_resolver.clear();
}
-void Body::didBlobHandleReceiveError(DOMException* exception)
+void Body::didFetchDataLoadedString(const String& str)
{
- if (!m_resolver)
+ ASSERT(m_fetchDataLoader);
+ m_fetchDataLoader.clear();
+
+ if (!executionContext() || executionContext()->activeDOMObjectsAreStopped())
return;
- m_streamSource->error();
- m_resolver->reject(exception);
+
+ switch (m_responseType) {
+ case ResponseAsJSON:
+ resolveJSON(str);
+ break;
+ case ResponseAsText:
+ m_resolver->resolve(str);
+ break;
+ default:
+ ASSERT_NOT_REACHED();
+ }
+
+ m_streamSource->close();
m_resolver.clear();
}
« no previous file with comments | « Source/modules/fetch/Body.h ('k') | Source/modules/fetch/BodyStreamBuffer.h » ('j') | no next file with comments »

Powered by Google App Engine
This is Rietveld 408576698