Add JingleThreadWrapper.

JingleThreadWrapper wraps chromium thread with a talk_base::Thread interface.

BUG=None
TEST=Unittests

Review URL: http://codereview.chromium.org/6747017

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@79576 0039d316-1c4b-4281-b951-d872f2087c98

5 files changed, 431 insertions, 15 deletions

diff --git a/jingle/glue/thread_wrapper.cc b/jingle/glue/thread_wrapper.cc
new file mode 100644
index 0000000..f4551de
--- /dev/null
+++ b/jingle/glue/thread_wrapper.cc
@@ -0,0 +1,155 @@
+// Copyright (c) 2011 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 "jingle/glue/thread_wrapper.h"
+
+namespace jingle_glue {
+
+JingleThreadWrapper::JingleThreadWrapper(MessageLoop* message_loop)
+    : message_loop_(message_loop) {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+
+  talk_base::ThreadManager::SetCurrent(this);
+  message_loop_->AddDestructionObserver(this);
+}
+
+JingleThreadWrapper::~JingleThreadWrapper() {
+}
+
+void JingleThreadWrapper::WillDestroyCurrentMessageLoop() {
+  talk_base::ThreadManager::SetCurrent(NULL);
+  message_loop_->RemoveDestructionObserver(this);
+  delete this;
+}
+
+void JingleThreadWrapper::Post(
+    talk_base::MessageHandler* handler, uint32 message_id,
+    talk_base::MessageData* data, bool time_sensitive) {
+  PostTaskInternal(0, handler, message_id, data);
+}
+
+void JingleThreadWrapper::PostDelayed(
+    int delay_ms, talk_base::MessageHandler* handler,
+    uint32 message_id, talk_base::MessageData* data) {
+  PostTaskInternal(delay_ms, handler, message_id, data);
+}
+
+void JingleThreadWrapper::Clear(talk_base::MessageHandler* handler, uint32 id,
+                                talk_base::MessageList* removed) {
+  base::AutoLock auto_lock(lock_);
+
+  for (MessagesQueue::iterator it = messages_.begin();
+       it != messages_.end();) {
+    if (it->second.Match(handler, id)) {
+      if (removed) {
+        removed->push_back(it->second);
+      } else {
+        delete it->second.pdata;
+      }
+      MessagesQueue::iterator next = it;
+      ++next;
+      messages_.erase(it);
+      it = next;
+    } else {
+      ++it;
+    }
+  }
+}
+
+void JingleThreadWrapper::PostTaskInternal(
+    int delay_ms, talk_base::MessageHandler* handler,
+    uint32 message_id, talk_base::MessageData* data) {
+  int task_id;
+  talk_base::Message message;
+  message.phandler = handler;
+  message.message_id = message_id;
+  message.pdata = data;
+  {
+    base::AutoLock auto_lock(lock_);
+    task_id = ++last_task_id_;
+    messages_.insert(std::pair<int, talk_base::Message>(task_id, message));
+  }
+
+  if (delay_ms <= 0) {
+    message_loop_->PostTask(FROM_HERE, NewRunnableMethod(
+        this, &JingleThreadWrapper::RunTask, task_id));
+  } else {
+    message_loop_->PostDelayedTask(
+        FROM_HERE,
+        NewRunnableMethod(this, &JingleThreadWrapper::RunTask, task_id),
+        delay_ms);
+  }
+}
+
+void JingleThreadWrapper::RunTask(int task_id) {
+  bool have_message = false;
+  talk_base::Message message;
+  {
+    base::AutoLock auto_lock(lock_);
+    MessagesQueue::iterator it = messages_.find(task_id);
+    if (it != messages_.end()) {
+      have_message = true;
+      message = it->second;
+      messages_.erase(it);
+    }
+  }
+
+  if (have_message) {
+    message.phandler->OnMessage(&message);
+    delete message.pdata;
+  }
+}
+
+// All methods below are marked as not reached. See comments in the
+// header for more details.
+void JingleThreadWrapper::Quit() {
+  NOTREACHED();
+}
+
+bool JingleThreadWrapper::IsQuitting() {
+  NOTREACHED();
+  return false;
+}
+
+void JingleThreadWrapper::Restart() {
+  NOTREACHED();
+}
+
+bool JingleThreadWrapper::Get(talk_base::Message*, int, bool) {
+  NOTREACHED();
+  return false;
+}
+
+bool JingleThreadWrapper::Peek(talk_base::Message*, int) {
+  NOTREACHED();
+  return false;
+}
+
+void JingleThreadWrapper::PostAt(uint32, talk_base::MessageHandler*,
+                                 uint32, talk_base::MessageData*) {
+  NOTREACHED();
+}
+
+void JingleThreadWrapper::Dispatch(talk_base::Message* msg) {
+  NOTREACHED();
+}
+
+void JingleThreadWrapper::ReceiveSends() {
+  NOTREACHED();
+}
+
+int JingleThreadWrapper::GetDelay() {
+  NOTREACHED();
+  return 0;
+}
+
+void JingleThreadWrapper::Stop() {
+  NOTREACHED();
+}
+
+void JingleThreadWrapper::Run() {
+  NOTREACHED();
+}
+
+}  // namespace jingle_glue
diff --git a/jingle/glue/thread_wrapper.h b/jingle/glue/thread_wrapper.h
new file mode 100644
index 0000000..755d047
--- /dev/null
+++ b/jingle/glue/thread_wrapper.h
@@ -0,0 +1,89 @@
+// Copyright (c) 2011 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 JINGLE_GLUE_THREAD_WRAPPER_H_
+#define JINGLE_GLUE_THREAD_WRAPPER_H_
+
+#include <map>
+
+#include "base/message_loop.h"
+#include "base/synchronization/lock.h"
+#include "third_party/libjingle/source/talk/base/thread.h"
+
+class MessageLoop;
+
+namespace jingle_glue {
+
+// JingleThreadWrapper wraps Chromium threads using talk_base::Thread
+// interface. The object must be created on a thread it belongs
+// to. Each JingleThreadWrapper deletes itself when MessageLoop is
+// destroyed. Currently only the bare minimum that is used by P2P
+// part of libjingle is implemented.
+class JingleThreadWrapper
+    : public MessageLoop::DestructionObserver,
+      public talk_base::Thread {
+ public:
+  JingleThreadWrapper(MessageLoop* message_loop);
+
+  // MessageLoop::DestructionObserver implementation.
+  virtual void WillDestroyCurrentMessageLoop() OVERRIDE;
+
+  // talk_base::MessageQueue overrides.
+  virtual void Post(talk_base::MessageHandler *phandler, uint32 id = 0,
+                    talk_base::MessageData *pdata = NULL,
+                    bool time_sensitive = false) OVERRIDE;
+  virtual void PostDelayed(
+      int delay_ms, talk_base::MessageHandler* handler, uint32 id = 0,
+      talk_base::MessageData* data = NULL) OVERRIDE;
+  virtual void Clear(talk_base::MessageHandler* handler,
+                     uint32 id = talk_base::MQID_ANY,
+                     talk_base::MessageList* removed = NULL) OVERRIDE;
+
+  // Following methods are not supported.They are overriden just to
+  // ensure that they are not called (each of them contain NOTREACHED
+  // in the body). Some of this methods can be implemented if it
+  // becomes neccessary to use libjingle code that calls them.
+  virtual void Quit() OVERRIDE;
+  virtual bool IsQuitting() OVERRIDE;
+  virtual void Restart() OVERRIDE;
+  virtual bool Get(talk_base::Message* msg, int delay_ms = talk_base::kForever,
+                   bool process_io = true) OVERRIDE;
+  virtual bool Peek(talk_base::Message* msg, int delay_ms = 0) OVERRIDE;
+  virtual void PostAt(
+      uint32 timestamp, talk_base::MessageHandler* handler,
+      uint32 id = 0, talk_base::MessageData* data = NULL) OVERRIDE;
+  virtual void Dispatch(talk_base::Message* msg) OVERRIDE;
+  virtual void ReceiveSends() OVERRIDE;
+  virtual int GetDelay() OVERRIDE;
+
+  // talk_base::Thread overrides.
+  virtual void Stop() OVERRIDE;
+  virtual void Run() OVERRIDE;
+
+ private:
+  typedef std::map<int, talk_base::Message> MessagesQueue;
+
+  virtual ~JingleThreadWrapper();
+
+  void PostTaskInternal(
+      int delay_ms, talk_base::MessageHandler* handler,
+      uint32 message_id, talk_base::MessageData* data);
+  void RunTask(int task_id);
+
+  // Chromium thread used to execute messages posted on this thread.
+  MessageLoop* message_loop_;
+
+  // |lock_| must be locked when accessing |messages_|.
+  base::Lock lock_;
+  int last_task_id_;
+  MessagesQueue messages_;
+};
+
+}
+
+// Safe to disable refcounting because JingleThreadWrapper deletes
+// itself with the thread.
+DISABLE_RUNNABLE_METHOD_REFCOUNT(jingle_glue::JingleThreadWrapper);
+
+#endif  // JINGLE_GLUE_THREAD_WRAPPER_H_
diff --git a/jingle/glue/thread_wrapper_unittest.cc b/jingle/glue/thread_wrapper_unittest.cc
new file mode 100644
index 0000000..1c8943a
--- /dev/null
+++ b/jingle/glue/thread_wrapper_unittest.cc
@@ -0,0 +1,158 @@
+// Copyright (c) 2011 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 "base/message_loop.h"
+#include "jingle/glue/thread_wrapper.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using ::testing::InSequence;
+
+namespace jingle_glue {
+
+static const uint32 kTestMessage1 = 1;
+static const uint32 kTestMessage2 = 2;
+
+static const int kTestDelayMs1 = 10;
+static const int kTestDelayMs2 = 20;
+static const int kTestDelayMs3 = 30;
+static const int kTestDelayMs4 = 40;
+static const int kMaxTestDelay = 40;
+
+class MockMessageHandler : public talk_base::MessageHandler {
+ public:
+  MOCK_METHOD1(OnMessage, void(talk_base::Message* msg));
+};
+
+class ThreadWrapperTest : public testing::Test {
+ protected:
+  ThreadWrapperTest() {
+  }
+
+  talk_base::Thread* thread() {
+    return talk_base::Thread::Current();
+  }
+
+  virtual void SetUp() OVERRIDE {
+    wrapper_ = new JingleThreadWrapper(&message_loop_);
+  }
+
+  // ThreadWrapper destroyes itself when |message_loop_| is destroyed.
+  JingleThreadWrapper* wrapper_;
+  MessageLoop message_loop_;
+  MockMessageHandler handler1_;
+  MockMessageHandler handler2_;
+};
+
+TEST_F(ThreadWrapperTest, Create) {
+  EXPECT_EQ(thread(), static_cast<talk_base::Thread*>(wrapper_));
+}
+
+MATCHER_P3(MatchMessage, handler, message_id, data, "") {
+  return arg->phandler == handler &&
+      arg->message_id == message_id &&
+      arg->pdata == data;
+}
+
+TEST_F(ThreadWrapperTest, Post) {
+  talk_base::MessageData* data1_ = new talk_base::MessageData();
+  talk_base::MessageData* data2_ = new talk_base::MessageData();
+  talk_base::MessageData* data3_ = new talk_base::MessageData();
+  talk_base::MessageData* data4_ = new talk_base::MessageData();
+
+  thread()->Post(&handler1_, kTestMessage1, data1_);
+  thread()->Post(&handler1_, kTestMessage2, data2_);
+  thread()->Post(&handler2_, kTestMessage1, data3_);
+  thread()->Post(&handler2_, kTestMessage1, data4_);
+
+  InSequence in_seq;
+
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage1, data1_)));
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage2, data2_)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, data3_)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, data4_)));
+
+  message_loop_.RunAllPending();
+}
+
+TEST_F(ThreadWrapperTest, PostDelayed) {
+  talk_base::MessageData* data1_ = new talk_base::MessageData();
+  talk_base::MessageData* data2_ = new talk_base::MessageData();
+  talk_base::MessageData* data3_ = new talk_base::MessageData();
+  talk_base::MessageData* data4_ = new talk_base::MessageData();
+
+  // Post messages in reverse order and verify that they are called in
+  // the order of the delays.
+  thread()->PostDelayed(kTestDelayMs4, &handler2_, kTestMessage1, data4_);
+  thread()->PostDelayed(kTestDelayMs3, &handler2_, kTestMessage1, data3_);
+  thread()->PostDelayed(kTestDelayMs2, &handler1_, kTestMessage2, data2_);
+  thread()->PostDelayed(kTestDelayMs1, &handler1_, kTestMessage1, data1_);
+
+  InSequence in_seq;
+
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage1, data1_)));
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage2, data2_)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, data3_)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, data4_)));
+
+  message_loop_.PostDelayedTask(FROM_HERE, new MessageLoop::QuitTask(),
+                                kMaxTestDelay);
+  message_loop_.Run();
+}
+
+TEST_F(ThreadWrapperTest, Clear) {
+  thread()->Post(&handler1_, kTestMessage1, NULL);
+  thread()->Post(&handler1_, kTestMessage2, NULL);
+  thread()->Post(&handler2_, kTestMessage1, NULL);
+  thread()->Post(&handler2_, kTestMessage2, NULL);
+
+  thread()->Clear(&handler1_, kTestMessage2);
+
+  InSequence in_seq;
+
+  talk_base::MessageData* null_data = NULL;
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage1, null_data)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, null_data)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage2, null_data)));
+
+  message_loop_.RunAllPending();
+}
+
+TEST_F(ThreadWrapperTest, ClearDelayed) {
+  // Post messages in reverse order and verify that they are called in
+  // the order of the delays.
+  thread()->PostDelayed(kTestDelayMs4, &handler2_, kTestMessage1, NULL);
+  thread()->PostDelayed(kTestDelayMs3, &handler2_, kTestMessage1, NULL);
+  thread()->PostDelayed(kTestDelayMs2, &handler1_, kTestMessage2, NULL);
+  thread()->PostDelayed(kTestDelayMs1, &handler1_, kTestMessage1, NULL);
+
+  thread()->Clear(&handler1_, kTestMessage2);
+
+  InSequence in_seq;
+
+  talk_base::MessageData* null_data = NULL;
+  EXPECT_CALL(handler1_, OnMessage(
+      MatchMessage(&handler1_, kTestMessage1, null_data)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, null_data)));
+  EXPECT_CALL(handler2_, OnMessage(
+      MatchMessage(&handler2_, kTestMessage1, null_data)));
+
+  message_loop_.PostDelayedTask(FROM_HERE, new MessageLoop::QuitTask(),
+                                kMaxTestDelay);
+  message_loop_.Run();
+}
+
+}  // namespace jingle_glue
diff --git a/jingle/jingle.gyp b/jingle/jingle.gyp
index 7077884..1a50938 100644
--- a/jingle/jingle.gyp
+++ b/jingle/jingle.gyp
@@ -7,6 +7,22 @@
     'chromium_code': 1,
   },  # variables
   'targets': [
+    # A library of various utils for integration with libjingle.
+    {
+      'target_name': 'jingle_glue',
+      'type': '<(library)',
+      'sources': [
+        'glue/thread_wrapper.cc',
+        'glue/thread_wrapper.h',
+      ],
+      'dependencies': [
+        '../base/base.gyp:base',
+        '../third_party/libjingle/libjingle.gyp:libjingle',
+      ],
+      'export_dependent_settings': [
+        '../third_party/libjingle/libjingle.gyp:libjingle',
+      ],
+    },
     # A library for sending and receiving peer-issued notifications.
     #
     # TODO(akalin): Separate out the XMPP stuff from this library into
@@ -103,11 +119,10 @@
       ],
     },
     {
-      'target_name': 'notifier_unit_tests',
+      'target_name': 'jingle_unittests',
       'type': 'executable',
       'sources': [
-        # TODO(akalin): Write our own test suite and runner.
-        '../base/test/run_all_unittests.cc',
+        'glue/thread_wrapper_unittest.cc',
         'notifier/base/chrome_async_socket_unittest.cc',
         'notifier/base/fake_ssl_client_socket_unittest.cc',
         'notifier/base/xmpp_connection_unittest.cc',
@@ -119,11 +134,13 @@
         'notifier/listener/push_notifications_subscribe_task_unittest.cc',
         'notifier/listener/talk_mediator_unittest.cc',
         'notifier/listener/xml_element_util_unittest.cc',
+        'run_all_unittests.cc',
       ],
       'include_dirs': [
         '..',
       ],
       'dependencies': [
+        'jingle_glue',
         'notifier',
         'notifier_test_util',
         '../base/base.gyp:base',
@@ -134,18 +151,6 @@
         '../testing/gtest.gyp:gtest',
         '../third_party/libjingle/libjingle.gyp:libjingle',
       ],
-      # TODO(akalin): Remove this once we have our own test suite and
-      # runner.
-      'conditions': [
-        ['OS == "linux" or OS == "freebsd" or OS == "openbsd" or OS == "solaris"', {
-          'dependencies': [
-            # Needed to handle the #include chain:
-            #   base/test/test_suite.h
-            #   gtk/gtk.h
-            '../build/linux/system.gyp:gtk',
-          ],
-        }],
-      ],
     },
   ],
 }
diff --git a/jingle/run_all_unittests.cc b/jingle/run_all_unittests.cc
new file mode 100644
index 0000000..7fd6ef2
--- /dev/null
+++ b/jingle/run_all_unittests.cc
@@ -0,0 +1,9 @@
+// Copyright (c) 2011 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 "base/test/test_suite.h"
+
+int main(int argc, char** argv) {
+  return base::TestSuite(argc, argv).Run();
+}