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

Source/modules/fetch/BodyStreamBuffer.h

Index: Source/modules/fetch/BodyStreamBuffer.h
diff --git a/Source/modules/fetch/BodyStreamBuffer.h b/Source/modules/fetch/BodyStreamBuffer.h
index cdb47baab116261c68167786d96f4c924f8e15d1..1e1fb23e4aa462786cf16215b434455af93bc72b 100644
--- a/Source/modules/fetch/BodyStreamBuffer.h
+++ b/Source/modules/fetch/BodyStreamBuffer.h
@@ -7,6 +7,9 @@
 
 #include "core/dom/DOMException.h"
 #include "modules/ModulesExport.h"
+#include "modules/fetch/DataConsumerHandleUtil.h"
+#include "modules/fetch/FetchDataConsumerHandle.h"
+#include "modules/fetch/FetchDataLoader.h"
 #include "platform/blob/BlobData.h"
 #include "platform/heap/Heap.h"
 #include "public/platform/WebDataConsumerHandle.h"
@@ -18,71 +21,227 @@ namespace blink {
 
 class DOMArrayBuffer;
 
-class MODULES_EXPORT BodyStreamBuffer final : public GarbageCollectedFinalized<BodyStreamBuffer> {
+/*
+// #define createDebugHandleAlways(handle) FetchDebugDataConsumerHandle::create(handle, __FILE__, __LINE__)
+#define createDebugHandleAlways(handle) (handle)
+#define createDebugHandle(handle) (handle)
+
+class FetchDebugDataConsumerHandle : public FetchDataConsumerHandle {
 public:
-    class Observer : public GarbageCollectedFinalized<Observer> {
+    static PassOwnPtr<FetchDataConsumerHandle> create(PassOwnPtr<FetchDataConsumerHandle> handle, const char* file, int line) { return adoptPtr(new FetchDebugDataConsumerHandle(handle, file, line)); }
+private:
+    FetchDebugDataConsumerHandle(PassOwnPtr<FetchDataConsumerHandle> handle, const char *file, int line)
+        : m_handle(handle)
+        , m_name(m_handle->debugName()) { }
+
+    class ReaderImpl final : public FetchDataConsumerHandle::Reader {
     public:
-        virtual ~Observer() { }
-        virtual void onWrite() = 0;
-        virtual void onClose() = 0;
-        virtual void onError() = 0;
-        DEFINE_INLINE_VIRTUAL_TRACE() { }
+        ReaderImpl(FetchDebugDataConsumerHandle* handle, PassOwnPtr<FetchDataConsumerHandle::Reader> reader) : m_handle(handle), m_name(handle->debugName()), m_reader(reader), m_isInTwoPhaseRead(false) { }
+        ~ReaderImpl()
+        {
+            ASSERT(!m_isInTwoPhaseRead);
+            print("dtor\n", (long long)currentThread(), this);
+        }
+        Result read(void* data, size_t size, Flags flags, size_t* readSize) override
+        {
+            ASSERT(!m_isInTwoPhaseRead);
+            print("read(%lld)\n", (long long)size);
+            Result result = m_reader->read(data, size, flags, readSize);
+            print("read(%lld) -> %d (%lld)\n", (long long)size, result, (long long)*readSize);
+            return result;
+        }
+
+        Result beginRead(const void** buffer, Flags flags, size_t* available) override
+        {
+            ASSERT(!m_isInTwoPhaseRead);
+            print("beginRead()\n");
+            Result result = m_reader->beginRead(buffer, flags, available);
+            print("beginRead() -> %d (%lld)\n", result, (long long)*available);
+            if (result == Ok)
+                m_isInTwoPhaseRead = true;
+            return result;
+        }
+        Result endRead(size_t readSize) override
+        {
+            ASSERT(m_isInTwoPhaseRead);
+            m_isInTwoPhaseRead = false;
+            print("endRead(%lld)\n", (long long)readSize);
+            Result result = m_reader->endRead(readSize);
+            print("endRead(%lld) -> %d\n", (long long)readSize, result);
+            return result;
+        }
+        PassRefPtr<BlobDataHandle> drainAsBlobDataHandle() override
+        {
+            print("drainAsBlobDataHandle()\n");
+            RefPtr<BlobDataHandle> blobDataHandle = m_reader->drainAsBlobDataHandle();
+            print("drainAsBlobDataHandle() -> %p\n", blobDataHandle.get());
+            return blobDataHandle.release();
+        }
+        void print(const char* format, ...)
+        {
+            fprintf(stderr, "[Thread:%lld] [Handle:%p (%s)] [Reader:%p] [TargetReader:%p]: ",
+                (long long)currentThread(),
+                m_handle,
+                m_name.c_str(),
+                this,
+                m_reader.get());
+
+            va_list args;
+            va_start(args, format);
+            vfprintf(stderr, format, args);
+            va_end(args);
+        }
+    private:
+        FetchDebugDataConsumerHandle* m_handle; // debugging only
+        std::string m_name;
+        OwnPtr<FetchDataConsumerHandle::Reader> m_reader;
+        bool m_isInTwoPhaseRead;
     };
-
-    class Canceller : public GarbageCollected<Canceller> {
+    class ClientWrapper : public Client {
     public:
-        virtual void cancel() = 0;
-        DEFINE_INLINE_VIRTUAL_TRACE() { }
+        ClientWrapper(FetchDebugDataConsumerHandle* handle, Client* client) : m_handle(handle), m_name(handle->debugName()), m_reader(nullptr), m_client(client) { }
+        void didGetReadable() override
+        {
+            print("didGetReadable\n");
+            if (m_client)
+                m_client->didGetReadable();
+            print("didGetReadable done\n");
+        }
+        void setReader(Reader* reader) { m_reader = reader; }
+
+        void print(const char* format, ...)
+        {
+            fprintf(stderr, "[Thread:%lld] [Handle:%p (%s)] [Reader:%p] [Client:%p] [TargetClient:%p]: ",
+                (long long)currentThread(),
+                m_handle,
+                m_name.c_str(),
+                m_reader,
+                this,
+                m_client);
+
+            va_list args;
+            va_start(args, format);
+            vfprintf(stderr, format, args);
+            va_end(args);
+        }
+    private:
+        FetchDebugDataConsumerHandle* m_handle; // debugging only
+        std::string m_name;
+        Reader* m_reader; // logging only
+
+        Client* m_client;
     };
+    Reader* obtainReaderInternal(Client* client) override
+    {
+        print("obtainReaderInternal(Client=%p)\n", client);
+        ClientWrapper* clientWrapper = new ClientWrapper(this, client);
+        Reader* reader = new ReaderImpl(this, m_handle->obtainReader(clientWrapper)); // FIXME: Leaking
+        clientWrapper->setReader(reader);
+        print("obtainReaderInternal(Client=%p) -> %p\n", client, reader);
+        return reader;
+    }
+    void print(const char* format, ...)
+    {
+        fprintf(stderr, "[Thread:%lld] [Handle:%p (%s)]: ",
+            (long long)currentThread(),
+            this,
+            debugName());
+
+        va_list args;
+        va_start(args, format);
+        vfprintf(stderr, format, args);
+        va_end(args);
+    }
+
+    const char* debugName() const override { return m_name.c_str(); }
+
+    OwnPtr<FetchDataConsumerHandle> m_handle;
+    std::string m_name;
+};
+*/
 
-    class BlobHandleCreatorClient : public GarbageCollectedFinalized<BlobHandleCreatorClient> {
+class MODULES_EXPORT BodyStreamBuffer final : public GarbageCollectedFinalized<BodyStreamBuffer> {
+public:
+    static BodyStreamBuffer* create(PassOwnPtr<FetchDataConsumerHandle> handle) { return new BodyStreamBuffer(handle); }
+    static BodyStreamBuffer* createEmpty() { return BodyStreamBuffer::create(createFetchDataConsumerHandleFromWebHandle(createDoneDataConsumerHandle())); }
+
+    FetchDataConsumerHandle* handle() const { return m_handle.get(); }
+    PassOwnPtr<FetchDataConsumerHandle> releaseHandle() { return m_handle.release(); }
+
+    void startLoading(FetchDataLoader* fetchDataLoader, FetchDataLoader::Client* client)
+    {
+        ASSERT(!m_fetchDataLoader);
+        m_fetchDataLoader = fetchDataLoader;
+        m_fetchDataLoader->start(handle(), new ClientWithFinishNotification(this, client));
+    }
+
+    void didFetchDataLoadFinished()
+    {
+        ASSERT(m_fetchDataLoader);
+        m_fetchDataLoader.clear();
+    }
+
+    DEFINE_INLINE_TRACE()
+    {
+        visitor->trace(m_fetchDataLoader);
+    }
+
+private:
+    explicit BodyStreamBuffer(PassOwnPtr<FetchDataConsumerHandle> handle) : m_handle(handle) { }
+
+    class ClientWithFinishNotification final : public GarbageCollectedFinalized<ClientWithFinishNotification>, public FetchDataLoader::Client {
+        USING_GARBAGE_COLLECTED_MIXIN(ClientWithFinishNotification);
     public:
-        virtual ~BlobHandleCreatorClient() { }
-        virtual void didCreateBlobHandle(PassRefPtr<BlobDataHandle>) = 0;
-        virtual void didFail(DOMException*) = 0;
-        DEFINE_INLINE_VIRTUAL_TRACE() { }
+        ClientWithFinishNotification(BodyStreamBuffer* buffer, FetchDataLoader::Client* client)
+            : m_buffer(buffer)
+            , m_client(client)
+        {
+        }
+
+        DEFINE_INLINE_VIRTUAL_TRACE()
+        {
+            visitor->trace(m_buffer);
+            visitor->trace(m_client);
+            FetchDataLoader::Client::trace(visitor);
+        }
+    private:
+        void didFetchDataLoadedBlobHandle(PassRefPtr<BlobDataHandle> blobDataHandle) override
+        {
+            m_buffer->didFetchDataLoadFinished();
+            if (m_client)
+                m_client->didFetchDataLoadedBlobHandle(blobDataHandle);
+        }
+        void didFetchDataLoadedArrayBuffer(PassRefPtr<DOMArrayBuffer> arrayBuffer) override
+        {
+            m_buffer->didFetchDataLoadFinished();
+            if (m_client)
+                m_client->didFetchDataLoadedArrayBuffer(arrayBuffer);
+        }
+        void didFetchDataLoadedString(const String& str) override
+        {
+            m_buffer->didFetchDataLoadFinished();
+            if (m_client)
+                m_client->didFetchDataLoadedString(str);
+        }
+        void didFetchDataLoadedStream() override
+        {
+            m_buffer->didFetchDataLoadFinished();
+            if (m_client)
+                m_client->didFetchDataLoadedStream();
+        }
+        void didFetchDataLoadFailed() override
+        {
+            m_buffer->didFetchDataLoadFinished();
+            if (m_client)
+                m_client->didFetchDataLoadFailed();
+        }
+
+        Member<BodyStreamBuffer> m_buffer;
+        Member<FetchDataLoader::Client> m_client;
     };
-    explicit BodyStreamBuffer(Canceller*);
-    ~BodyStreamBuffer() { }
-
-    PassRefPtr<DOMArrayBuffer> read();
-    bool isClosed() const { return m_isClosed; }
-    bool hasError() const { return m_exception; }
-    DOMException* exception() const { return m_exception; }
-
-    // Can't call after close() or error() was called.
-    void write(PassRefPtr<DOMArrayBuffer>);
-    // Can't call after close() or error() was called.
-    void close();
-    // Can't call after close() or error() was called.
-    void error(DOMException*);
-    void cancel() { m_canceller->cancel(); }
-
-    // This function registers an observer so it fails and returns false when an
-    // observer was already registered.
-    bool readAllAndCreateBlobHandle(const String& contentType, BlobHandleCreatorClient*);
-
-    // This function registers an observer so it fails and returns false when an
-    // observer was already registered.
-    bool startTee(BodyStreamBuffer* out1, BodyStreamBuffer* out2);
-
-    // When an observer was registered this function fails and returns false.
-    bool registerObserver(Observer*);
-    void unregisterObserver();
-    bool isObserverRegistered() const { return m_observer.get(); }
-    DECLARE_TRACE();
-
-    // Creates a BodyStreamBuffer from |handle| as the source.
-    // On failure, BodyStreamBuffer::error() is called with a NetworkError
-    // with |failureMessage|.
-    static BodyStreamBuffer* create(PassOwnPtr<WebDataConsumerHandle> /* handle */, const String& failureMessage);
 
-private:
-    Deque<RefPtr<DOMArrayBuffer>> m_queue;
-    bool m_isClosed;
-    Member<DOMException> m_exception;
-    Member<Observer> m_observer;
-    Member<Canceller> m_canceller;
+    OwnPtr<FetchDataConsumerHandle> m_handle;
+    Member<FetchDataLoader> m_fetchDataLoader;
 };
 
 } // namespace blink