Introduce CompositeDataConsumerHandle.

Source/modules/fetch/CompositeDataConsumerHandleTest.cpp

Index: Source/modules/fetch/CompositeDataConsumerHandleTest.cpp
diff --git a/Source/modules/fetch/CompositeDataConsumerHandleTest.cpp b/Source/modules/fetch/CompositeDataConsumerHandleTest.cpp
new file mode 100644
index 0000000000000000000000000000000000000000..0ae7ab7ceb83fd361ff626c5a7198c48dc02003f
--- /dev/null
+++ b/Source/modules/fetch/CompositeDataConsumerHandleTest.cpp
@@ -0,0 +1,313 @@
+// 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/CompositeDataConsumerHandle.h"
+
+#include "platform/Task.h"
+#include "platform/ThreadSafeFunctional.h"
+#include "platform/heap/Handle.h"
+#include "public/platform/Platform.h"
+#include "public/platform/WebThread.h"
+#include "public/platform/WebTraceLocation.h"
+#include "public/platform/WebWaitableEvent.h"
+#include "wtf/Locker.h"
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+namespace blink {
+
+namespace {
+
+using Result = WebDataConsumerHandle::Result;
+using Flags = WebDataConsumerHandle::Flags;
+using ::testing::InSequence;
+using ::testing::Return;
+using ::testing::StrictMock;
+using Checkpoint = StrictMock<::testing::MockFunction<void(int)>>;
+
+const Result kShouldWait = WebDataConsumerHandle::ShouldWait;
+const Result kDone = WebDataConsumerHandle::Done;
+const Result kOk = WebDataConsumerHandle::Ok;
+const Result kUnexpectedError = WebDataConsumerHandle::UnexpectedError;
+const Flags kNone = WebDataConsumerHandle::FlagNone;
+
+class NoopClient final : public WebDataConsumerHandle::Client {
+public:
+    void didGetReadable() override { }
+};
+
+class MockHandle : public WebDataConsumerHandle {
+public:
+    static PassOwnPtr<StrictMock<MockHandle>> create() { return adoptPtr(new StrictMock<MockHandle>); }
+
+    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(registerClient, void(Client*));
+    MOCK_METHOD0(unregisterClient, void());
+};
+
+class ThreadingTestBase {
+public:
+    class DataConsumerHandle final : public WebDataConsumerHandle {
+    public:
+        DataConsumerHandle(const String& name, ThreadingTestBase* test) : m_name(name.isolatedCopy()), m_test(test) { }
+        Result read(void*, size_t, Flags, size_t*) override { return ShouldWait; }
+        Result beginRead(const void**, Flags, size_t*) override { return ShouldWait; }
+        Result endRead(size_t) override { return ShouldWait; }
+        void registerClient(Client*) override
+        {
+            m_test->record(m_name, "registerClient");
+        }
+        void unregisterClient() override
+        {
+            m_test->record(m_name, "unregisterClient");
+        }
+    private:
+        const String m_name;
+        ThreadingTestBase* m_test;
+    };
+
+    void record(const String& instanceName, const String& functionName)
+    {
+        String threadName;
+        if (m_readingThread->isCurrentThread()) {
+            threadName = "the reading thread";
+        } else if (m_updatingThread->isCurrentThread()) {
+            threadName = "the updating thread";
+        } else {
+            threadName = "an unknown thread";
+        }
+        MutexLocker locker(m_loggingMutex);
+        m_result.append(instanceName.isolatedCopy() + "->" + functionName.isolatedCopy() + " is called on " + threadName + ".\n");
+    }
+    const String& result()
+    {
+        MutexLocker locker(m_loggingMutex);
+        return m_result;
+    }
+
+    static void signalDone(ThreadingTestBase* test)
+    {
+        test->m_waitableEvent->signal();
+    }
+
+protected:
+    OwnPtr<WebThread> m_readingThread;
+    OwnPtr<WebThread> m_updatingThread;
+    OwnPtr<CompositeDataConsumerHandle> m_handle;
+    NoopClient m_client;
+    OwnPtr<WebWaitableEvent> m_waitableEvent;
+    Mutex m_loggingMutex;
+    String m_result;
+};
+
+class ThreadingRegistrationTest : public ThreadingTestBase {
+public:
+    void run()
+    {
+        m_readingThread = adoptPtr(Platform::current()->createThread("reading thread"));
+        m_updatingThread = adoptPtr(Platform::current()->createThread("updating thread"));
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", this)));
+
+        m_readingThread->postTask(FROM_HERE, new Task(threadSafeBind(&registerClient, AllowCrossThreadAccess(this))));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    static void registerClient(ThreadingRegistrationTest* test)
+    {
+        test->m_handle->registerClient(&test->m_client);
+        test->m_updatingThread->postTask(FROM_HERE, new Task(threadSafeBind(&update, AllowCrossThreadAccess(test))));
+    }
+    static void update(ThreadingRegistrationTest* test)
+    {
+        test->m_handle->update(adoptPtr(new DataConsumerHandle("handle2", test)));
+        test->m_readingThread->postTask(FROM_HERE, new Task(threadSafeBind(&signalDone, AllowCrossThreadAccess(test))));
+    }
+};
+
+class ThreadingRegistrationDeleteTest : public ThreadingTestBase {
+public:
+    void run()
+    {
+        m_readingThread = adoptPtr(Platform::current()->createThread("reading thread"));
+        m_updatingThread = adoptPtr(Platform::current()->createThread("updating thread"));
+        m_waitableEvent = adoptPtr(Platform::current()->createWaitableEvent());
+        m_handle = CompositeDataConsumerHandle::create(adoptPtr(new DataConsumerHandle("handle1", this)));
+
+        m_readingThread->postTask(FROM_HERE, new Task(threadSafeBind(&registerClient, AllowCrossThreadAccess(this))));
+
+        m_waitableEvent->wait();
+    }
+
+private:
+    static void registerClient(ThreadingRegistrationDeleteTest* test)
+    {
+        test->m_handle->registerClient(&test->m_client);
+        test->m_updatingThread->postTask(FROM_HERE, new Task(threadSafeBind(&update, AllowCrossThreadAccess(test))));
+    }
+    static void update(ThreadingRegistrationDeleteTest* test)
+    {
+        test->m_handle->update(adoptPtr(new DataConsumerHandle("handle2", test)));
+        test->m_handle.clear();
+        test->m_readingThread->postTask(FROM_HERE, new Task(threadSafeBind(&signalDone, AllowCrossThreadAccess(test))));
+    }
+};
+
+TEST(CompositeDataConsumerHandleTest, CreateWaitingHandle)
+{
+    char buffer[20];
+    const void* p = nullptr;
+    size_t size = 0;
+    OwnPtr<WebDataConsumerHandle> handle = CompositeDataConsumerHandle::createWaitingHandle();
+
+    EXPECT_EQ(kShouldWait, handle->read(buffer, sizeof(buffer), kNone, &size));
+    EXPECT_EQ(kShouldWait, handle->beginRead(&p, kNone, &size));
+    EXPECT_EQ(kUnexpectedError, handle->endRead(99));
+}
+
+TEST(CompositeDataConsumerHandleTest, CreateDoneHandle)
+{
+    char buffer[20];
+    const void* p = nullptr;
+    size_t size = 0;
+    OwnPtr<WebDataConsumerHandle> handle = CompositeDataConsumerHandle::createDoneHandle();
+
+    EXPECT_EQ(kDone, handle->read(buffer, sizeof(buffer), kNone, &size));
+    EXPECT_EQ(kDone, handle->beginRead(&p, kNone, &size));
+    EXPECT_EQ(kUnexpectedError, handle->endRead(99));
+}
+
+TEST(CompositeDataConsumerHandleTest, Read)
+{
+    char buffer[20];
+    size_t size = 0;
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, read(buffer, sizeof(buffer), kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(*handle2, read(buffer, sizeof(buffer), kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(3));
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    EXPECT_EQ(kOk, handle->read(buffer, sizeof(buffer), kNone, &size));
+    checkpoint.Call(1);
+    handle->update(handle2.release());
+    checkpoint.Call(2);
+    EXPECT_EQ(kOk, handle->read(buffer, sizeof(buffer), kNone, &size));
+    checkpoint.Call(3);
+}
+
+TEST(CompositeDataConsumerHandleTest, TwoPhaseRead)
+{
+    const void* p = nullptr;
+    size_t size = 0;
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*handle1, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(checkpoint, Call(3));
+    EXPECT_CALL(*handle2, beginRead(&p, kNone, &size)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(4));
+    EXPECT_CALL(*handle2, endRead(0)).WillOnce(Return(kOk));
+    EXPECT_CALL(checkpoint, Call(5));
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    EXPECT_EQ(kOk, handle->beginRead(&p, kNone, &size));
+    checkpoint.Call(1);
+    EXPECT_EQ(kOk, handle->endRead(0));
+    checkpoint.Call(2);
+    handle->update(handle2.release());
+    checkpoint.Call(3);
+    EXPECT_EQ(kOk, handle->beginRead(&p, kNone, &size));
+    checkpoint.Call(4);
+    EXPECT_EQ(kOk, handle->endRead(0));
+    checkpoint.Call(5);
+}
+
+TEST(CompositeDataConsumerHandleTest, ClientRegistration)
+{
+    Checkpoint checkpoint;
+
+    OwnPtr<MockHandle> handle1 = MockHandle::create();
+    OwnPtr<MockHandle> handle2 = MockHandle::create();
+    OwnPtr<MockHandle> handle3 = MockHandle::create();
+    NoopClient client;
+
+    InSequence s;
+    EXPECT_CALL(checkpoint, Call(0));
+    EXPECT_CALL(*handle1, registerClient(&client));
+    EXPECT_CALL(checkpoint, Call(1));
+    EXPECT_CALL(*handle1, unregisterClient());
+    EXPECT_CALL(checkpoint, Call(2));
+    EXPECT_CALL(*handle1, registerClient(&client));
+    EXPECT_CALL(checkpoint, Call(3));
+    EXPECT_CALL(*handle1, unregisterClient());
+    EXPECT_CALL(*handle2, registerClient(&client));
+    EXPECT_CALL(checkpoint, Call(4));
+    EXPECT_CALL(*handle2, unregisterClient());
+    EXPECT_CALL(checkpoint, Call(5));
+    EXPECT_CALL(checkpoint, Call(6));
+
+    OwnPtr<CompositeDataConsumerHandle> handle = CompositeDataConsumerHandle::create(handle1.release());
+    checkpoint.Call(0);
+    handle->registerClient(&client);
+    checkpoint.Call(1);
+    handle->unregisterClient();
+    checkpoint.Call(2);
+    handle->registerClient(&client);
+    checkpoint.Call(3);
+    handle->update(handle2.release());
+    checkpoint.Call(4);
+    handle->unregisterClient();
+    checkpoint.Call(5);
+    handle->update(handle3.release());
+    checkpoint.Call(6);
+}
+
+TEST(CompositeDataConsumerHandleTest, RegisterClientOnDifferentThreads)
+{
+    ThreadingRegistrationTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "handle1->registerClient is called on the reading thread.\n"
+        "handle1->unregisterClient is called on the reading thread.\n"
+        "handle2->registerClient is called on the reading thread.\n",
+        test.result());
+}
+
+TEST(CompositeDataConsumerHandleTest, DeleteHandleWhileUpdating)
+{
+    ThreadingRegistrationDeleteTest test;
+    test.run();
+
+    EXPECT_EQ(
+        "handle1->registerClient is called on the reading thread.\n",
+        test.result());
+}
+
+} // namespace
+
+} // namespace blink