Cleanup: Replace MessageQueue pointers with Thread pointers.

This is part of a CL series merging rtc::MessageQueue into rtc::Thread.

Bug: webrtc:9883
Change-Id: I4a1bcd44c9523b6402b3f05b50597bdc2e6615e3
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/165345
Reviewed-by: Karl Wiberg <kwiberg@webrtc.org>
Reviewed-by: Steve Anton <steveanton@webrtc.org>
Commit-Queue: Sebastian Jansson <srte@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#30216}

examples/peerconnection/client/linux/main.cc

@@ -18,7 +18,6 @@
 #include "examples/peerconnection/client/flag_defs.h"
 #include "examples/peerconnection/client/linux/main_wnd.h"
 #include "examples/peerconnection/client/peer_connection_client.h"
-#include "rtc_base/message_queue.h"
 #include "rtc_base/physical_socket_server.h"
 #include "rtc_base/ref_counted_object.h"
 #include "rtc_base/ssl_adapter.h"
@@ -32,9 +31,7 @@ class CustomSocketServer : public rtc::PhysicalSocketServer {
       : wnd_(wnd), conductor_(NULL), client_(NULL) {}
   virtual ~CustomSocketServer() {}
 
-  void SetMessageQueue(rtc::MessageQueue* queue) override {
-    message_queue_ = queue;
-  }
+  void SetMessageQueue(rtc::Thread* queue) override { message_queue_ = queue; }
 
   void set_client(PeerConnectionClient* client) { client_ = client; }
   void set_conductor(Conductor* conductor) { conductor_ = conductor; }
@@ -58,7 +55,7 @@ class CustomSocketServer : public rtc::PhysicalSocketServer {
   }
 
  protected:
-  rtc::MessageQueue* message_queue_;
+  rtc::Thread* message_queue_;
   GtkMainWnd* wnd_;
   Conductor* conductor_;
   PeerConnectionClient* client_;

rtc_base/firewall_socket_server.cc

@@ -210,7 +210,7 @@ AsyncSocket* FirewallSocketServer::CreateAsyncSocket(int family, int type) {
   return WrapSocket(server_->CreateAsyncSocket(family, type), type);
 }
 
-void FirewallSocketServer::SetMessageQueue(MessageQueue* queue) {
+void FirewallSocketServer::SetMessageQueue(Thread* queue) {
   server_->SetMessageQueue(queue);
 }
 

rtc_base/firewall_socket_server.h

@@ -80,7 +80,7 @@ class FirewallSocketServer : public SocketServer {
   Socket* CreateSocket(int family, int type) override;
   AsyncSocket* CreateAsyncSocket(int family, int type) override;
 
-  void SetMessageQueue(MessageQueue* queue) override;
+  void SetMessageQueue(Thread* queue) override;
   bool Wait(int cms, bool process_io) override;
   void WakeUp() override;
 

rtc_base/message_queue.cc

@@ -162,7 +162,6 @@ MessageQueue::MessageQueue(SocketServer* ss, bool init_queue)
   // server, and provide it to the MessageQueue, since the Thread controls
   // the I/O model, and MQ is agnostic to those details.  Anyway, this causes
   // messagequeue_unittest to depend on network libraries... yuck.
-  ss_->SetMessageQueue(this);
   if (init_queue) {
     DoInit();
   }

rtc_base/nat_socket_factory.cc

@@ -388,7 +388,7 @@ AsyncSocket* NATSocketServer::CreateAsyncSocket(int family, int type) {
   return new NATSocket(this, family, type);
 }
 
-void NATSocketServer::SetMessageQueue(MessageQueue* queue) {
+void NATSocketServer::SetMessageQueue(Thread* queue) {
   msg_queue_ = queue;
   server_->SetMessageQueue(queue);
 }

rtc_base/nat_socket_factory.h

@@ -19,13 +19,13 @@
 
 #include "rtc_base/async_socket.h"
 #include "rtc_base/constructor_magic.h"
-#include "rtc_base/message_queue.h"
 #include "rtc_base/nat_server.h"
 #include "rtc_base/nat_types.h"
 #include "rtc_base/socket.h"
 #include "rtc_base/socket_address.h"
 #include "rtc_base/socket_factory.h"
 #include "rtc_base/socket_server.h"
+#include "rtc_base/thread.h"
 
 namespace rtc {
 
@@ -138,7 +138,7 @@ class NATSocketServer : public SocketServer, public NATInternalSocketFactory {
   explicit NATSocketServer(SocketServer* ss);
 
   SocketServer* socketserver() { return server_; }
-  MessageQueue* queue() { return msg_queue_; }
+  Thread* queue() { return msg_queue_; }
 
   Translator* GetTranslator(const SocketAddress& ext_ip);
   Translator* AddTranslator(const SocketAddress& ext_ip,
@@ -150,7 +150,7 @@ class NATSocketServer : public SocketServer, public NATInternalSocketFactory {
   Socket* CreateSocket(int family, int type) override;
   AsyncSocket* CreateAsyncSocket(int family, int type) override;
 
-  void SetMessageQueue(MessageQueue* queue) override;
+  void SetMessageQueue(Thread* queue) override;
   bool Wait(int cms, bool process_io) override;
   void WakeUp() override;
 
@@ -162,7 +162,7 @@ class NATSocketServer : public SocketServer, public NATInternalSocketFactory {
 
  private:
   SocketServer* server_;
-  MessageQueue* msg_queue_;
+  Thread* msg_queue_;
   TranslatorMap nats_;
   RTC_DISALLOW_COPY_AND_ASSIGN(NATSocketServer);
 };

rtc_base/socket_server.h

@@ -17,7 +17,7 @@
 
 namespace rtc {
 
-class MessageQueue;
+class Thread;
 // Needs to be forward declared because there's a circular dependency between
 // NetworkMonitor and Thread.
 // TODO(deadbeef): Fix this.
@@ -36,7 +36,7 @@ class SocketServer : public SocketFactory {
   // When the socket server is installed into a Thread, this function is
   // called to allow the socket server to use the thread's message queue for
   // any messaging that it might need to perform.
-  virtual void SetMessageQueue(MessageQueue* queue) {}
+  virtual void SetMessageQueue(Thread* queue) {}
 
   // Sleeps until:
   //  1) cms milliseconds have elapsed (unless cms == kForever)

rtc_base/thread.cc

@@ -181,6 +181,7 @@ Thread::Thread(std::unique_ptr<SocketServer> ss)
 
 Thread::Thread(SocketServer* ss, bool do_init)
     : MessageQueue(ss, /*do_init=*/false) {
+  socketserver()->SetMessageQueue(this);
   SetName("Thread", this);  // default name
   if (do_init) {
     DoInit();
@@ -189,6 +190,7 @@ Thread::Thread(SocketServer* ss, bool do_init)
 
 Thread::Thread(std::unique_ptr<SocketServer> ss, bool do_init)
     : MessageQueue(std::move(ss), false) {
+  socketserver()->SetMessageQueue(this);
   SetName("Thread", this);  // default name
   if (do_init) {
     DoInit();

rtc_base/virtual_socket_server.cc

@@ -599,7 +599,7 @@ VirtualSocket* VirtualSocketServer::CreateSocketInternal(int family, int type) {
   return socket;
 }
 
-void VirtualSocketServer::SetMessageQueue(MessageQueue* msg_queue) {
+void VirtualSocketServer::SetMessageQueue(Thread* msg_queue) {
   msg_queue_ = msg_queue;
   if (msg_queue_) {
     msg_queue_->SignalQueueDestroyed.connect(
@@ -614,7 +614,7 @@ bool VirtualSocketServer::Wait(int cmsWait, bool process_io) {
   }
   // Note: we don't need to do anything with |process_io| since we don't have
   // any real I/O. Received packets come in the form of queued messages, so
-  // MessageQueue will ensure WakeUp is called if another thread sends a
+  // Thread will ensure WakeUp is called if another thread sends a
   // packet.
   wakeup_.Wait(cmsWait);
   return true;

rtc_base/virtual_socket_server.h

@@ -107,7 +107,7 @@ class VirtualSocketServer : public SocketServer, public sigslot::has_slots<> {
   AsyncSocket* CreateAsyncSocket(int family, int type) override;
 
   // SocketServer:
-  void SetMessageQueue(MessageQueue* queue) override;
+  void SetMessageQueue(Thread* queue) override;
   bool Wait(int cms, bool process_io) override;
   void WakeUp() override;
 
@@ -267,7 +267,7 @@ class VirtualSocketServer : public SocketServer, public sigslot::has_slots<> {
 
   // Used to implement Wait/WakeUp.
   Event wakeup_;
-  MessageQueue* msg_queue_;
+  Thread* msg_queue_;
   bool stop_on_idle_;
   in_addr next_ipv4_;
   in6_addr next_ipv6_;

rtc_base/win32_socket_server.cc

@@ -698,7 +698,7 @@ AsyncSocket* Win32SocketServer::CreateAsyncSocket(int family, int type) {
   return nullptr;
 }
 
-void Win32SocketServer::SetMessageQueue(MessageQueue* queue) {
+void Win32SocketServer::SetMessageQueue(Thread* queue) {
   message_queue_ = queue;
 }
 

rtc_base/win32_socket_server.h

@@ -14,7 +14,6 @@
 #if defined(WEBRTC_WIN)
 #include "rtc_base/async_socket.h"
 #include "rtc_base/critical_section.h"
-#include "rtc_base/message_queue.h"
 #include "rtc_base/socket.h"
 #include "rtc_base/socket_factory.h"
 #include "rtc_base/socket_server.h"
@@ -103,7 +102,7 @@ class Win32SocketServer : public SocketServer {
   Socket* CreateSocket(int family, int type) override;
   AsyncSocket* CreateAsyncSocket(int family, int type) override;
 
-  void SetMessageQueue(MessageQueue* queue) override;
+  void SetMessageQueue(Thread* queue) override;
   bool Wait(int cms, bool process_io) override;
   void WakeUp() override;
 
@@ -122,7 +121,7 @@ class Win32SocketServer : public SocketServer {
   };
 
   static const wchar_t kWindowName[];
-  MessageQueue* message_queue_;
+  Thread* message_queue_;
   MessageWindow wnd_;
   CriticalSection cs_;
   bool posted_;

sdk/android/src/jni/pc/peer_connection_factory.cc

@@ -68,10 +68,10 @@ std::unique_ptr<T> TakeOwnershipOfUniquePtr(jlong native_pointer) {
 
 typedef void (*JavaMethodPointer)(JNIEnv*, const JavaRef<jobject>&);
 
-// Post a message on the given queue that will call the Java method on the given
-// Java object.
+// Post a message on the given thread that will call the Java method on the
+// given Java object.
 void PostJavaCallback(JNIEnv* env,
-                      rtc::MessageQueue* queue,
+                      rtc::Thread* queue,
                       const rtc::Location& posted_from,
                       const JavaRef<jobject>& j_object,
                       JavaMethodPointer java_method_pointer) {

test/network/fake_network_socket_server.cc

@@ -319,7 +319,7 @@ rtc::AsyncSocket* FakeNetworkSocketServer::CreateAsyncSocket(int family,
   return out;
 }
 
-void FakeNetworkSocketServer::SetMessageQueue(rtc::MessageQueue* msg_queue) {
+void FakeNetworkSocketServer::SetMessageQueue(rtc::Thread* msg_queue) {
   msg_queue_ = msg_queue;
   if (msg_queue_) {
     msg_queue_->SignalQueueDestroyed.connect(

test/network/fake_network_socket_server.h

@@ -18,7 +18,6 @@
 #include "rtc_base/async_socket.h"
 #include "rtc_base/critical_section.h"
 #include "rtc_base/event.h"
-#include "rtc_base/message_queue.h"
 #include "rtc_base/socket_server.h"
 #include "rtc_base/third_party/sigslot/sigslot.h"
 #include "system_wrappers/include/clock.h"
@@ -47,7 +46,7 @@ class FakeNetworkSocketServer : public rtc::SocketServer,
   // rtc::SocketServer methods:
   // Called by the network thread when this server is installed, kicking off the
   // message handler loop.
-  void SetMessageQueue(rtc::MessageQueue* msg_queue) override;
+  void SetMessageQueue(rtc::Thread* msg_queue) override;
   bool Wait(int cms, bool process_io) override;
   void WakeUp() override;
 
@@ -57,7 +56,7 @@ class FakeNetworkSocketServer : public rtc::SocketServer,
   Clock* const clock_;
   const EndpointsContainer* endpoints_container_;
   rtc::Event wakeup_;
-  rtc::MessageQueue* msg_queue_;
+  rtc::Thread* msg_queue_;
 
   rtc::CriticalSection lock_;
   std::vector<FakeNetworkSocket*> sockets_ RTC_GUARDED_BY(lock_);