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

jingle/glue/thread_wrapper.cc

 // Copyright (c) 2012 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"
 
 #include "base/bind.h"
 #include "base/bind_helpers.h"
 #include "base/lazy_instance.h"
 #include "base/threading/thread_local.h"
@@ skipping 13 matching lines @@
   talk_base::Message message;
   base::WaitableEvent done_event;
 };
 
 base::LazyInstance<base::ThreadLocalPointer<JingleThreadWrapper> >
     g_jingle_thread_wrapper = LAZY_INSTANCE_INITIALIZER;
 
 // static
 void JingleThreadWrapper::EnsureForCurrentThread() {
   if (JingleThreadWrapper::current() == NULL) {
-    g_jingle_thread_wrapper.Get().Set(
-        new JingleThreadWrapper(MessageLoop::current()));
+    MessageLoop* message_loop = MessageLoop::current();

[Wez, 2012/08/09 23:30:25]

This will only work if called from a MessageLoop; if you used
ThreadTaskRunnerHandle::Get() then it'd work for any thread with a TaskRunner,
although you wouldn't be able to hook destruction.

[Sergey Ulanov, 2012/08/13 21:15:34]

Yes. That's intentional. We need it to add destruction observer. So after this
change there are two ways to use this class:

- JingleThreadWrapper::EnsureForCurrentThread() - this autodestroys the wrapper,
  but only works on threads with MessageLoop.
- Explicitly create and destroy JingleThreadWrapper - requires to delete the
wrapper explicitly, but allows to use this object on any thread.

[Wez, 2012/08/14 00:06:04]

Would it be useful to have a LeakyEnsureForCurrentTaskRunner()?

[Sergey Ulanov, 2012/08/14 01:02:14]

Don't think it's necessary. It's always possible to just create
JingleThreadWrapper without destroying it.

+    g_jingle_thread_wrapper.Get().Set(new JingleThreadWrapper(
+        message_loop->message_loop_proxy()));
+    message_loop->AddDestructionObserver(current());
   }
 
   DCHECK_EQ(talk_base::Thread::Current(), current());
 }
 
 // static
 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);
+  DCHECK(task_runner->BelongsToCurrentThread());
+  DCHECK(!talk_base::Thread::Current());
   talk_base::MessageQueueManager::Instance()->Add(this);
-  message_loop_->AddDestructionObserver(this);
-
   WrapCurrent();
 }
 
 JingleThreadWrapper::~JingleThreadWrapper() {
   Clear(NULL, talk_base::MQID_ANY, NULL);
 }
 
 void JingleThreadWrapper::WillDestroyCurrentMessageLoop() {
   DCHECK_EQ(talk_base::Thread::Current(), current());
   UnwrapCurrent();
   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;
 }
 
 void JingleThreadWrapper::Post(
     talk_base::MessageHandler* handler, uint32 message_id,
     talk_base::MessageData* data, bool time_sensitive) {
   PostTaskInternal(0, handler, message_id, data);
 }
@@ skipping 72 matching lines @@
 
   PendingSend pending_send(message);
   {
     base::AutoLock auto_lock(lock_);
     pending_send_messages_.push_back(&pending_send);
   }
 
   // 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,
+                                    base::Unretained(this)));

[Wez, 2012/08/09 23:30:25]

Should this be a WeakPtr?

[Sergey Ulanov, 2012/08/13 21:15:34]

Yes. Thanks for catching this. Fixed.

 
 
   while (!pending_send.done_event.IsSignaled()) {
     base::WaitableEvent* events[] = {&pending_send.done_event,
                                      &current_thread->pending_send_event_};
     size_t event = base::WaitableEvent::WaitMany(events, arraysize(events));
     DCHECK(event == 0 || event == 1);
 
     if (event == 1)
       current_thread->ProcessPendingSends();
@@ skipping 29 matching lines @@
   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,
-                            base::Bind(&JingleThreadWrapper::RunTask,
-                                       base::Unretained(this), task_id));
+    task_runner_->PostTask(FROM_HERE,
+                           base::Bind(&JingleThreadWrapper::RunTask,
+                                      base::Unretained(this), task_id));

[Wez, 2012/08/09 23:30:25]

WeakPtr?

[Sergey Ulanov, 2012/08/13 21:15:34]

Done.

   } 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,
+                                             base::Unretained(this), task_id),

[Wez, 2012/08/09 23:30:25]

WeakPtr?

[Sergey Ulanov, 2012/08/13 21:15:34]

Done.

+                                  base::TimeDelta::FromMilliseconds(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()) {
@@ skipping 58 matching lines @@
 
 void JingleThreadWrapper::Stop() {
   NOTREACHED();
 }
 
 void JingleThreadWrapper::Run() {
   NOTREACHED();
 }
 
 }  // namespace jingle_glue