Use per-frame TaskRunner instead of thread's default in DataConsumerHandle

third_party/WebKit/Source/modules/fetch/DataConsumerHandleTestUtil.h

 // 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.
 
 #ifndef DataConsumerHandleTestUtil_h
 #define DataConsumerHandleTestUtil_h
 
 #include "bindings/core/v8/ScriptState.h"
 #include "core/testing/NullExecutionContext.h"
 #include "gin/public/isolate_holder.h"
@@ skipping 42 matching lines @@
             // Creating an isolate in addition to GarbageCollection.
             ScriptExecution,
             // Creating an execution context in addition to ScriptExecution.
             WithExecutionContext,
         };
 
         Thread(const char* name, InitializationPolicy = GarbageCollection);
         ~Thread();
 
         WebThreadSupportingGC* thread() { return m_thread.get(); }
+        WebTaskRunner* taskRunner() { return m_thread->platformThread().getWebTaskRunner(); }
         ExecutionContext* getExecutionContext() { return m_executionContext.get(); }
         ScriptState* getScriptState() { return m_scriptState.get(); }
         v8::Isolate* isolate() { return m_isolateHolder->isolate(); }
 
     private:
         void initialize();
         void shutdown();
 
         std::unique_ptr<WebThreadSupportingGC> m_thread;
         const InitializationPolicy m_initializationPolicy;
@@ skipping 47 matching lines @@
                 ASSERT(m_holder);
                 m_holder->readingThread()->postTask(location, std::move(task));
             }
             void postTaskToUpdatingThread(const WebTraceLocation& location, std::unique_ptr<CrossThreadClosure> task)
             {
                 MutexLocker locker(m_holderMutex);
                 ASSERT(m_holder);
                 m_holder->updatingThread()->postTask(location, std::move(task));
             }
 
+            WebTaskRunner* getReadingThreadTaskRunner()
+            {
+                return m_holder->readingThread()->platformThread().getWebTaskRunner();
+            }
+
         private:
             Context()
                 : m_holder(nullptr) { }
             String currentThreadName()
             {
                 MutexLocker locker(m_holderMutex);
                 if (m_holder) {
                     if (m_holder->readingThread()->isCurrentThread())
                         return "the reading thread";
                     if (m_holder->updatingThread()->isCurrentThread())
@@ skipping 60 matching lines @@
             USING_FAST_MALLOC(DataConsumerHandle);
         public:
             static std::unique_ptr<WebDataConsumerHandle> create(const String& name, PassRefPtr<Context> context)
             {
                 return wrapUnique(new DataConsumerHandle(name, context));
             }
 
         private:
             DataConsumerHandle(const String& name, PassRefPtr<Context> context) : m_name(name.isolatedCopy()), m_context(context) { }
 
-            std::unique_ptr<Reader> obtainReader(Client*) { return WTF::wrapUnique(new ReaderImpl(m_name, m_context)); }
+            std::unique_ptr<Reader> obtainReader(Client*, std::unique_ptr<WebTaskRunner>)
+            {
+                return WTF::wrapUnique(new ReaderImpl(m_name, m_context));
+            }
             const char* debugName() const override { return "ThreadingTestBase::DataConsumerHandle"; }
 
             const String m_name;
             RefPtr<Context> m_context;
         };
 
         void resetReader() { m_reader = nullptr; }
         void signalDone() { m_waitableEvent->signal(); }
         const String& result() { return m_context->result(); }
         void postTaskToReadingThread(const WebTraceLocation& location, std::unique_ptr<CrossThreadClosure> task)
@@ skipping 35 matching lines @@
             m_waitableEvent = wrapUnique(new WaitableEvent());
             m_handle = std::move(handle);
 
             postTaskToReadingThreadAndWait(BLINK_FROM_HERE, crossThreadBind(&Self::obtainReader, wrapPassRefPtr(this)));
         }
 
     private:
         ThreadingHandleNotificationTest() = default;
         void obtainReader()
         {
-            m_reader = m_handle->obtainReader(this);
+            m_reader = m_handle->obtainReader(this, m_context->getReadingThreadTaskRunner()->clone());
         }
         void didGetReadable() override
         {
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::resetReader, wrapPassRefPtr(this)));
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::signalDone, wrapPassRefPtr(this)));
         }
 
         std::unique_ptr<WebDataConsumerHandle> m_handle;
     };
 
     class ThreadingHandleNoNotificationTest : public ThreadingTestBase, public WebDataConsumerHandle::Client {
     public:
         using Self = ThreadingHandleNoNotificationTest;
         static PassRefPtr<Self> create() { return adoptRef(new Self); }
 
         void run(std::unique_ptr<WebDataConsumerHandle> handle)
         {
             ThreadHolder holder(this);
             m_waitableEvent = wrapUnique(new WaitableEvent());
             m_handle = std::move(handle);
 
             postTaskToReadingThreadAndWait(BLINK_FROM_HERE, crossThreadBind(&Self::obtainReader, wrapPassRefPtr(this)));
         }
 
     private:
         ThreadingHandleNoNotificationTest() = default;
         void obtainReader()
         {
-            m_reader = m_handle->obtainReader(this);
+            m_reader = m_handle->obtainReader(this, m_context->getReadingThreadTaskRunner()->clone());
             m_reader = nullptr;
             postTaskToReadingThread(BLINK_FROM_HERE, crossThreadBind(&Self::signalDone, wrapPassRefPtr(this)));
         }
         void didGetReadable() override
         {
             ASSERT_NOT_REACHED();
         }
 
         std::unique_ptr<WebDataConsumerHandle> m_handle;
     };
 
     class MockFetchDataConsumerHandle : public FetchDataConsumerHandle {
     public:
         static std::unique_ptr<::testing::StrictMock<MockFetchDataConsumerHandle>> create() { return wrapUnique(new ::testing::StrictMock<MockFetchDataConsumerHandle>); }
-        MOCK_METHOD1(obtainFetchDataReader, std::unique_ptr<Reader>(Client*));
+        MOCK_METHOD1(obtainFetchDataReaderMock, std::unique_ptr<Reader>(Client*));
+
+        std::unique_ptr<Reader> obtainFetchDataReader(Client* client, std::unique_ptr<WebTaskRunner>)
+        {
+            return obtainFetchDataReaderMock(client);
+        }
 
     private:
         const char* debugName() const override { return "MockFetchDataConsumerHandle"; }
     };
 
     class MockFetchDataConsumerReader : public FetchDataConsumerHandle::Reader {
     public:
         static std::unique_ptr<::testing::StrictMock<MockFetchDataConsumerReader>> create() { return wrapUnique(new ::testing::StrictMock<MockFetchDataConsumerReader>); }
 
         using Result = WebDataConsumerHandle::Result;
@@ skipping 73 matching lines @@
         // creator thread. This function must be called BEFORE any reader is
         // obtained.
         void add(const Command&);
 
         class Context final : public ThreadSafeRefCounted<Context> {
         public:
             static PassRefPtr<Context> create() { return adoptRef(new Context); }
 
             // This function cannot be called after creating a tee.
             void add(const Command&);
-            void attachReader(WebDataConsumerHandle::Client*);
+            void attachReader(WebDataConsumerHandle::Client*, std::unique_ptr<WebTaskRunner>);
             void detachReader();
             void detachHandle();
             Result beginRead(const void** buffer, Flags, size_t* available);
             Result endRead(size_t readSize);
             WaitableEvent* detached() { return m_detached.get(); }
 
         private:
             Context();
             bool isEmpty() const { return m_commands.isEmpty(); }
             const Command& top();
             void consume(size_t);
             size_t offset() const { return m_offset; }
             void notify();
             void notifyInternal();
 
             Deque<Command> m_commands;
             size_t m_offset;
-            WebThread* m_readerThread;
+            std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
             Client* m_client;
             Result m_result;
             bool m_isHandleAttached;
             Mutex m_mutex;
             std::unique_ptr<WaitableEvent> m_detached;
         };
 
         Context* getContext() { return m_context.get(); }
-        std::unique_ptr<Reader> obtainReader(Client*) override;
+        std::unique_ptr<Reader> obtainReader(Client*, std::unique_ptr<WebTaskRunner>) override;
 
     private:
         class ReaderImpl;
 
         ReplayingHandle();
         const char* debugName() const override { return "ReplayingHandle"; }
 
         RefPtr<Context> m_context;
     };
 
@@ skipping 10 matching lines @@
     };
 
     // HandleReader reads all data from the given WebDataConsumerHandle using
     // Reader::read on the thread on which it is created. When reading is done
     // or failed, it calls the given callback with the result.
     class HandleReader final : public WebDataConsumerHandle::Client {
         USING_FAST_MALLOC(HandleReader);
     public:
         using OnFinishedReading = WTF::Function<void(std::unique_ptr<HandleReadResult>)>;
 
-        HandleReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>);
+        HandleReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>, std::unique_ptr<WebTaskRunner> readerTaskRunner);
         void didGetReadable() override;
 
     private:
         void runOnFinishedReading(std::unique_ptr<HandleReadResult>);
 
+        std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
         std::unique_ptr<WebDataConsumerHandle::Reader> m_reader;
         std::unique_ptr<OnFinishedReading> m_onFinishedReading;
         Vector<char> m_data;
     };
 
     // HandleTwoPhaseReader does the same as HandleReader, but it uses
     // |beginRead| / |endRead| instead of |read|.
     class HandleTwoPhaseReader final : public WebDataConsumerHandle::Client {
         USING_FAST_MALLOC(HandleTwoPhaseReader);
     public:
         using OnFinishedReading = WTF::Function<void(std::unique_ptr<HandleReadResult>)>;
 
-        HandleTwoPhaseReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>);
+        HandleTwoPhaseReader(std::unique_ptr<WebDataConsumerHandle>, std::unique_ptr<OnFinishedReading>, std::unique_ptr<WebTaskRunner>);
         void didGetReadable() override;
 
     private:
         void runOnFinishedReading(std::unique_ptr<HandleReadResult>);
 
+        std::unique_ptr<WebTaskRunner> m_readerTaskRunner;
         std::unique_ptr<WebDataConsumerHandle::Reader> m_reader;
         std::unique_ptr<OnFinishedReading> m_onFinishedReading;
         Vector<char> m_data;
     };
 
     // HandleReaderRunner<T> creates a dedicated thread and run T on the thread
     // where T is one of HandleReader and HandleTwophaseReader.
     template <typename T>
     class HandleReaderRunner final {
         STACK_ALLOCATED();
     public:
         explicit HandleReaderRunner(std::unique_ptr<WebDataConsumerHandle> handle)
             : m_thread(wrapUnique(new Thread("reading thread")))
             , m_event(wrapUnique(new WaitableEvent()))
             , m_isDone(false)
         {
-            m_thread->thread()->postTask(BLINK_FROM_HERE, crossThreadBind(&HandleReaderRunner::start, crossThreadUnretained(this), passed(std::move(handle))));
+            m_thread->thread()->postTask(BLINK_FROM_HERE, crossThreadBind(&HandleReaderRunner::start, crossThreadUnretained(this), passed(std::move(handle)), passed(m_thread->taskRunner()->clone())));
         }
         ~HandleReaderRunner()
         {
             wait();
         }
 
         std::unique_ptr<HandleReadResult> wait()
         {
             if (m_isDone)
                 return nullptr;
             m_event->wait();
             m_isDone = true;
             return std::move(m_result);
         }
 
     private:
-        void start(std::unique_ptr<WebDataConsumerHandle> handle)
+        void start(std::unique_ptr<WebDataConsumerHandle> handle, std::unique_ptr<WebTaskRunner> readerTaskRunner)
         {
-            m_handleReader = wrapUnique(new T(std::move(handle), WTF::bind(&HandleReaderRunner::onFinished, WTF::unretained(this))));
+            m_handleReader = wrapUnique(new T(std::move(handle), WTF::bind(&HandleReaderRunner::onFinished, WTF::unretained(this)), std::move(readerTaskRunner)));
         }
 
         void onFinished(std::unique_ptr<HandleReadResult> result)
         {
             m_handleReader = nullptr;
             m_result = std::move(result);
             m_event->signal();
         }
 
         std::unique_ptr<Thread> m_thread;
         std::unique_ptr<WaitableEvent> m_event;
         std::unique_ptr<HandleReadResult> m_result;
         bool m_isDone;
 
         std::unique_ptr<T> m_handleReader;
     };
 };
 
+std::unique_ptr<WebTaskRunner> getCurrentTaskRunner();
+
 } // namespace blink
 
 #endif // DataConsumerHandleTestUtil_h