Streams Implementation Update: async read

Source/core/streams/ReadableStream.cpp

Index: Source/core/streams/ReadableStream.cpp
diff --git a/Source/core/streams/ReadableStream.cpp b/Source/core/streams/ReadableStream.cpp
index 92ac8a87bceb92eb7f076a89f3d1bb8532553598..684e1358ad5e7b81c9271b425e3b255884e9ac8b 100644
--- a/Source/core/streams/ReadableStream.cpp
+++ b/Source/core/streams/ReadableStream.cpp
@@ -34,32 +34,6 @@ private:
     }
 };
 
-class ResolveWithReady : public ScriptFunction {
-public:
-    static v8::Handle<v8::Function> create(ScriptState* scriptState, ReadableStream* stream)
-    {
-        return (new ResolveWithReady(scriptState, stream))->bindToV8Function();
-    }
-
-    DEFINE_INLINE_TRACE()
-    {
-        visitor->trace(m_stream);
-        ScriptFunction::trace(visitor);
-    }
-
-private:
-    ResolveWithReady(ScriptState* scriptState, ReadableStream* stream)
-        : ScriptFunction(scriptState)
-        , m_stream(stream) { }
-
-    ScriptValue call(ScriptValue value) override
-    {
-        return ScriptValue(scriptState(), m_stream->ready(scriptState()).v8Value());
-    }
-
-    Member<ReadableStream> m_stream;
-};
-
 } // namespace
 
 ReadableStream::ReadableStream(ExecutionContext* executionContext, UnderlyingSource* source)
@@ -68,9 +42,7 @@ ReadableStream::ReadableStream(ExecutionContext* executionContext, UnderlyingSou
     , m_isStarted(false)
     , m_isDraining(false)
     , m_isPulling(false)
-    , m_state(Waiting)
-    , m_ready(new WaitPromise(executionContext, this, WaitPromise::Ready))
-    , m_closed(new ClosedPromise(executionContext, this, ClosedPromise::Closed))
+    , m_state(Readable)
 {
     suspendIfNeeded();
 }
@@ -79,14 +51,6 @@ ReadableStream::~ReadableStream()
 {
 }
 
-String ReadableStream::stateString() const
-{
-    if (m_reader)
-        return "waiting";
-
-    return stateToString(m_state);
-}
-
 bool ReadableStream::enqueuePreliminaryCheck()
 {
     // This is a bit different from what spec says: it says we should throw
@@ -107,93 +71,28 @@ bool ReadableStream::enqueuePostAction()
     if (m_state == Errored)
         return false;
 
-    if (m_state == Waiting) {
-        // ReadableStream::hasPendingActivity return value gets false when
-        // |m_state| is changed to Closed or Errored from Waiting or Readable.
-        // On the other hand, the wrappers should be kept alive when |m_ready|
-        // and |m_close| resolution and rejection are called. Hence we call
-        // ScriptPromiseProperty::resolve and ScriptPromiseProperty::reject
-        // *before* changing state, no matter if the state change actually
-        // changes hasPendingActivity return value.
-        m_ready->resolve(ToV8UndefinedGenerator());
-        m_state = Readable;
-    }
-
     return !shouldApplyBackpressure;
 }
 
 void ReadableStream::close()
 {
-    if (m_state == Waiting) {
-        m_ready->resolve(ToV8UndefinedGenerator());
-        m_closed->resolve(ToV8UndefinedGenerator());
-        if (m_reader)
-            m_reader->releaseLock();
-        m_state = Closed;
-    } else if (m_state == Readable) {
-        m_isDraining = true;
-    }
-}
-
-void ReadableStream::readInternalPreliminaryCheck(ExceptionState& exceptionState)
-{
-    if (m_state == Waiting) {
-        exceptionState.throwTypeError("read is called while state is waiting");
-        return;
-    }
-    if (m_state == Closed) {
-        exceptionState.throwTypeError("read is called while state is closed");
-        return;
-    }
-    if (m_state == Errored) {
-        exceptionState.throwDOMException(m_exception->code(), m_exception->message());
+    if (m_state != Readable)
         return;
-    }
+
+    if (isQueueEmpty())
+        closeInternal();
+    else
+        m_isDraining = true;
 }
 
 void ReadableStream::readInternalPostAction()
 {
     ASSERT(m_state == Readable);
-    if (isQueueEmpty()) {
-        if (m_isDraining) {
-            m_state = Closed;
-            m_closed->resolve(ToV8UndefinedGenerator());
-            if (m_reader)
-                m_reader->releaseLock();
-        } else {
-            m_ready->reset();
-            m_state = Waiting;
-        }
-    }
+    if (isQueueEmpty() && m_isDraining)
+        closeInternal();
     callPullIfNeeded();
 }
 
-ScriptValue ReadableStream::read(ScriptState* scriptState, ExceptionState& exceptionState)
-{
-    if (m_reader) {
-        exceptionState.throwTypeError("this stream is locked to a ReadableStreamReader");
-        return ScriptValue();
-    }
-    return readInternal(scriptState, exceptionState);
-}
-
-ScriptPromise ReadableStream::ready(ScriptState* scriptState)
-{
-    if (m_reader) {
-        return m_reader->released(scriptState).then(ResolveWithReady::create(scriptState, this));
-    }
-
-    if (m_state == Waiting) {
-        return readyInternal(scriptState).then(ResolveWithReady::create(scriptState, this));
-    }
-    return readyInternal(scriptState);
-}
-
-ScriptPromise ReadableStream::readyInternal(ScriptState* scriptState)
-{
-    return m_ready->promise(scriptState->world());
-}
-
 ScriptPromise ReadableStream::cancel(ScriptState* scriptState, ScriptValue reason)
 {
     if (m_reader)
@@ -203,44 +102,26 @@ ScriptPromise ReadableStream::cancel(ScriptState* scriptState, ScriptValue reaso
     if (m_state == Errored)
         return ScriptPromise::rejectWithDOMException(scriptState, m_exception);
 
-    ASSERT(m_state == Readable || m_state == Waiting);
-    if (m_state == Waiting)
-        m_ready->resolve(ToV8UndefinedGenerator());
-    clearQueue();
-    m_closed->resolve(ToV8UndefinedGenerator());
-    m_state = Closed;
-    return m_source->cancelSource(scriptState, reason).then(ConstUndefined::create(scriptState));
+    return cancelInternal(scriptState, reason);
 }
 
-ScriptPromise ReadableStream::closed(ScriptState* scriptState)
+ScriptPromise ReadableStream::cancelInternal(ScriptState* scriptState, ScriptValue reason)
 {
-    return m_closed->promise(scriptState->world());
+    closeInternal();
+    return m_source->cancelSource(scriptState, reason).then(ConstUndefined::create(scriptState));
 }
 
 void ReadableStream::error(PassRefPtrWillBeRawPtr<DOMException> exception)
 {
-    switch (m_state) {
-    case Waiting:
-        m_exception = exception;
-        m_ready->reject(m_exception);
-        m_closed->reject(m_exception);
-        m_state = Errored;
-        if (m_reader)
-            m_reader->releaseLock();
-        break;
-    case Readable:
-        clearQueue();
-        m_exception = exception;
-        m_ready->reset();
-        m_ready->reject(m_exception);
-        m_closed->reject(m_exception);
-        m_state = Errored;
-        if (m_reader)
-            m_reader->releaseLock();
-        break;
-    default:
-        break;
-    }
+    if (m_state != ReadableStream::Readable)
+        return;
+
+    m_exception = exception;
+    clearQueue();
+    rejectAllPendingReads(m_exception);
+    m_state = Errored;
+    if (m_reader)
+        m_reader->releaseLock();
 }
 
 void ReadableStream::didSourceStart()
@@ -251,14 +132,6 @@ void ReadableStream::didSourceStart()
 
 ReadableStreamReader* ReadableStream::getReader(ExceptionState& exceptionState)
 {
-    if (m_state == Closed) {
-        exceptionState.throwTypeError("this stream is already closed");
-        return nullptr;
-    }
-    if (m_state == Errored) {
-        exceptionState.throwDOMException(m_exception->code(), m_exception->message());
-        return nullptr;
-    }
     if (m_reader) {
         exceptionState.throwTypeError("already locked to a ReadableStreamReader");
         return nullptr;
@@ -285,9 +158,19 @@ void ReadableStream::callPullIfNeeded()
     m_source->pullSource();
 }
 
+void ReadableStream::closeInternal()
+{
+    ASSERT(m_state == Readable);
+    m_state = Closed;
+    resolveAllPendingReadsAsDone();
+    clearQueue();
+    if (m_reader)
+        m_reader->releaseLock();
+}
+
 bool ReadableStream::hasPendingActivity() const
 {
-    return m_state == Waiting || m_state == Readable;
+    return m_state == Readable;
 }
 
 void ReadableStream::stop()
@@ -299,27 +182,9 @@ void ReadableStream::stop()
 DEFINE_TRACE(ReadableStream)
 {
     visitor->trace(m_source);
-    visitor->trace(m_ready);
-    visitor->trace(m_closed);
     visitor->trace(m_exception);
     visitor->trace(m_reader);
     ActiveDOMObject::trace(visitor);
 }
 
-String ReadableStream::stateToString(State state)
-{
-    switch (state) {
-    case Readable:
-        return "readable";
-    case Waiting:
-        return "waiting";
-    case Closed:
-        return "closed";
-    case Errored:
-        return "errored";
-    }
-    ASSERT(false);
-    return String();
-}
-
 } // namespace blink