Update JingleThreadWrapper to allow it to be created using task runner.

Previously JingleThreadWrapper required a MessageLoop. In NaCl
we don't have MessageLoop for the main plugin thread, but have a task runner.
This CL allows to use JingleThreadWrapper on the main NaCl thread.

BUG=134216
TBR=tommi@chromium.org

Review URL: https://chromiumcodereview.appspot.com/10823224

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

content/renderer/media/media_stream_impl.cc

@@ -370,7 +370,7 @@ void MediaStreamImpl::OnSocketDispatcherDestroyed() {
 
 void MediaStreamImpl::InitializeWorkerThread(talk_base::Thread** thread,
                                              base::WaitableEvent* event) {
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
   jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
   *thread = jingle_glue::JingleThreadWrapper::current();
   event->Signal();
@@ -392,7 +392,7 @@ void MediaStreamImpl::DeleteIpcNetworkManager() {
 bool MediaStreamImpl::EnsurePeerConnectionFactory() {
   DCHECK(CalledOnValidThread());
   if (!signaling_thread_) {
-    jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+    jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
     jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
     signaling_thread_ = jingle_glue::JingleThreadWrapper::current();
   }

content/renderer/p2p/p2p_transport_impl.cc

@@ -58,7 +58,7 @@ bool P2PTransportImpl::Init(WebKit::WebFrame* web_frame,
 
   // Before proceeding, ensure we have libjingle thread wrapper for
   // the current thread.
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
 
   name_ = name;
   event_handler_ = event_handler;

jingle/glue/pseudotcp_adapter_unittest.cc

@@ -303,7 +303,7 @@ class TCPChannelTester : public base::RefCountedThreadSafe<TCPChannelTester> {
 class PseudoTcpAdapterTest : public testing::Test {
  protected:
   virtual void SetUp() OVERRIDE {
-    JingleThreadWrapper::EnsureForCurrentThread();
+    JingleThreadWrapper::EnsureForCurrentMessageLoop();
 
     host_socket_ = new FakeSocket();
     client_socket_ = new FakeSocket();

jingle/glue/thread_wrapper.cc

@@ -29,10 +29,12 @@ base::LazyInstance<base::ThreadLocalPointer<JingleThreadWrapper> >
     g_jingle_thread_wrapper = LAZY_INSTANCE_INITIALIZER;
 
 // static
-void JingleThreadWrapper::EnsureForCurrentThread() {
+void JingleThreadWrapper::EnsureForCurrentMessageLoop() {
   if (JingleThreadWrapper::current() == NULL) {
-    g_jingle_thread_wrapper.Get().Set(
-        new JingleThreadWrapper(MessageLoop::current()));
+    MessageLoop* message_loop = MessageLoop::current();
+    g_jingle_thread_wrapper.Get().Set(new JingleThreadWrapper(
+        message_loop->message_loop_proxy()));
+    message_loop->AddDestructionObserver(current());
   }
 
   DCHECK_EQ(talk_base::Thread::Current(), current());
@@ -43,19 +45,18 @@ JingleThreadWrapper* JingleThreadWrapper::current() {
   return g_jingle_thread_wrapper.Get().Get();
 }
 
-JingleThreadWrapper::JingleThreadWrapper(MessageLoop* message_loop)
+JingleThreadWrapper::JingleThreadWrapper(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner)
     : talk_base::Thread(new talk_base::NullSocketServer()),
-      message_loop_(message_loop),
+      task_runner_(task_runner),
       send_allowed_(false),
       last_task_id_(0),
-      pending_send_event_(true, false) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
-
-  talk_base::ThreadManager::Instance()->UnwrapCurrentThread();
-  talk_base::ThreadManager::Instance()->SetCurrentThread(this);
+      pending_send_event_(true, false),
+      weak_ptr_factory_(this),
+      weak_ptr_(weak_ptr_factory_.GetWeakPtr()) {
+  DCHECK(task_runner->BelongsToCurrentThread());
+  DCHECK(!talk_base::Thread::Current());
   talk_base::MessageQueueManager::Instance()->Add(this);
-  message_loop_->AddDestructionObserver(this);
-
   WrapCurrent();
 }
 
@@ -69,7 +70,6 @@ void JingleThreadWrapper::WillDestroyCurrentMessageLoop() {
   g_jingle_thread_wrapper.Get().Set(NULL);
   talk_base::ThreadManager::Instance()->SetCurrentThread(NULL);
   talk_base::MessageQueueManager::Instance()->Remove(this);
-  message_loop_->RemoveDestructionObserver(this);
   talk_base::SocketServer* ss = socketserver();
   delete this;
   delete ss;
@@ -162,9 +162,9 @@ void JingleThreadWrapper::Send(talk_base::MessageHandler *handler, uint32 id,
   // Need to signal |pending_send_event_| here in case the thread is
   // sending message to another thread.
   pending_send_event_.Signal();
-  message_loop_->PostTask(FROM_HERE,
-                          base::Bind(&JingleThreadWrapper::ProcessPendingSends,
-                                     base::Unretained(this)));
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&JingleThreadWrapper::ProcessPendingSends,
+                                    weak_ptr_));
 
 
   while (!pending_send.done_event.IsSignaled()) {
@@ -214,14 +214,14 @@ void JingleThreadWrapper::PostTaskInternal(
   }
 
   if (delay_ms <= 0) {
-    message_loop_->PostTask(FROM_HERE,
-                            base::Bind(&JingleThreadWrapper::RunTask,
-                                       base::Unretained(this), task_id));
+    task_runner_->PostTask(FROM_HERE,
+                           base::Bind(&JingleThreadWrapper::RunTask,
+                                      weak_ptr_, task_id));
   } else {
-    message_loop_->PostDelayedTask(FROM_HERE,
-                                   base::Bind(&JingleThreadWrapper::RunTask,
-                                              base::Unretained(this), task_id),
-                                   base::TimeDelta::FromMilliseconds(delay_ms));
+    task_runner_->PostDelayedTask(FROM_HERE,
+                                  base::Bind(&JingleThreadWrapper::RunTask,
+                                             weak_ptr_, task_id),
+                                  base::TimeDelta::FromMilliseconds(delay_ms));
   }
 }
 

jingle/glue/thread_wrapper.h

@@ -16,25 +16,34 @@
 
 namespace jingle_glue {
 
-// JingleThreadWrapper wraps Chromium threads using talk_base::Thread
-// interface. The object must be created by calling
-// EnsureForCurrentThread(). Each JingleThreadWrapper deletes itself
-// when MessageLoop is destroyed. Currently only the bare minimum that
-// is used by P2P part of libjingle is implemented.
+// JingleThreadWrapper implements talk_base::Thread interface on top of
+// Chromium's SingleThreadTaskRunner interface. Currently only the bare minimum
+// that is used by P2P part of libjingle is implemented. There are two ways to
+// create this object:
+//
+// - Call EnsureForCurrentMessageLoop(). This approach works only on threads
+//   that have MessageLoop In this case JingleThreadWrapper deletes itself
+//   automatically when MessageLoop is destroyed.
+// - Using JingleThreadWrapper() constructor. In this case the creating code
+//   must pass a valid task runner for the current thread and also delete the
+//   wrapper later.
 class JingleThreadWrapper
     : public MessageLoop::DestructionObserver,
       public talk_base::Thread {
  public:
-  // Create JingleThreadWrapper for the current thread if it hasn't
-  // been created yet.
-  static void EnsureForCurrentThread();
+  // Create JingleThreadWrapper for the current thread if it hasn't been created
+  // yet. The thread wrapper is destroyed automatically when the current
+  // MessageLoop is destroyed.
+  static void EnsureForCurrentMessageLoop();
 
   // Returns thread wrapper for the current thread. NULL is returned
-  // if EnsureForCurrentThread() has never been called for this
+  // if EnsureForCurrentMessageLoop() has never been called for this
   // thread.
   static JingleThreadWrapper* current();
 
-  JingleThreadWrapper(MessageLoop* message_loop);
+  explicit JingleThreadWrapper(
+     scoped_refptr<base::SingleThreadTaskRunner> task_runner);
+  virtual ~JingleThreadWrapper();
 
   // Sets whether the thread can be used to send messages
   // synchronously to another thread using Send() method. Set to false
@@ -90,16 +99,14 @@ class JingleThreadWrapper
   typedef std::map<int, talk_base::Message> MessagesQueue;
   struct PendingSend;
 
-  virtual ~JingleThreadWrapper();
-
   void PostTaskInternal(
       int delay_ms, talk_base::MessageHandler* handler,
       uint32 message_id, talk_base::MessageData* data);
   void RunTask(int task_id);
   void ProcessPendingSends();
 
-  // Chromium thread used to execute messages posted on this thread.
-  MessageLoop* message_loop_;
+  // Task runner used to execute messages posted on this thread.
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 
   bool send_allowed_;
 
@@ -109,8 +116,13 @@ class JingleThreadWrapper
   MessagesQueue messages_;
   std::list<PendingSend*> pending_send_messages_;
   base::WaitableEvent pending_send_event_;
+
+  base::WeakPtrFactory<JingleThreadWrapper> weak_ptr_factory_;
+  base::WeakPtr<JingleThreadWrapper> weak_ptr_;
+
+  DISALLOW_COPY_AND_ASSIGN(JingleThreadWrapper);
 };
 
-}
+}  // namespace jingle_glue
 
 #endif  // JINGLE_GLUE_THREAD_WRAPPER_H_

jingle/glue/thread_wrapper_unittest.cc

@@ -81,7 +81,7 @@ class ThreadWrapperTest : public testing::Test {
   }
 
   virtual void SetUp() OVERRIDE {
-    JingleThreadWrapper::EnsureForCurrentThread();
+    JingleThreadWrapper::EnsureForCurrentMessageLoop();
     thread_ = talk_base::Thread::Current();
   }
 
@@ -230,7 +230,7 @@ TEST_F(ThreadWrapperTest, SendSameThread) {
 
 void InitializeWrapperForNewThread(talk_base::Thread** thread,
                                    base::WaitableEvent* done_event) {
-  JingleThreadWrapper::EnsureForCurrentThread();
+  JingleThreadWrapper::EnsureForCurrentMessageLoop();
   JingleThreadWrapper::current()->set_send_allowed(true);
   *thread = JingleThreadWrapper::current();
   done_event->Signal();

remoting/client/plugin/chromoting_instance.cc

@@ -467,7 +467,7 @@ void ChromotingInstance::OnFirstFrameReceived() {
 void ChromotingInstance::Connect(const ClientConfig& config) {
   DCHECK(plugin_task_runner_->BelongsToCurrentThread());
 
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
 
   host_connection_.reset(new protocol::ConnectionToHost(true));
   scoped_ptr<AudioPlayer> audio_player(new PepperAudioPlayer(this));

remoting/protocol/libjingle_transport_factory.cc

@@ -312,7 +312,7 @@ LibjingleTransportFactory::LibjingleTransportFactory(
     : http_port_allocator_(port_allocator.get()),
       port_allocator_(port_allocator.Pass()),
       incoming_only_(incoming_only) {
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
 }
 
 LibjingleTransportFactory::LibjingleTransportFactory()
@@ -322,7 +322,7 @@ LibjingleTransportFactory::LibjingleTransportFactory()
       port_allocator_(new cricket::BasicPortAllocator(
           network_manager_.get(), socket_factory_.get())),
       incoming_only_(false) {
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentThread();
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
   port_allocator_->set_flags(
       cricket::PORTALLOCATOR_DISABLE_TCP |
       cricket::PORTALLOCATOR_DISABLE_STUN |