Streams Implementation Update: async read

Source/core/streams/ReadableStreamImpl.h

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #ifndef ReadableStreamImpl_h
 #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/streams/ReadableStream.h"
 #include "wtf/Deque.h"
 #include "wtf/RefPtr.h"
 #include "wtf/text/WTFString.h"
 #include <utility>
 
 namespace blink {
 
 // We define the default ChunkTypeTraits for frequently used types.
@@ skipping 52 matching lines @@
 
     ReadableStreamImpl(ExecutionContext* executionContext, UnderlyingSource* source)
         : ReadableStreamImpl(executionContext, source, new DefaultStrategy) { }
     ReadableStreamImpl(ExecutionContext* executionContext, UnderlyingSource* source, Strategy* strategy)
         : ReadableStream(executionContext, source)
         , m_strategy(strategy)
         , m_totalQueueSize(0) { }
     ~ReadableStreamImpl() override { }
 
     // ReadableStream methods
-    ScriptValue readInternal(ScriptState*, ExceptionState&) override;
+    ScriptPromise read(ScriptState*) override;
 
     bool enqueue(typename ChunkTypeTraits::PassType);
 
     // This function is intended to be used by internal code to withdraw
     // queued data. This pulls all data from this stream's queue, but
     // ReadableStream public APIs can work with the behavior (i.e. it behaves
     // as if multiple read-one-buffer calls were made).
     void readInternal(Deque<std::pair<typename ChunkTypeTraits::HoldType, size_t>>& queue);
 
     DEFINE_INLINE_VIRTUAL_TRACE()
     {
         visitor->trace(m_strategy);
         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
     {
         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;
 };
 
 template <typename ChunkTypeTraits>
 bool ReadableStreamImpl<ChunkTypeTraits>::enqueue(typename ChunkTypeTraits::PassType chunk)
 {
     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>
 void ReadableStreamImpl<ChunkTypeTraits>::readInternal(Deque<std::pair<typename ChunkTypeTraits::HoldType, size_t>>& queue)
 {
     // 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);
     m_totalQueueSize = 0;
     readInternalPostAction();
 }
 
 } // namespace blink
 
 #endif // ReadableStreamImpl_h