Change BodyStreamBuffer to be FetchDataConsumerHandle-based and enable backpressure in Fetch API

Source/modules/fetch/DataConsumerTeeTest.cpp

 // Copyright 2015 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/DataConsumerTee.h"
 
 #include "core/testing/DummyPageHolder.h"
 #include "core/testing/NullExecutionContext.h"
 #include "modules/fetch/DataConsumerHandleTestUtil.h"
 #include "platform/Task.h"
 #include "platform/ThreadSafeFunctional.h"
 #include "platform/WebThreadSupportingGC.h"
 #include "public/platform/Platform.h"
 #include "public/platform/WebThread.h"
 #include "public/platform/WebTraceLocation.h"
 #include "public/platform/WebWaitableEvent.h"
 #include "wtf/PassRefPtr.h"
 #include "wtf/RefPtr.h"
 
 #include <gtest/gtest.h>
 #include <string.h>
 #include <v8.h>
 
 namespace blink {
 namespace {
 
+using ::testing::InSequence;
+using ::testing::Return;
+using ::testing::StrictMock;
+using ::testing::_;
+using Checkpoint = StrictMock<::testing::MockFunction<void(int)>>;
+
 using Result = WebDataConsumerHandle::Result;
 using Thread = DataConsumerHandleTestUtil::Thread;
 const WebDataConsumerHandle::Flags kNone = WebDataConsumerHandle::FlagNone;
 const Result kOk = WebDataConsumerHandle::Ok;
 const Result kShouldWait = WebDataConsumerHandle::ShouldWait;
 const Result kDone = WebDataConsumerHandle::Done;
 const Result kUnexpectedError = WebDataConsumerHandle::UnexpectedError;
+const FetchDataConsumerHandle::Reader::BlobSizePolicy kDisallowBlobWithInvalidSize = FetchDataConsumerHandle::Reader::DisallowBlobWithInvalidSize;
+const FetchDataConsumerHandle::Reader::BlobSizePolicy kAllowBlobWithInvalidSize = FetchDataConsumerHandle::Reader::AllowBlobWithInvalidSize;
 
 using Command = DataConsumerHandleTestUtil::Command;
 using Handle = DataConsumerHandleTestUtil::ReplayingHandle;
 
 class HandleReader : public WebDataConsumerHandle::Client {
 public:
     HandleReader() : m_finalResult(kOk) { }
 
     // Need to wait for the event signal after this function is called.
     void start(PassOwnPtr<WebDataConsumerHandle> handle)
@@ skipping 85 matching lines @@
         m_reader = handle->obtainReader(this);
     }
 
     OwnPtr<Thread> m_thread;
     OwnPtr<WebDataConsumerHandle::Reader> m_reader;
     String m_readString;
     Result m_finalResult;
     OwnPtr<WebWaitableEvent> m_waitableEvent;
 };
 
+template<typename Handle>
 class TeeCreationThread {
 public:
-    void run(PassOwnPtr<WebDataConsumerHandle> src, OwnPtr<WebDataConsumerHandle>* dest1, OwnPtr<WebDataConsumerHandle>* dest2)
+    void run(PassOwnPtr<Handle> src, OwnPtr<Handle>* dest1, OwnPtr<Handle>* dest2)
     {
         m_thread = adoptPtr(new Thread("src thread", Thread::WithExecutionContext));
         m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
-        m_thread->thread()->postTask(FROM_HERE, new Task(threadSafeBind(&TeeCreationThread::runInternal, AllowCrossThreadAccess(this), src, AllowCrossThreadAccess(dest1), AllowCrossThreadAccess(dest2))));
+        m_thread->thread()->postTask(FROM_HERE, new Task(threadSafeBind(&TeeCreationThread<Handle>::runInternal, AllowCrossThreadAccess(this), src, AllowCrossThreadAccess(dest1), AllowCrossThreadAccess(dest2))));
         m_waitableEvent->wait();
     }
 
     Thread* thread() { return m_thread.get(); }
 
 private:
-    void runInternal(PassOwnPtr<WebDataConsumerHandle> src, OwnPtr<WebDataConsumerHandle>* dest1, OwnPtr<WebDataConsumerHandle>* dest2)
+    void runInternal(PassOwnPtr<Handle> src, OwnPtr<Handle>* dest1, OwnPtr<Handle>* dest2)
     {
         DataConsumerTee::create(m_thread->executionContext(), src, dest1, dest2);
         m_waitableEvent->signal();
     }
 
     OwnPtr<Thread> m_thread;
     OwnPtr<WebWaitableEvent> m_waitableEvent;
 };
 
+// TODO(hiroshige): Merge similar classes into DataConsumerHandleTestUtil.h/cpp.
+class MockFetchDataConsumerHandle : public FetchDataConsumerHandle {
+public:
+    static PassOwnPtr<StrictMock<MockFetchDataConsumerHandle>> create() { return adoptPtr(new StrictMock<MockFetchDataConsumerHandle>); }
+    MOCK_METHOD1(obtainReaderInternal, Reader*(Client*));
+};
+
+class MockFetchDataConsumerReader : public FetchDataConsumerHandle::Reader {
+public:
+    static PassOwnPtr<StrictMock<MockFetchDataConsumerReader>> create() { return adoptPtr(new StrictMock<MockFetchDataConsumerReader>); }
+
+    using Result = WebDataConsumerHandle::Result;
+    using Flags = WebDataConsumerHandle::Flags;
+    MOCK_METHOD4(read, Result(void*, size_t, Flags, size_t*));
+    MOCK_METHOD3(beginRead, Result(const void**, Flags, size_t*));
+    MOCK_METHOD1(endRead, Result(size_t));
+    MOCK_METHOD1(drainAsBlobDataHandle, PassRefPtr<BlobDataHandle>(BlobSizePolicy));
+
+    ~MockFetchDataConsumerReader() override
+    {
+        destruct();
+    }
+    MOCK_METHOD0(destruct, void());
+};
+
 TEST(DataConsumerTeeTest, CreateDone)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Done));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     r1.waitableEvent()->wait();
@@ skipping 12 matching lines @@
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Wait));
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Data, "world"));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Done));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     r1.waitableEvent()->wait();
@@ skipping 13 matching lines @@
 
     src->add(Command(Command::Wait));
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Data, "world"));
     src->add(Command(Command::Wait));
     src->add(Command(Command::Done));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleTwoPhaseReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     r1.waitableEvent()->wait();
     r2.waitableEvent()->wait();
 
     EXPECT_EQ(kDone, r1.finalResult());
     EXPECT_EQ("hello, world", r1.readString());
 
     EXPECT_EQ(kDone, r2.finalResult());
     EXPECT_EQ("hello, world", r2.readString());
 }
 
 TEST(DataConsumerTeeTest, Error)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
     src->add(Command(Command::Error));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     r1.waitableEvent()->wait();
     r2.waitableEvent()->wait();
 
     EXPECT_EQ(kUnexpectedError, r1.finalResult());
     EXPECT_EQ(kUnexpectedError, r2.finalResult());
 }
 
 void postStop(Thread* thread)
 {
     thread->executionContext()->stopActiveDOMObjects();
 }
 
 TEST(DataConsumerTeeTest, StopSource)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     // We can pass a raw pointer because the subsequent |wait| calls ensure
     // t->thread() is alive.
     t->thread()->thread()->postTask(FROM_HERE, new Task(threadSafeBind(postStop, AllowCrossThreadAccess(t->thread()))));
 
     r1.waitableEvent()->wait();
     r2.waitableEvent()->wait();
 
     EXPECT_EQ(kUnexpectedError, r1.finalResult());
     EXPECT_EQ(kUnexpectedError, r2.finalResult());
 }
 
 TEST(DataConsumerTeeTest, DetachSource)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r2.start(dest2.release());
 
     t = nullptr;
 
     r1.waitableEvent()->wait();
     r2.waitableEvent()->wait();
 
     EXPECT_EQ(kUnexpectedError, r1.finalResult());
     EXPECT_EQ(kUnexpectedError, r2.finalResult());
 }
 
 TEST(DataConsumerTeeTest, DetachSourceAfterReadingDone)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
     src->add(Command(Command::Done));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     HandleReader r1, r2;
     r1.start(dest1.release());
     r1.waitableEvent()->wait();
 
     EXPECT_EQ(kDone, r1.finalResult());
@@ skipping 10 matching lines @@
 
 TEST(DataConsumerTeeTest, DetachOneDestination)
 {
     OwnPtr<Handle> src(Handle::create());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
     src->add(Command(Command::Done));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     dest1 = nullptr;
 
     HandleReader r2;
     r2.start(dest2.release());
     r2.waitableEvent()->wait();
 
     EXPECT_EQ(kDone, r2.finalResult());
     EXPECT_EQ("hello, world", r2.readString());
 }
 
 TEST(DataConsumerTeeTest, DetachBothDestinationsShouldStopSourceReader)
 {
     OwnPtr<Handle> src(Handle::create());
     RefPtr<Handle::Context> context(src->context());
     OwnPtr<WebDataConsumerHandle> dest1, dest2;
 
     src->add(Command(Command::Data, "hello, "));
     src->add(Command(Command::Data, "world"));
 
-    OwnPtr<TeeCreationThread> t = adoptPtr(new TeeCreationThread());
+    OwnPtr<TeeCreationThread<WebDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<WebDataConsumerHandle>());
     t->run(src.release(), &dest1, &dest2);
 
     ASSERT_TRUE(dest1);
     ASSERT_TRUE(dest2);
 
     dest1 = nullptr;
     dest2 = nullptr;
 
     // Collect garbage to finalize the source reader.
     Heap::collectAllGarbage();
     context->detached()->wait();
 }
 
+TEST(FetchDataConsumerTeeTest, Create)
+{
+    RefPtr<BlobDataHandle> blobDataHandle = BlobDataHandle::create();
+    OwnPtr<MockFetchDataConsumerHandle> src(MockFetchDataConsumerHandle::create());
+    OwnPtr<MockFetchDataConsumerReader> reader(MockFetchDataConsumerReader::create());
+
+    Checkpoint checkpoint;
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*src, obtainReaderInternal(_)).WillOnce(Return(reader.get()));
+    EXPECT_CALL(*reader, drainAsBlobDataHandle(kAllowBlobWithInvalidSize)).WillOnce(Return(blobDataHandle));
+    EXPECT_CALL(*reader, destruct());
+    EXPECT_CALL(checkpoint, Call(2));
+
+    // |reader| is adopted by |obtainReader|.
+    ASSERT_TRUE(reader.leakPtr());
+
+    OwnPtr<FetchDataConsumerHandle> dest1, dest2;
+    OwnPtr<TeeCreationThread<FetchDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<FetchDataConsumerHandle>());
+
+    checkpoint.Call(1);
+    t->run(src.release(), &dest1, &dest2);
+    checkpoint.Call(2);
+
+    ASSERT_TRUE(dest1);
+    ASSERT_TRUE(dest2);
+    EXPECT_EQ(blobDataHandle, dest1->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+    EXPECT_EQ(blobDataHandle, dest2->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+}
+
+TEST(FetchDataConsumerTeeTest, CreateFromBlobWithInvalidSize)
+{
+    RefPtr<BlobDataHandle> blobDataHandle = BlobDataHandle::create(BlobData::create(), -1);
+    OwnPtr<MockFetchDataConsumerHandle> src(MockFetchDataConsumerHandle::create());
+    OwnPtr<MockFetchDataConsumerReader> reader(MockFetchDataConsumerReader::create());
+
+    Checkpoint checkpoint;
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*src, obtainReaderInternal(_)).WillOnce(Return(reader.get()));
+    EXPECT_CALL(*reader, drainAsBlobDataHandle(kAllowBlobWithInvalidSize)).WillOnce(Return(blobDataHandle));
+    EXPECT_CALL(*reader, destruct());
+    EXPECT_CALL(checkpoint, Call(2));
+
+    // |reader| is adopted by |obtainReader|.
+    ASSERT_TRUE(reader.leakPtr());
+
+    OwnPtr<FetchDataConsumerHandle> dest1, dest2;
+    OwnPtr<TeeCreationThread<FetchDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<FetchDataConsumerHandle>());
+
+    checkpoint.Call(1);
+    t->run(src.release(), &dest1, &dest2);
+    checkpoint.Call(2);
+
+    ASSERT_TRUE(dest1);
+    ASSERT_TRUE(dest2);
+    EXPECT_FALSE(dest1->obtainReader(nullptr)->drainAsBlobDataHandle(kDisallowBlobWithInvalidSize));
+    EXPECT_EQ(blobDataHandle, dest1->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+    EXPECT_FALSE(dest2->obtainReader(nullptr)->drainAsBlobDataHandle(kDisallowBlobWithInvalidSize));
+    EXPECT_EQ(blobDataHandle, dest2->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+}
+
+TEST(FetchDataConsumerTeeTest, CreateDone)
+{
+    OwnPtr<Handle> src(Handle::create());
+    OwnPtr<FetchDataConsumerHandle> dest1, dest2;
+
+    src->add(Command(Command::Done));
+
+    OwnPtr<TeeCreationThread<FetchDataConsumerHandle>> t = adoptPtr(new TeeCreationThread<FetchDataConsumerHandle>());
+    t->run(createFetchDataConsumerHandleFromWebHandle(src.release()), &dest1, &dest2);
+
+    ASSERT_TRUE(dest1);
+    ASSERT_TRUE(dest2);
+
+    EXPECT_FALSE(dest1->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+    EXPECT_FALSE(dest2->obtainReader(nullptr)->drainAsBlobDataHandle(kAllowBlobWithInvalidSize));
+
+    HandleReader r1, r2;
+    r1.start(dest1.release());
+    r2.start(dest2.release());
+
+    r1.waitableEvent()->wait();
+    r2.waitableEvent()->wait();
+
+    EXPECT_EQ(kDone, r1.finalResult());
+    EXPECT_EQ(String(), r1.readString());
+
+    EXPECT_EQ(kDone, r2.finalResult());
+    EXPECT_EQ(String(), r2.readString());
+}
+
 } // namespace
 } // namespace blink