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..260ce2c28e7efba89e9f3aa628ca7147319fcc0f 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,215 @@ 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(bool allowInvalidSize) override
+        {
+            print("drainAsBlobDataHandle(%d)\n", (int)allowInvalidSize);
+            RefPtr<BlobDataHandle> blobDataHandle = m_reader->drainAsBlobDataHandle(allowInvalidSize);
+            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 DrainingBodyStreamBuffer;
+
+class MODULES_EXPORT BodyStreamBuffer final : public GarbageCollectedFinalized<BodyStreamBuffer> {
+public:
+    static BodyStreamBuffer* create(PassOwnPtr<FetchDataConsumerHandle> handle) { return new BodyStreamBuffer(handle); }
+    static BodyStreamBuffer* createEmpty();
+
+    FetchDataConsumerHandle* handle() const;
+    PassOwnPtr<FetchDataConsumerHandle> releaseHandle();
+
+    class DrainingStreamNotificationClient : public GarbageCollectedMixin {
     public:
-        virtual ~BlobHandleCreatorClient() { }
-        virtual void didCreateBlobHandle(PassRefPtr<BlobDataHandle>) = 0;
-        virtual void didFail(DOMException*) = 0;
-        DEFINE_INLINE_VIRTUAL_TRACE() { }
+        virtual ~DrainingStreamNotificationClient() { }
+        // Called after FetchDataLoader::Client methods.
+        virtual void didFetchDataLoadFinishedFromDrainingStream() = 0;
     };
-    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);
+
+    DEFINE_INLINE_TRACE()
+    {
+        visitor->trace(m_fetchDataLoader);
+        visitor->trace(m_drainingStreamNotificationClient);
+    }
+
+    void didFetchDataLoadFinished();
 
 private:
-    Deque<RefPtr<DOMArrayBuffer>> m_queue;
-    bool m_isClosed;
-    Member<DOMException> m_exception;
-    Member<Observer> m_observer;
-    Member<Canceller> m_canceller;
+    explicit BodyStreamBuffer(PassOwnPtr<FetchDataConsumerHandle> handle) : m_handle(handle) { }
+
+    void setDrainingStreamNotificationClient(DrainingStreamNotificationClient*);
+
+    void startLoading(FetchDataLoader*, FetchDataLoader::Client*);
+    // Call DrainingStreamNotificationClient.
+    void doDrainingStreamNotification();
+    // Clear DrainingStreamNotificationClient without calling.
+    void clearDrainingStreamNotification();
+
+    friend class DrainingBodyStreamBuffer;
+
+    OwnPtr<FetchDataConsumerHandle> m_handle;
+    Member<FetchDataLoader> m_fetchDataLoader;
+    Member<DrainingStreamNotificationClient> m_drainingStreamNotificationClient;
+};
+
+// DrainingBodyStreamBuffer wraps BodyStreamBuffer returned from
+// Body::createDrainingStream() and calls DrainingStreamNotificationClient
+// callbacks unless leak() is called:
+// - If startLoading() is called, the callback is called after loading finished.
+// - If drainAsBlobDataHandle() is called, the callback is called immediately.
+// - If leak() is called, the callback is no longer called.
+// Any calls to DrainingBodyStreamBuffer methods after a call to either of
+// methods above is no-op.
+// After calling one of the methods above, we don't have to keep
+// DrainingBodyStreamBuffer alive.
+// If DrainingBodyStreamBuffer is destructed before any of above is called,
+// the callback is called at destruction.
+class MODULES_EXPORT DrainingBodyStreamBuffer final {
+public:
+    static PassOwnPtr<DrainingBodyStreamBuffer> create(BodyStreamBuffer* buffer, BodyStreamBuffer::DrainingStreamNotificationClient* client)
+    {
+        return adoptPtr(new DrainingBodyStreamBuffer(buffer, client));
+    }
+    ~DrainingBodyStreamBuffer();
+    void startLoading(FetchDataLoader*, FetchDataLoader::Client*);
+    BodyStreamBuffer* leak();
+    PassRefPtr<BlobDataHandle> drainAsBlobDataHandle(FetchDataConsumerHandle::Reader::BlobSizePolicy);
+
+private:
+    explicit DrainingBodyStreamBuffer(BodyStreamBuffer*, BodyStreamBuffer::DrainingStreamNotificationClient*);
+
+    Persistent<BodyStreamBuffer> m_buffer;
 };
 
 } // namespace blink