Change BodyStreamBuffer to be FetchDataConsumerHandle-based and enable backpressure in Fetch API

Source/modules/fetch/BodyStreamBuffer.cpp

Index: Source/modules/fetch/BodyStreamBuffer.cpp
diff --git a/Source/modules/fetch/BodyStreamBuffer.cpp b/Source/modules/fetch/BodyStreamBuffer.cpp
index 3c98e3ee5d588101a847150e7632d971969ca05e..0b02692e7ca0e261761286cf990a17104b1b7537 100644
--- a/Source/modules/fetch/BodyStreamBuffer.cpp
+++ b/Source/modules/fetch/BodyStreamBuffer.cpp
@@ -12,129 +12,6 @@ namespace blink {
 
 namespace {
 
-class BlobCreator final : public BodyStreamBuffer::Observer {
-public:
-    BlobCreator(BodyStreamBuffer* buffer, const String& contentType, BodyStreamBuffer::BlobHandleCreatorClient* client)
-        : m_buffer(buffer)
-        , m_client(client)
-        , m_blobData(BlobData::create())
-    {
-        m_blobData->setContentType(contentType);
-    }
-    ~BlobCreator() override { }
-    DEFINE_INLINE_VIRTUAL_TRACE()
-    {
-        visitor->trace(m_buffer);
-        visitor->trace(m_client);
-        BodyStreamBuffer::Observer::trace(visitor);
-    }
-    void onWrite() override
-    {
-        ASSERT(m_buffer);
-        while (RefPtr<DOMArrayBuffer> buf = m_buffer->read()) {
-            m_blobData->appendBytes(buf->data(), buf->byteLength());
-        }
-    }
-    void onClose() override
-    {
-        ASSERT(m_buffer);
-        const long long size = m_blobData->length();
-        m_client->didCreateBlobHandle(BlobDataHandle::create(m_blobData.release(), size));
-        cleanup();
-    }
-    void onError() override
-    {
-        ASSERT(m_buffer);
-        m_client->didFail(m_buffer->exception());
-        cleanup();
-    }
-    void start()
-    {
-        ASSERT(!m_buffer->isObserverRegistered());
-        m_buffer->registerObserver(this);
-        onWrite();
-        if (m_buffer->hasError()) {
-            return onError();
-        }
-        if (m_buffer->isClosed())
-            return onClose();
-    }
-    void cleanup()
-    {
-        m_buffer->unregisterObserver();
-        m_buffer.clear();
-        m_client.clear();
-        m_blobData.clear();
-    }
-private:
-    Member<BodyStreamBuffer> m_buffer;
-    Member<BodyStreamBuffer::BlobHandleCreatorClient> m_client;
-    OwnPtr<BlobData> m_blobData;
-};
-
-class StreamTeePump : public BodyStreamBuffer::Observer {
-public:
-    StreamTeePump(BodyStreamBuffer* inBuffer, BodyStreamBuffer* outBuffer1, BodyStreamBuffer* outBuffer2)
-        : m_inBuffer(inBuffer)
-        , m_outBuffer1(outBuffer1)
-        , m_outBuffer2(outBuffer2)
-    {
-    }
-    void onWrite() override
-    {
-        while (RefPtr<DOMArrayBuffer> buf = m_inBuffer->read()) {
-            if (!m_outBuffer1->isClosed() && !m_outBuffer1->hasError())
-                m_outBuffer1->write(buf);
-            if (!m_outBuffer2->isClosed() && !m_outBuffer2->hasError())
-                m_outBuffer2->write(buf);
-        }
-    }
-    void onClose() override
-    {
-        if (!m_outBuffer1->isClosed() && !m_outBuffer1->hasError())
-            m_outBuffer1->close();
-        if (!m_outBuffer2->isClosed() && !m_outBuffer2->hasError())
-            m_outBuffer2->close();
-        cleanup();
-    }
-    void onError() override
-    {
-        if (!m_outBuffer1->isClosed() && !m_outBuffer1->hasError())
-            m_outBuffer1->error(m_inBuffer->exception());
-        if (!m_outBuffer2->isClosed() && !m_outBuffer2->hasError())
-            m_outBuffer2->error(m_inBuffer->exception());
-        cleanup();
-    }
-    DEFINE_INLINE_VIRTUAL_TRACE()
-    {
-        BodyStreamBuffer::Observer::trace(visitor);
-        visitor->trace(m_inBuffer);
-        visitor->trace(m_outBuffer1);
-        visitor->trace(m_outBuffer2);
-    }
-    void start()
-    {
-        m_inBuffer->registerObserver(this);
-        onWrite();
-        if (m_inBuffer->hasError())
-            return onError();
-        if (m_inBuffer->isClosed())
-            return onClose();
-    }
-
-private:
-    void cleanup()
-    {
-        m_inBuffer->unregisterObserver();
-        m_inBuffer.clear();
-        m_outBuffer1.clear();
-        m_outBuffer2.clear();
-    }
-    Member<BodyStreamBuffer> m_inBuffer;
-    Member<BodyStreamBuffer> m_outBuffer1;
-    Member<BodyStreamBuffer> m_outBuffer2;
-};
-
 // WebDataConsumerHandleAdapter is used to migrate incrementally
 // from BodyStreamBuffer to FetchDataConsumerHandle and will be removed
 // after the migration.
@@ -264,24 +141,6 @@ void BodyStreamBuffer::error(DOMException* exception)
         m_observer->onError();
 }
 
-bool BodyStreamBuffer::readAllAndCreateBlobHandle(const String& contentType, BlobHandleCreatorClient* client)
-{
-    if (m_observer)
-        return false;
-    BlobCreator* blobCreator = new BlobCreator(this, contentType, client);
-    blobCreator->start();
-    return true;
-}
-
-bool BodyStreamBuffer::startTee(BodyStreamBuffer* out1, BodyStreamBuffer* out2)
-{
-    if (m_observer)
-        return false;
-    StreamTeePump* teePump = new StreamTeePump(this, out1, out2);
-    teePump->start();
-    return true;
-}
-
 bool BodyStreamBuffer::registerObserver(Observer* observer)
 {
     if (m_observer)