Use ExclusiveStreamReader to lock Body.

Source/modules/fetch/Body.cpp

 // 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.
 
 #include "config.h"
 #include "modules/fetch/Body.h"
 
 #include "bindings/core/v8/ExceptionState.h"
 #include "bindings/core/v8/ScriptPromiseResolver.h"
 #include "bindings/core/v8/ScriptState.h"
 #include "bindings/core/v8/V8ArrayBuffer.h"
 #include "bindings/core/v8/V8ThrowException.h"
 #include "core/dom/DOMArrayBuffer.h"
 #include "core/fileapi/Blob.h"
 #include "core/fileapi/FileReaderLoader.h"
 #include "core/fileapi/FileReaderLoaderClient.h"
 #include "core/frame/UseCounter.h"
+#include "core/streams/ExclusiveStreamReader.h"
 #include "core/streams/UnderlyingSource.h"
 #include "modules/fetch/BodyStreamBuffer.h"
 
 namespace blink {
 
 class Body::BlobHandleReceiver final : public BodyStreamBuffer::BlobHandleCreatorClient {
 public:
     explicit BlobHandleReceiver(Body* body)
         : m_body(body)
     {
@@ skipping 198 matching lines @@
     // Created when createDrainingStream is called to drain the data.
     Member<BodyStreamBuffer> m_drainingStreamBuffer;
     Member<ReadableStreamImpl<ReadableStreamChunkTypeTraits<DOMArrayBuffer>>> m_stream;
     State m_state;
     // The count of the chunks which were enqueued to the ReadableStream.
     size_t m_queueCount;
 };
 
 ScriptPromise Body::readAsync(ScriptState* scriptState, ResponseType type)
 {
-    if (m_bodyUsed)
+    if (bodyUsed())
         return ScriptPromise::reject(scriptState, V8ThrowException::createTypeError(scriptState->isolate(), "Already read"));
 
     // When the main thread sends a V8::TerminateExecution() signal to a worker
     // thread, any V8 API on the worker thread starts returning an empty
     // handle. This can happen in Body::readAsync. To avoid the situation, we
     // first check the ExecutionContext and return immediately if it's already
     // gone (which means that the V8::TerminateExecution() signal has been sent
     // to this worker thread).
     ExecutionContext* executionContext = scriptState->executionContext();
     if (!executionContext)
         return ScriptPromise();
 
-    m_bodyUsed = true;
+    setBodyUsed();
     m_responseType = type;
 
     ASSERT(!m_resolver);
     m_resolver = ScriptPromiseResolver::create(scriptState);
     ScriptPromise promise = m_resolver->promise();
 
     if (m_stream) {
         ASSERT(m_streamSource);
         bool dataLost;
         m_streamSource->createDrainingStream(&dataLost)->readAllAndCreateBlobHandle(contentTypeForBuffer(), new BlobHandleReceiver(this));
@@ skipping 79 matching lines @@
         ASSERT(!m_streamSource);
         m_streamSource = new ReadableStreamSource(this);
         m_stream = new ReadableStreamImpl<ReadableStreamChunkTypeTraits<DOMArrayBuffer>>(executionContext(), m_streamSource);
         m_streamSource->startStream(m_stream);
     }
     return m_stream;
 }
 
 bool Body::bodyUsed() const
 {
-    return m_bodyUsed;
+    return m_bodyUsed || (m_stream && m_stream->isLocked());
 }
 
 void Body::setBodyUsed()
 {
+    ASSERT(!m_bodyUsed);
+    ASSERT(!m_stream || !m_stream->isLocked());
+    // Note that technically we can set BodyUsed even when the stream is
+    // closed or errored, but getReader doesn't work then.
+    if (m_stream && m_stream->stateInternal() != ReadableStream::Closed && m_stream->stateInternal() != ReadableStream::Errored) {
+        TrackExceptionState exceptionState;
+        m_streamReader = m_stream->getReader(exceptionState);
+        ASSERT(!exceptionState.hadException());
+    }
     m_bodyUsed = true;
 }
 
 bool Body::streamAccessed() const
 {
     return m_stream;
 }
 
 BodyStreamBuffer* Body::createDrainingStream(bool* dataLost)
 {
@@ skipping 20 matching lines @@
     if (m_stream && m_stream->hasPendingActivity())
         return true;
     return false;
 }
 
 DEFINE_TRACE(Body)
 {
     visitor->trace(m_resolver);
     visitor->trace(m_stream);
     visitor->trace(m_streamSource);
+    visitor->trace(m_streamReader);
     ActiveDOMObject::trace(visitor);
 }
 
 Body::Body(ExecutionContext* context)
     : ActiveDOMObject(context)
     , m_bodyUsed(false)
     , m_responseType(ResponseType::ResponseUnknown)
 {
 }
 
@@ skipping 72 matching lines @@
 
 void Body::didBlobHandleReceiveError(PassRefPtrWillBeRawPtr<DOMException> exception)
 {
     if (!m_resolver)
         return;
     m_resolver->reject(exception);
     m_resolver.clear();
 }
 
 } // namespace blink