Streams Implementation Update: async read

Source/core/streams/ReadableStreamImpl.h

Index: Source/core/streams/ReadableStreamImpl.h
diff --git a/Source/core/streams/ReadableStreamImpl.h b/Source/core/streams/ReadableStreamImpl.h
index 08362e152561aa068cb4fdaae0d1ed965c7ffa05..bde0d74493be2c396e90902a13132387766e754a 100644
--- a/Source/core/streams/ReadableStreamImpl.h
+++ b/Source/core/streams/ReadableStreamImpl.h
@@ -6,11 +6,15 @@
 #define ReadableStreamImpl_h
 
 #include "bindings/core/v8/ExceptionState.h"
+#include "bindings/core/v8/ScriptPromise.h"
+#include "bindings/core/v8/ScriptPromiseResolver.h"
 #include "bindings/core/v8/ScriptState.h"
 #include "bindings/core/v8/ScriptValue.h"
 #include "bindings/core/v8/V8ArrayBuffer.h"
 #include "bindings/core/v8/V8Binding.h"
+#include "bindings/core/v8/V8IteratorResultValue.h"
 #include "core/dom/DOMArrayBuffer.h"
+#include "core/dom/DOMException.h"
 #include "core/streams/ReadableStream.h"
 #include "wtf/Deque.h"
 #include "wtf/RefPtr.h"
@@ -82,7 +86,7 @@ public:
     ~ReadableStreamImpl() override { }
 
     // ReadableStream methods
-    ScriptValue readInternal(ScriptState*, ExceptionState&) override;
+    ScriptPromise read(ScriptState*) override;
 
     bool enqueue(typename ChunkTypeTraits::PassType);
 
@@ -95,10 +99,19 @@ public:
     DEFINE_INLINE_VIRTUAL_TRACE()
     {
         visitor->trace(m_strategy);
+#if ENABLE(OILPAN)
+        visitor->trace(m_pendingReads);
+#endif
         ReadableStream::trace(visitor);
     }
 
 private:
+#if ENABLE(OILPAN)
+    using PendingReads = HeapDeque<Member<ScriptPromiseResolver>>;
+#else
+    using PendingReads = Deque<RefPtr<ScriptPromiseResolver>>;
+#endif
+
     // ReadableStream methods
     bool isQueueEmpty() const override { return m_queue.isEmpty(); }
     void clearQueue() override
@@ -106,13 +119,33 @@ private:
         m_queue.clear();
         m_totalQueueSize = 0;
     }
+
+    void resolveAllPendingReadsAsDone() override
+    {
+        for (auto& resolver : m_pendingReads) {
+            ScriptState::Scope scope(resolver->scriptState());
+            resolver->resolve(v8IteratorResultDone(resolver->scriptState()));
+        }
+        m_pendingReads.clear();
+    }
+
+    void rejectAllPendingReads(PassRefPtrWillBeRawPtr<DOMException> r) override
+    {
+        RefPtrWillBeRawPtr<DOMException> reason = r;
+        for (auto& resolver : m_pendingReads)
+            resolver->reject(reason);
+        m_pendingReads.clear();
+    }
+
     bool shouldApplyBackpressure() override
     {
         return m_strategy->shouldApplyBackpressure(m_totalQueueSize, this);
     }
+    bool hasPendingReads() const override { return !m_pendingReads.isEmpty(); }
 
     Member<Strategy> m_strategy;
     Deque<std::pair<typename ChunkTypeTraits::HoldType, size_t>> m_queue;
+    PendingReads m_pendingReads;
     size_t m_totalQueueSize;
 };
 
@@ -122,26 +155,37 @@ bool ReadableStreamImpl<ChunkTypeTraits>::enqueue(typename ChunkTypeTraits::Pass
     size_t size = m_strategy->size(chunk, this);
     if (!enqueuePreliminaryCheck())
         return false;
-    m_queue.append(std::make_pair(chunk, size));
-    m_totalQueueSize += size;
+
+    if (m_pendingReads.isEmpty()) {
+        m_queue.append(std::make_pair(chunk, size));
+        m_totalQueueSize += size;
+        return enqueuePostAction();
+    }
+
+    RefPtrWillBeRawPtr<ScriptPromiseResolver> resolver = m_pendingReads.takeFirst();
+    ScriptState* scriptState = resolver->scriptState();
+    ScriptState::Scope scope(scriptState);
+    resolver->resolve(v8IteratorResult(scriptState, chunk));
     return enqueuePostAction();
 }
 
 template <typename ChunkTypeTraits>
-ScriptValue ReadableStreamImpl<ChunkTypeTraits>::readInternal(ScriptState* scriptState, ExceptionState& exceptionState)
+ScriptPromise ReadableStreamImpl<ChunkTypeTraits>::read(ScriptState* scriptState)
 {
-    readInternalPreliminaryCheck(exceptionState);
-    if (exceptionState.hadException())
-        return ScriptValue();
     ASSERT(stateInternal() == Readable);
-    ASSERT(!m_queue.isEmpty());
+    if (m_queue.isEmpty()) {
+        m_pendingReads.append(ScriptPromiseResolver::create(scriptState));
+        return m_pendingReads.last()->promise();
+    }
+
     auto pair = m_queue.takeFirst();
     typename ChunkTypeTraits::HoldType chunk = pair.first;
     size_t size = pair.second;
     ASSERT(m_totalQueueSize >= size);
     m_totalQueueSize -= size;
     readInternalPostAction();
-    return ChunkTypeTraits::toScriptValue(scriptState, chunk);
+
+    return ScriptPromise::cast(scriptState, v8IteratorResult(scriptState, chunk));
 }
 
 template <typename ChunkTypeTraits>
@@ -149,7 +193,7 @@ void ReadableStreamImpl<ChunkTypeTraits>::readInternal(Deque<std::pair<typename
 {
     // We omit the preliminary check. Check it by yourself.
     ASSERT(stateInternal() == Readable);
-    ASSERT(!m_queue.isEmpty());
+    ASSERT(m_pendingReads.isEmpty());
     ASSERT(queue.isEmpty());
 
     queue.swap(m_queue);