Revert 148418 - Always use chromium threads for IO in remoting host

BUG=137140


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

TBR=sergeyu@chromium.org
Review URL: https://chromiumcodereview.appspot.com/10829040

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

11 files changed, 1597 insertions, 40 deletions

diff --git a/remoting/jingle_glue/jingle_thread.cc b/remoting/jingle_glue/jingle_thread.cc
new file mode 100644
index 0000000..40161dc
--- /dev/null
+++ b/remoting/jingle_glue/jingle_thread.cc
@@ -0,0 +1,204 @@
+// 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 "remoting/jingle_glue/jingle_thread.h"
+
+#include "base/basictypes.h"
+#include "base/logging.h"
+#include "base/message_loop_proxy.h"
+#include "base/message_pump.h"
+#include "base/time.h"
+#include "third_party/libjingle/source/talk/base/ssladapter.h"
+
+namespace remoting {
+
+const uint32 kRunTasksMessageId = 1;
+const uint32 kStopMessageId = 2;
+
+namespace {
+
+class JingleMessagePump : public base::MessagePump,
+                          public talk_base::MessageHandler {
+ public:
+  JingleMessagePump(talk_base::Thread* thread)
+      : thread_(thread), delegate_(NULL), stopping_(false) {
+  }
+
+  virtual void Run(Delegate* delegate) {
+    delegate_ = delegate;
+
+    thread_->Thread::Run();
+    // Call Restart() so that we can run again.
+    thread_->Restart();
+
+    delegate_ = NULL;
+  }
+
+  virtual void Quit() {
+    if (!stopping_) {
+      stopping_ = true;
+
+      // Shutdown gracefully: make sure that we excute all messages
+      // left in the queue before exiting. Thread::Quit() would not do
+      // that.
+      thread_->Post(this, kStopMessageId);
+    }
+  }
+
+  virtual void ScheduleWork() {
+    thread_->Post(this, kRunTasksMessageId);
+  }
+
+  virtual void ScheduleDelayedWork(const base::TimeTicks& time) {
+    delayed_work_time_ = time;
+    ScheduleNextDelayedTask();
+  }
+
+  void OnMessage(talk_base::Message* msg) {
+    if (msg->message_id == kRunTasksMessageId) {
+      DCHECK(delegate_);
+
+      // Clear currently pending messages in case there were delayed tasks.
+      // Will schedule it again from ScheduleNextDelayedTask() if neccessary.
+      thread_->Clear(this, kRunTasksMessageId);
+
+      // Process all pending tasks.
+      while (true) {
+        if (delegate_->DoWork())
+          continue;
+        if (delegate_->DoDelayedWork(&delayed_work_time_))
+          continue;
+        if (delegate_->DoIdleWork())
+          continue;
+        break;
+      }
+
+      ScheduleNextDelayedTask();
+    } else if (msg->message_id == kStopMessageId) {
+      DCHECK(stopping_);
+      // Stop the thread only if there are no more non-delayed
+      // messages left in the queue, otherwise post another task to
+      // try again later.
+      int delay = thread_->GetDelay();
+      if (delay > 0 || delay == talk_base::kForever) {
+        stopping_ = false;
+        thread_->Quit();
+      } else {
+        thread_->Post(this, kStopMessageId);
+      }
+    } else {
+      NOTREACHED();
+    }
+  }
+
+ protected:
+  virtual ~JingleMessagePump() {}
+
+ private:
+  void ScheduleNextDelayedTask() {
+    if (!delayed_work_time_.is_null()) {
+      base::TimeTicks now = base::TimeTicks::Now();
+      int delay = static_cast<int>((delayed_work_time_ - now).InMilliseconds());
+      if (delay > 0) {
+        thread_->PostDelayed(delay, this, kRunTasksMessageId);
+      } else {
+        thread_->Post(this, kRunTasksMessageId);
+      }
+    }
+  }
+
+  talk_base::Thread* thread_;
+  Delegate* delegate_;
+  base::TimeTicks delayed_work_time_;
+  bool stopping_;
+};
+
+}  // namespace
+
+JingleThreadMessageLoop::JingleThreadMessageLoop(talk_base::Thread* thread)
+    : MessageLoop(MessageLoop::TYPE_IO) {
+  pump_ = new JingleMessagePump(thread);
+}
+
+JingleThreadMessageLoop::~JingleThreadMessageLoop() {
+}
+
+TaskPump::TaskPump() {
+}
+
+void TaskPump::WakeTasks() {
+  talk_base::Thread::Current()->Post(this);
+}
+
+int64 TaskPump::CurrentTime() {
+  return static_cast<int64>(talk_base::Time());
+}
+
+void TaskPump::OnMessage(talk_base::Message* pmsg) {
+  RunTasks();
+}
+
+JingleThread::JingleThread()
+    : task_pump_(NULL),
+      started_event_(true, false),
+      stopped_event_(true, false),
+      message_loop_(NULL) {
+}
+
+JingleThread::~JingleThread() {
+  // It is important to call Stop here. If we wait for the base class to
+  // call Stop in its d'tor, then JingleThread::Run() will access member
+  // variables that are already gone. See similar comments in
+  // base/threading/thread.h.
+  if (message_loop_)
+    Stop();
+}
+
+bool JingleThread::Start() {
+  if (!Thread::Start())
+    return false;
+  started_event_.Wait();
+  return true;
+}
+
+void JingleThread::Run() {
+  JingleThreadMessageLoop message_loop(this);
+  message_loop_ = &message_loop;
+  message_loop_proxy_ = base::MessageLoopProxy::current();
+
+  TaskPump task_pump;
+  task_pump_ = &task_pump;
+
+  // Signal after we've initialized |message_loop_| and |task_pump_|.
+  started_event_.Signal();
+
+  message_loop.Run();
+
+  stopped_event_.Signal();
+
+  task_pump_ = NULL;
+  message_loop_ = NULL;
+}
+
+void JingleThread::Stop() {
+  message_loop_->PostTask(FROM_HERE, MessageLoop::QuitClosure());
+  stopped_event_.Wait();
+
+  // This will wait until the thread is actually finished.
+  Thread::Stop();
+}
+
+MessageLoop* JingleThread::message_loop() {
+  return message_loop_;
+}
+
+base::MessageLoopProxy* JingleThread::message_loop_proxy() {
+  return message_loop_proxy_;
+}
+
+TaskPump* JingleThread::task_pump() {
+  return task_pump_;
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/jingle_thread.h b/remoting/jingle_glue/jingle_thread.h
new file mode 100644
index 0000000..2c5422e
--- /dev/null
+++ b/remoting/jingle_glue/jingle_thread.h
@@ -0,0 +1,77 @@
+// 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.
+
+#ifndef REMOTING_JINGLE_GLUE_JINGLE_THREAD_H_
+#define REMOTING_JINGLE_GLUE_JINGLE_THREAD_H_
+
+#include "base/message_loop.h"
+#include "base/tracked_objects.h"
+#include "base/synchronization/waitable_event.h"
+#include "third_party/libjingle/source/talk/base/messagequeue.h"
+#include "third_party/libjingle/source/talk/base/taskrunner.h"
+#include "third_party/libjingle/source/talk/base/thread.h"
+
+namespace base {
+class MessageLoopProxy;
+}  // namespace base
+
+namespace remoting {
+
+class TaskPump : public talk_base::MessageHandler,
+                 public talk_base::TaskRunner {
+ public:
+  TaskPump();
+
+  // TaskRunner methods.
+  virtual void WakeTasks() OVERRIDE;
+  virtual int64 CurrentTime() OVERRIDE;
+
+  // MessageHandler methods.
+  virtual void OnMessage(talk_base::Message* pmsg) OVERRIDE;
+};
+
+class JingleThreadMessageLoop : public MessageLoop {
+ public:
+  explicit JingleThreadMessageLoop(talk_base::Thread* thread);
+  virtual ~JingleThreadMessageLoop();
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(JingleThreadMessageLoop);
+};
+
+// TODO(sergeyu): This class should be changed to inherit from Chromiums
+// base::Thread instead of libjingle's thread.
+class JingleThread : public talk_base::Thread {
+ public:
+  JingleThread();
+  virtual ~JingleThread();
+
+  bool Start();
+
+  // Main function for the thread. Should not be called directly.
+  virtual void Run() OVERRIDE;
+
+  // Stop the thread.
+  virtual void Stop() OVERRIDE;
+
+  // Returns Chromiums message loop for this thread.
+  MessageLoop* message_loop();
+  base::MessageLoopProxy* message_loop_proxy();
+
+  // Returns task pump if the thread is running, otherwise NULL is returned.
+  TaskPump* task_pump();
+
+ private:
+  TaskPump* task_pump_;
+  base::WaitableEvent started_event_;
+  base::WaitableEvent stopped_event_;
+  MessageLoop* message_loop_;
+  scoped_refptr<base::MessageLoopProxy> message_loop_proxy_;
+
+  DISALLOW_COPY_AND_ASSIGN(JingleThread);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_JINGLE_THREAD_H_
diff --git a/remoting/jingle_glue/jingle_thread_unittest.cc b/remoting/jingle_glue/jingle_thread_unittest.cc
new file mode 100644
index 0000000..e73c8b2
--- /dev/null
+++ b/remoting/jingle_glue/jingle_thread_unittest.cc
@@ -0,0 +1,92 @@
+// 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 "base/bind.h"
+#include "base/bind_helpers.h"
+#include "base/message_loop.h"
+#include "base/time.h"
+#include "base/synchronization/waitable_event.h"
+#include "remoting/jingle_glue/jingle_thread.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace remoting {
+
+class MockCallback {
+ public:
+  MOCK_METHOD0(Run, void());
+};
+
+namespace {
+
+// Delay used to test delayed tasks. Shouldn't be too big, so that we don't
+// slow down the test, yet, should be big enough to be measurable.
+int kDelayMs = 50;  // 0.05 s.
+int kDelayTimeoutMs = 10000;  // 10 s.
+
+}  // namespace
+
+TEST(JingleThreadTest, PostTask) {
+  JingleThread thread;
+  MockCallback task;
+  EXPECT_CALL(task, Run());
+
+  thread.Start();
+  thread.message_loop()->PostTask(
+      FROM_HERE, base::Bind(&MockCallback::Run, base::Unretained(&task)));
+  thread.Stop();
+}
+
+TEST(JingleThreadTest, PostNonNestableTask) {
+  JingleThread thread;
+  MockCallback task;
+  EXPECT_CALL(task, Run());
+
+  thread.Start();
+  thread.message_loop()->PostNonNestableTask(
+      FROM_HERE, base::Bind(&MockCallback::Run, base::Unretained(&task)));
+  thread.Stop();
+}
+
+ACTION_P(SignalEvent, event) {
+  event->Signal();
+}
+
+TEST(JingleThreadTest, PostDelayedTask) {
+  JingleThread thread;
+  MockCallback task;
+  base::WaitableEvent event(true, false);
+  EXPECT_CALL(task, Run()).WillOnce(SignalEvent(&event));
+
+  thread.Start();
+  base::Time start = base::Time::Now();
+  thread.message_loop()->PostDelayedTask(
+      FROM_HERE, base::Bind(&MockCallback::Run, base::Unretained(&task)),
+      base::TimeDelta::FromMilliseconds(kDelayMs));
+  event.TimedWait(base::TimeDelta::FromMilliseconds(kDelayTimeoutMs));
+  base::Time end = base::Time::Now();
+  thread.Stop();
+
+  EXPECT_GE((end - start).InMillisecondsRoundedUp(), kDelayMs);
+}
+
+TEST(JingleThreadTest, PostNonNestableDelayedTask) {
+  JingleThread thread;
+  MockCallback task;
+  base::WaitableEvent event(true, false);
+  EXPECT_CALL(task, Run()).WillOnce(SignalEvent(&event));
+
+  thread.Start();
+  base::Time start = base::Time::Now();
+  thread.message_loop()->PostNonNestableDelayedTask(
+      FROM_HERE, base::Bind(&MockCallback::Run, base::Unretained(&task)),
+      base::TimeDelta::FromMilliseconds(kDelayMs));
+  event.TimedWait(base::TimeDelta::FromMilliseconds(kDelayTimeoutMs));
+  base::Time end = base::Time::Now();
+  thread.Stop();
+
+  EXPECT_GE((end - start).InMillisecondsRoundedUp(), kDelayMs);
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/ssl_adapter.cc b/remoting/jingle_glue/ssl_adapter.cc
new file mode 100644
index 0000000..5d84822
--- /dev/null
+++ b/remoting/jingle_glue/ssl_adapter.cc
@@ -0,0 +1,22 @@
+// 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 "remoting/jingle_glue/ssl_adapter.h"
+
+#if defined(OS_WIN)
+#include "third_party/libjingle/source/talk/base/ssladapter.h"
+#else
+#include "remoting/jingle_glue/ssl_socket_adapter.h"
+#endif
+
+namespace remoting {
+
+talk_base::SSLAdapter* CreateSSLAdapter(talk_base::AsyncSocket* socket) {
+  talk_base::SSLAdapter* ssl_adapter =
+      remoting::SSLSocketAdapter::Create(socket);
+  DCHECK(ssl_adapter);
+  return ssl_adapter;
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/ssl_adapter.h b/remoting/jingle_glue/ssl_adapter.h
new file mode 100644
index 0000000..56b03be
--- /dev/null
+++ b/remoting/jingle_glue/ssl_adapter.h
@@ -0,0 +1,33 @@
+// Copyright (c) 2009 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 REMOTING_JINGLE_GLUE_SSL_ADAPTER_H_
+#define REMOTING_JINGLE_GLUE_SSL_ADAPTER_H_
+
+namespace talk_base {
+class AsyncSocket;
+class SSLAdapter;
+}  // namespace talk_base
+
+namespace remoting {
+
+// Wraps the given socket in a platform-dependent SSLAdapter
+// implementation.
+talk_base::SSLAdapter* CreateSSLAdapter(talk_base::AsyncSocket* socket);
+
+// Utility template class that overrides CreateSSLAdapter() to use the
+// above function.
+template <class SocketFactory>
+class SSLAdapterSocketFactory : public SocketFactory {
+ public:
+  virtual talk_base::SSLAdapter* CreateSSLAdapter(
+      talk_base::AsyncSocket* socket) {
+    return ::remoting::CreateSSLAdapter(socket);
+  }
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_SSL_ADAPTER_H_
+
diff --git a/remoting/jingle_glue/ssl_socket_adapter.cc b/remoting/jingle_glue/ssl_socket_adapter.cc
new file mode 100644
index 0000000..08ba785
--- /dev/null
+++ b/remoting/jingle_glue/ssl_socket_adapter.cc
@@ -0,0 +1,467 @@
+// 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 "remoting/jingle_glue/ssl_socket_adapter.h"
+
+#include "base/base64.h"
+#include "base/compiler_specific.h"
+#include "base/message_loop.h"
+#include "jingle/glue/utils.h"
+#include "net/base/address_list.h"
+#include "net/base/cert_verifier.h"
+#include "net/base/host_port_pair.h"
+#include "net/base/net_errors.h"
+#include "net/base/ssl_config_service.h"
+#include "net/base/transport_security_state.h"
+#include "net/socket/client_socket_factory.h"
+#include "net/url_request/url_request_context.h"
+
+namespace remoting {
+
+SSLSocketAdapter* SSLSocketAdapter::Create(AsyncSocket* socket) {
+  return new SSLSocketAdapter(socket);
+}
+
+SSLSocketAdapter::SSLSocketAdapter(AsyncSocket* socket)
+    : SSLAdapter(socket),
+      ignore_bad_cert_(false),
+      cert_verifier_(net::CertVerifier::CreateDefault()),
+      transport_security_state_(new net::TransportSecurityState()),
+      ssl_state_(SSLSTATE_NONE),
+      read_pending_(false),
+      write_pending_(false) {
+  transport_socket_ = new TransportSocket(socket, this);
+}
+
+SSLSocketAdapter::~SSLSocketAdapter() {
+}
+
+int SSLSocketAdapter::StartSSL(const char* hostname, bool restartable) {
+  DCHECK(!restartable);
+  hostname_ = hostname;
+
+  if (socket_->GetState() != Socket::CS_CONNECTED) {
+    ssl_state_ = SSLSTATE_WAIT;
+    return 0;
+  } else {
+    return BeginSSL();
+  }
+}
+
+int SSLSocketAdapter::BeginSSL() {
+  if (!MessageLoop::current()) {
+    // Certificate verification is done via the Chrome message loop.
+    // Without this check, if we don't have a chrome message loop the
+    // SSL connection just hangs silently.
+    LOG(DFATAL) << "Chrome message loop (needed by SSL certificate "
+                << "verification) does not exist";
+    return net::ERR_UNEXPECTED;
+  }
+
+  // SSLConfigService is not thread-safe, and the default values for SSLConfig
+  // are correct for us, so we don't use the config service to initialize this
+  // object.
+  net::SSLConfig ssl_config;
+  net::SSLClientSocketContext context(
+      cert_verifier_.get(), NULL, transport_security_state_.get(), "");
+
+  transport_socket_->set_addr(talk_base::SocketAddress(hostname_, 0));
+  ssl_socket_.reset(
+      net::ClientSocketFactory::GetDefaultFactory()->CreateSSLClientSocket(
+          transport_socket_, net::HostPortPair(hostname_, 443), ssl_config,
+          context));
+
+  int result = ssl_socket_->Connect(
+      base::Bind(&SSLSocketAdapter::OnConnected, base::Unretained(this)));
+
+  if (result == net::ERR_IO_PENDING || result == net::OK) {
+    return 0;
+  } else {
+    LOG(ERROR) << "Could not start SSL: " << net::ErrorToString(result);
+    return result;
+  }
+}
+
+int SSLSocketAdapter::Send(const void* buf, size_t len) {
+  if (ssl_state_ == SSLSTATE_ERROR) {
+    SetError(EINVAL);
+    return -1;
+  }
+
+  if (ssl_state_ == SSLSTATE_NONE) {
+    // Propagate the call to underlying socket if SSL is not connected
+    // yet (connection is not encrypted until StartSSL() is called).
+    return AsyncSocketAdapter::Send(buf, len);
+  }
+
+  if (write_pending_) {
+    SetError(EWOULDBLOCK);
+    return -1;
+  }
+
+  write_buffer_ = new net::DrainableIOBuffer(new net::IOBuffer(len), len);
+  memcpy(write_buffer_->data(), buf, len);
+
+  DoWrite();
+
+  return len;
+}
+
+int SSLSocketAdapter::Recv(void* buf, size_t len) {
+  switch (ssl_state_) {
+    case SSLSTATE_NONE: {
+      return AsyncSocketAdapter::Recv(buf, len);
+    }
+
+    case SSLSTATE_WAIT: {
+      SetError(EWOULDBLOCK);
+      return -1;
+    }
+
+    case SSLSTATE_CONNECTED: {
+      if (read_pending_) {
+        SetError(EWOULDBLOCK);
+        return -1;
+      }
+
+      int bytes_read = 0;
+
+      // Process any data we have left from the previous read.
+      if (read_buffer_) {
+        int size = std::min(read_buffer_->RemainingCapacity(),
+                            static_cast<int>(len));
+        memcpy(buf, read_buffer_->data(), size);
+        read_buffer_->set_offset(read_buffer_->offset() + size);
+        if (!read_buffer_->RemainingCapacity())
+          read_buffer_ = NULL;
+
+        if (size == static_cast<int>(len))
+          return size;
+
+        // If we didn't fill the caller's buffer then dispatch a new
+        // Read() in case there's more data ready.
+        buf = reinterpret_cast<char*>(buf) + size;
+        len -= size;
+        bytes_read = size;
+        DCHECK(!read_buffer_);
+      }
+
+      // Dispatch a Read() request to the SSL layer.
+      read_buffer_ = new net::GrowableIOBuffer();
+      read_buffer_->SetCapacity(len);
+      int result = ssl_socket_->Read(
+          read_buffer_, len,
+          base::Bind(&SSLSocketAdapter::OnRead, base::Unretained(this)));
+      if (result >= 0)
+        memcpy(buf, read_buffer_->data(), len);
+
+      if (result == net::ERR_IO_PENDING) {
+        read_pending_ = true;
+        if (bytes_read) {
+          return bytes_read;
+        } else {
+          SetError(EWOULDBLOCK);
+          return -1;
+        }
+      }
+
+      if (result < 0) {
+        SetError(EINVAL);
+        ssl_state_ = SSLSTATE_ERROR;
+        LOG(ERROR) << "Error reading from SSL socket " << result;
+        return -1;
+      }
+      read_buffer_ = NULL;
+      return result + bytes_read;
+    }
+
+    case SSLSTATE_ERROR: {
+      SetError(EINVAL);
+      return -1;
+    }
+  }
+
+  NOTREACHED();
+  return -1;
+}
+
+void SSLSocketAdapter::OnConnected(int result) {
+  if (result == net::OK) {
+    ssl_state_ = SSLSTATE_CONNECTED;
+    OnConnectEvent(this);
+  } else {
+    LOG(WARNING) << "OnConnected failed with error " << result;
+  }
+}
+
+void SSLSocketAdapter::OnRead(int result) {
+  DCHECK(read_pending_);
+  read_pending_ = false;
+  if (result > 0) {
+    DCHECK_GE(read_buffer_->capacity(), result);
+    read_buffer_->SetCapacity(result);
+  } else {
+    if (result < 0)
+      ssl_state_ = SSLSTATE_ERROR;
+  }
+  AsyncSocketAdapter::OnReadEvent(this);
+}
+
+void SSLSocketAdapter::OnWritten(int result) {
+  DCHECK(write_pending_);
+  write_pending_ = false;
+  if (result >= 0) {
+    write_buffer_->DidConsume(result);
+    if (!write_buffer_->BytesRemaining()) {
+      write_buffer_ = NULL;
+    } else {
+      DoWrite();
+    }
+  } else {
+    ssl_state_ = SSLSTATE_ERROR;
+  }
+  AsyncSocketAdapter::OnWriteEvent(this);
+}
+
+void SSLSocketAdapter::DoWrite() {
+  DCHECK_GT(write_buffer_->BytesRemaining(), 0);
+  DCHECK(!write_pending_);
+
+  while (true) {
+    int result = ssl_socket_->Write(
+        write_buffer_, write_buffer_->BytesRemaining(),
+        base::Bind(&SSLSocketAdapter::OnWritten, base::Unretained(this)));
+
+    if (result > 0) {
+      write_buffer_->DidConsume(result);
+      if (!write_buffer_->BytesRemaining()) {
+        write_buffer_ = NULL;
+        return;
+      }
+      continue;
+    }
+
+    if (result == net::ERR_IO_PENDING) {
+      write_pending_ = true;
+    } else {
+      SetError(EINVAL);
+      ssl_state_ = SSLSTATE_ERROR;
+    }
+    return;
+  }
+}
+
+void SSLSocketAdapter::OnConnectEvent(talk_base::AsyncSocket* socket) {
+  if (ssl_state_ != SSLSTATE_WAIT) {
+    AsyncSocketAdapter::OnConnectEvent(socket);
+  } else {
+    ssl_state_ = SSLSTATE_NONE;
+    int result = BeginSSL();
+    if (0 != result) {
+      // TODO(zork): Handle this case gracefully.
+      LOG(WARNING) << "BeginSSL() failed with " << result;
+    }
+  }
+}
+
+TransportSocket::TransportSocket(talk_base::AsyncSocket* socket,
+                                 SSLSocketAdapter *ssl_adapter)
+    : read_buffer_len_(0),
+      write_buffer_len_(0),
+      socket_(socket),
+      was_used_to_convey_data_(false) {
+  socket_->SignalReadEvent.connect(this, &TransportSocket::OnReadEvent);
+  socket_->SignalWriteEvent.connect(this, &TransportSocket::OnWriteEvent);
+}
+
+TransportSocket::~TransportSocket() {
+}
+
+int TransportSocket::Connect(const net::CompletionCallback& callback) {
+  // Connect is never called by SSLClientSocket, instead SSLSocketAdapter
+  // calls Connect() on socket_ directly.
+  NOTREACHED();
+  return false;
+}
+
+void TransportSocket::Disconnect() {
+  socket_->Close();
+}
+
+bool TransportSocket::IsConnected() const {
+  return (socket_->GetState() == talk_base::Socket::CS_CONNECTED);
+}
+
+bool TransportSocket::IsConnectedAndIdle() const {
+  // Not implemented.
+  NOTREACHED();
+  return false;
+}
+
+int TransportSocket::GetPeerAddress(net::IPEndPoint* address) const {
+  talk_base::SocketAddress socket_address = socket_->GetRemoteAddress();
+  if (jingle_glue::SocketAddressToIPEndPoint(socket_address, address)) {
+    return net::OK;
+  } else {
+    return net::ERR_FAILED;
+  }
+}
+
+int TransportSocket::GetLocalAddress(net::IPEndPoint* address) const {
+  talk_base::SocketAddress socket_address = socket_->GetLocalAddress();
+  if (jingle_glue::SocketAddressToIPEndPoint(socket_address, address)) {
+    return net::OK;
+  } else {
+    return net::ERR_FAILED;
+  }
+}
+
+const net::BoundNetLog& TransportSocket::NetLog() const {
+  return net_log_;
+}
+
+void TransportSocket::SetSubresourceSpeculation() {
+  NOTREACHED();
+}
+
+void TransportSocket::SetOmniboxSpeculation() {
+  NOTREACHED();
+}
+
+bool TransportSocket::WasEverUsed() const {
+  // We don't use this in ClientSocketPools, so this should never be used.
+  NOTREACHED();
+  return was_used_to_convey_data_;
+}
+
+bool TransportSocket::UsingTCPFastOpen() const {
+  return false;
+}
+
+int64 TransportSocket::NumBytesRead() const {
+  NOTREACHED();
+  return -1;
+}
+
+base::TimeDelta TransportSocket::GetConnectTimeMicros() const {
+  NOTREACHED();
+  return base::TimeDelta::FromMicroseconds(-1);
+}
+
+bool TransportSocket::WasNpnNegotiated() const {
+  NOTREACHED();
+  return false;
+}
+
+net::NextProto TransportSocket::GetNegotiatedProtocol() const {
+  NOTREACHED();
+  return net::kProtoUnknown;
+}
+
+bool TransportSocket::GetSSLInfo(net::SSLInfo* ssl_info) {
+  NOTREACHED();
+  return false;
+}
+
+int TransportSocket::Read(net::IOBuffer* buf, int buf_len,
+                          const net::CompletionCallback& callback) {
+  DCHECK(buf);
+  DCHECK(read_callback_.is_null());
+  DCHECK(!read_buffer_.get());
+  int result = socket_->Recv(buf->data(), buf_len);
+  if (result < 0) {
+    result = net::MapSystemError(socket_->GetError());
+    if (result == net::ERR_IO_PENDING) {
+      read_callback_ = callback;
+      read_buffer_ = buf;
+      read_buffer_len_ = buf_len;
+    }
+  }
+  if (result != net::ERR_IO_PENDING)
+    was_used_to_convey_data_ = true;
+  return result;
+}
+
+int TransportSocket::Write(net::IOBuffer* buf, int buf_len,
+                           const net::CompletionCallback& callback) {
+  DCHECK(buf);
+  DCHECK(write_callback_.is_null());
+  DCHECK(!write_buffer_.get());
+  int result = socket_->Send(buf->data(), buf_len);
+  if (result < 0) {
+    result = net::MapSystemError(socket_->GetError());
+    if (result == net::ERR_IO_PENDING) {
+      write_callback_ = callback;
+      write_buffer_ = buf;
+      write_buffer_len_ = buf_len;
+    }
+  }
+  if (result != net::ERR_IO_PENDING)
+    was_used_to_convey_data_ = true;
+  return result;
+}
+
+bool TransportSocket::SetReceiveBufferSize(int32 size) {
+  // Not implemented.
+  return false;
+}
+
+bool TransportSocket::SetSendBufferSize(int32 size) {
+  // Not implemented.
+  return false;
+}
+
+void TransportSocket::OnReadEvent(talk_base::AsyncSocket* socket) {
+  if (!read_callback_.is_null()) {
+    DCHECK(read_buffer_.get());
+    net::CompletionCallback callback = read_callback_;
+    scoped_refptr<net::IOBuffer> buffer = read_buffer_;
+    int buffer_len = read_buffer_len_;
+
+    read_callback_.Reset();
+    read_buffer_ = NULL;
+    read_buffer_len_ = 0;
+
+    int result = socket_->Recv(buffer->data(), buffer_len);
+    if (result < 0) {
+      result = net::MapSystemError(socket_->GetError());
+      if (result == net::ERR_IO_PENDING) {
+        read_callback_ = callback;
+        read_buffer_ = buffer;
+        read_buffer_len_ = buffer_len;
+        return;
+      }
+    }
+    was_used_to_convey_data_ = true;
+    callback.Run(result);
+  }
+}
+
+void TransportSocket::OnWriteEvent(talk_base::AsyncSocket* socket) {
+  if (!write_callback_.is_null()) {
+    DCHECK(write_buffer_.get());
+    net::CompletionCallback callback = write_callback_;
+    scoped_refptr<net::IOBuffer> buffer = write_buffer_;
+    int buffer_len = write_buffer_len_;
+
+    write_callback_.Reset();
+    write_buffer_ = NULL;
+    write_buffer_len_ = 0;
+
+    int result = socket_->Send(buffer->data(), buffer_len);
+    if (result < 0) {
+      result = net::MapSystemError(socket_->GetError());
+      if (result == net::ERR_IO_PENDING) {
+        write_callback_ = callback;
+        write_buffer_ = buffer;
+        write_buffer_len_ = buffer_len;
+        return;
+      }
+    }
+    was_used_to_convey_data_ = true;
+    callback.Run(result);
+  }
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/ssl_socket_adapter.h b/remoting/jingle_glue/ssl_socket_adapter.h
new file mode 100644
index 0000000..e62c048
--- /dev/null
+++ b/remoting/jingle_glue/ssl_socket_adapter.h
@@ -0,0 +1,158 @@
+// 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.
+
+#ifndef REMOTING_JINGLE_GLUE_SSL_SOCKET_ADAPTER_H_
+#define REMOTING_JINGLE_GLUE_SSL_SOCKET_ADAPTER_H_
+
+#include "base/memory/scoped_ptr.h"
+#include "net/base/completion_callback.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "net/base/net_log.h"
+#include "net/socket/ssl_client_socket.h"
+#include "net/socket/stream_socket.h"
+#include "third_party/libjingle/source/talk/base/asyncsocket.h"
+#include "third_party/libjingle/source/talk/base/ssladapter.h"
+
+namespace net {
+class CertVerifier;
+class TransportSecurityState;
+}  // namespace net
+
+namespace remoting {
+
+class SSLSocketAdapter;
+
+// TODO(sergeyu): Write unittests for this code!
+
+// This class provides a wrapper to libjingle's talk_base::AsyncSocket that
+// implements Chromium's net::StreamSocket interface.  It's used by
+// SSLSocketAdapter to enable Chromium's SSL implementation to work over
+// libjingle's socket class.
+class TransportSocket : public net::StreamSocket, public sigslot::has_slots<> {
+ public:
+  TransportSocket(talk_base::AsyncSocket* socket,
+                  SSLSocketAdapter *ssl_adapter);
+  virtual ~TransportSocket();
+
+  void set_addr(const talk_base::SocketAddress& addr) {
+    addr_ = addr;
+  }
+
+  // net::StreamSocket implementation.
+  virtual int Connect(const net::CompletionCallback& callback) OVERRIDE;
+  virtual void Disconnect() OVERRIDE;
+  virtual bool IsConnected() const OVERRIDE;
+  virtual bool IsConnectedAndIdle() const OVERRIDE;
+  virtual int GetPeerAddress(net::IPEndPoint* address) const OVERRIDE;
+  virtual int GetLocalAddress(net::IPEndPoint* address) const OVERRIDE;
+  virtual const net::BoundNetLog& NetLog() const OVERRIDE;
+  virtual void SetSubresourceSpeculation() OVERRIDE;
+  virtual void SetOmniboxSpeculation() OVERRIDE;
+  virtual bool WasEverUsed() const OVERRIDE;
+  virtual bool UsingTCPFastOpen() const OVERRIDE;
+  virtual int64 NumBytesRead() const OVERRIDE;
+  virtual base::TimeDelta GetConnectTimeMicros() const OVERRIDE;
+  virtual bool WasNpnNegotiated() const OVERRIDE;
+  virtual net::NextProto GetNegotiatedProtocol() const OVERRIDE;
+  virtual bool GetSSLInfo(net::SSLInfo* ssl_info) OVERRIDE;
+
+  // net::Socket implementation.
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   const net::CompletionCallback& callback) OVERRIDE;
+  virtual int Write(net::IOBuffer* buf, int buf_len,
+                    const net::CompletionCallback& callback) OVERRIDE;
+  virtual bool SetReceiveBufferSize(int32 size) OVERRIDE;
+  virtual bool SetSendBufferSize(int32 size) OVERRIDE;
+
+ private:
+  friend class SSLSocketAdapter;
+
+  void OnReadEvent(talk_base::AsyncSocket* socket);
+  void OnWriteEvent(talk_base::AsyncSocket* socket);
+
+  // Holds the user's completion callback when Write and Read are called.
+  net::CompletionCallback read_callback_;
+  net::CompletionCallback write_callback_;
+
+  scoped_refptr<net::IOBuffer> read_buffer_;
+  int read_buffer_len_;
+  scoped_refptr<net::IOBuffer> write_buffer_;
+  int write_buffer_len_;
+
+  net::BoundNetLog net_log_;
+
+  talk_base::AsyncSocket *socket_;
+  talk_base::SocketAddress addr_;
+
+  bool was_used_to_convey_data_;
+
+  DISALLOW_COPY_AND_ASSIGN(TransportSocket);
+};
+
+// This provides a talk_base::AsyncSocketAdapter interface around Chromium's
+// net::SSLClientSocket class.  This allows remoting to use Chromium's SSL
+// implementation instead of OpenSSL.
+class SSLSocketAdapter : public talk_base::SSLAdapter {
+ public:
+  explicit SSLSocketAdapter(talk_base::AsyncSocket* socket);
+  virtual ~SSLSocketAdapter();
+
+  // StartSSL returns 0 if successful, or non-zero on failure.
+  // If StartSSL is called while the socket is closed or connecting, the SSL
+  // negotiation will begin as soon as the socket connects.
+  //
+  // restartable is not implemented, and must be set to false.
+  virtual int StartSSL(const char* hostname, bool restartable) OVERRIDE;
+
+  // Create the default SSL adapter for this platform.
+  static SSLSocketAdapter* Create(AsyncSocket* socket);
+
+  virtual int Send(const void* pv, size_t cb) OVERRIDE;
+  virtual int Recv(void* pv, size_t cb) OVERRIDE;
+
+ private:
+  friend class TransportSocket;
+
+  enum SSLState {
+    SSLSTATE_NONE,
+    SSLSTATE_WAIT,
+    SSLSTATE_CONNECTED,
+    SSLSTATE_ERROR,
+  };
+
+  void OnConnected(int result);
+  void OnRead(int result);
+  void OnWritten(int result);
+
+  void DoWrite();
+
+  virtual void OnConnectEvent(talk_base::AsyncSocket* socket) OVERRIDE;
+
+  int BeginSSL();
+
+  bool ignore_bad_cert_;
+  std::string hostname_;
+  TransportSocket* transport_socket_;
+
+  // |cert_verifier_| must be defined before |ssl_socket_|, so that
+  // it's destroyed after |ssl_socket_|.
+  scoped_ptr<net::CertVerifier> cert_verifier_;
+  scoped_ptr<net::TransportSecurityState> transport_security_state_;
+  scoped_ptr<net::SSLClientSocket> ssl_socket_;
+
+  SSLState ssl_state_;
+
+  bool read_pending_;
+  scoped_refptr<net::GrowableIOBuffer> read_buffer_;
+
+  bool write_pending_;
+  scoped_refptr<net::DrainableIOBuffer> write_buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(SSLSocketAdapter);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_SSL_SOCKET_ADAPTER_H_
diff --git a/remoting/jingle_glue/xmpp_signal_strategy.cc b/remoting/jingle_glue/xmpp_signal_strategy.cc
index 5179aa1..5969817 100644
--- a/remoting/jingle_glue/xmpp_signal_strategy.cc
+++ b/remoting/jingle_glue/xmpp_signal_strategy.cc
@@ -7,13 +7,10 @@
 #include "base/bind.h"
 #include "base/logging.h"
 #include "base/string_util.h"
-#include "jingle/glue/chrome_async_socket.h"
-#include "jingle/glue/task_pump.h"
-#include "jingle/glue/xmpp_client_socket_factory.h"
 #include "jingle/notifier/base/gaia_token_pre_xmpp_auth.h"
-#include "net/socket/client_socket_factory.h"
-#include "net/url_request/url_request_context_getter.h"
-#include "third_party/libjingle/source/talk/base/thread.h"
+#include "remoting/jingle_glue/jingle_thread.h"
+#include "remoting/jingle_glue/xmpp_socket_adapter.h"
+#include "third_party/libjingle/source/talk/base/asyncsocket.h"
 #include "third_party/libjingle/source/talk/xmpp/prexmppauth.h"
 #include "third_party/libjingle/source/talk/xmpp/saslcookiemechanism.h"
 
@@ -25,11 +22,6 @@ const char kDefaultResourceName[] = "chromoting";
 // connections that are idle for more than a minute.
 const int kKeepAliveIntervalSeconds = 50;
 
-// Read buffer size used by ChromeAsyncSocket for read and write buffers. Most
-// of XMPP messages are smaller than 4kB.
-const size_t kReadBufferSize = 4096;
-const size_t kWriteBufferSize = 4096;
-
 void DisconnectXmppClient(buzz::XmppClient* client) {
   client->Disconnect();
 }
@@ -38,12 +30,11 @@ void DisconnectXmppClient(buzz::XmppClient* client) {
 
 namespace remoting {
 
-XmppSignalStrategy::XmppSignalStrategy(
-    scoped_refptr<net::URLRequestContextGetter> request_context_getter,
-    const std::string& username,
-    const std::string& auth_token,
-    const std::string& auth_token_service)
-   : request_context_getter_(request_context_getter),
+XmppSignalStrategy::XmppSignalStrategy(JingleThread* jingle_thread,
+                                       const std::string& username,
+                                       const std::string& auth_token,
+                                       const std::string& auth_token_service)
+   : thread_(jingle_thread),
      username_(username),
      auth_token_(auth_token),
      auth_token_service_(auth_token_service),
@@ -73,15 +64,9 @@ void XmppSignalStrategy::Connect() {
   settings.set_auth_token(buzz::AUTH_MECHANISM_GOOGLE_TOKEN, auth_token_);
   settings.set_server(talk_base::SocketAddress("talk.google.com", 5222));
 
-  scoped_ptr<jingle_glue::XmppClientSocketFactory> socket_factory(
-      new jingle_glue::XmppClientSocketFactory(
-          net::ClientSocketFactory::GetDefaultFactory(),
-          net::SSLConfig(), request_context_getter_, false));
-  buzz::AsyncSocket* socket = new jingle_glue::ChromeAsyncSocket(
-    socket_factory.release(), kReadBufferSize, kWriteBufferSize);
+  buzz::AsyncSocket* socket = new XmppSocketAdapter(settings, false);
 
-  task_runner_.reset(new jingle_glue::TaskPump());
-  xmpp_client_ = new buzz::XmppClient(task_runner_.get());
+  xmpp_client_ = new buzz::XmppClient(thread_->task_pump());
   xmpp_client_->Connect(settings, "", socket, CreatePreXmppAuth(settings));
   xmpp_client_->SignalStateChange.connect(
       this, &XmppSignalStrategy::OnConnectionStateChanged);
diff --git a/remoting/jingle_glue/xmpp_signal_strategy.h b/remoting/jingle_glue/xmpp_signal_strategy.h
index e18019e..790b0bd 100644
--- a/remoting/jingle_glue/xmpp_signal_strategy.h
+++ b/remoting/jingle_glue/xmpp_signal_strategy.h
@@ -21,14 +21,6 @@
 #include "third_party/libjingle/source/talk/base/sigslot.h"
 #include "third_party/libjingle/source/talk/xmpp/xmppclient.h"
 
-namespace net {
-class URLRequestContextGetter;
-}  // namespace net
-
-namespace talk_base {
-class TaskRunner;
-}  // namespace talk_base
-
 namespace remoting {
 
 class JingleThread;
@@ -38,11 +30,10 @@ class XmppSignalStrategy : public base::NonThreadSafe,
                            public buzz::XmppStanzaHandler,
                            public sigslot::has_slots<> {
  public:
-  XmppSignalStrategy(
-      scoped_refptr<net::URLRequestContextGetter> request_context_getter,
-      const std::string& username,
-      const std::string& auth_token,
-      const std::string& auth_token_service);
+  XmppSignalStrategy(JingleThread* thread,
+                     const std::string& username,
+                     const std::string& auth_token,
+                     const std::string& auth_token_service);
   virtual ~XmppSignalStrategy();
 
   // SignalStrategy interface.
@@ -79,12 +70,12 @@ class XmppSignalStrategy : public base::NonThreadSafe,
 
   void SendKeepAlive();
 
-  scoped_refptr<net::URLRequestContextGetter> request_context_getter_;
+  JingleThread* thread_;
+
   std::string username_;
   std::string auth_token_;
   std::string auth_token_service_;
   std::string resource_name_;
-  scoped_ptr<talk_base::TaskRunner> task_runner_;
   buzz::XmppClient* xmpp_client_;
 
   State state_;
diff --git a/remoting/jingle_glue/xmpp_socket_adapter.cc b/remoting/jingle_glue/xmpp_socket_adapter.cc
new file mode 100644
index 0000000..de66c5f
--- /dev/null
+++ b/remoting/jingle_glue/xmpp_socket_adapter.cc
@@ -0,0 +1,439 @@
+// 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 "remoting/jingle_glue/xmpp_socket_adapter.h"
+
+#include <iomanip>
+#include <string>
+
+#include "base/logging.h"
+#include "remoting/jingle_glue/ssl_adapter.h"
+#include "third_party/libjingle/source/talk/base/byteorder.h"
+#include "third_party/libjingle/source/talk/base/common.h"
+#include "third_party/libjingle/source/talk/base/firewallsocketserver.h"
+#include "third_party/libjingle/source/talk/base/socketadapters.h"
+#include "third_party/libjingle/source/talk/base/ssladapter.h"
+#include "third_party/libjingle/source/talk/base/thread.h"
+#include "third_party/libjingle/source/talk/xmpp/xmppengine.h"
+
+namespace remoting {
+
+XmppSocketAdapter::XmppSocketAdapter(const buzz::XmppClientSettings& xcs,
+                                     bool allow_unverified_certs)
+  : state_(STATE_CLOSED),
+    error_(ERROR_NONE),
+    wsa_error_(0),
+    socket_(NULL),
+    protocol_(xcs.protocol()),
+    firewall_(false),
+    write_buffer_(NULL),
+    write_buffer_length_(0),
+    write_buffer_capacity_(0),
+    allow_unverified_certs_(allow_unverified_certs) {
+  proxy_.type = xcs.proxy();
+  proxy_.address.SetIP(xcs.proxy_host());
+  proxy_.address.SetPort(xcs.proxy_port());
+  proxy_.username = xcs.proxy_user();
+  proxy_.password = xcs.proxy_pass();
+}
+
+XmppSocketAdapter::~XmppSocketAdapter() {
+  FreeState();
+
+  // Clean up any previous socket - cannot delete socket on close because close
+  // happens during the child socket's stack callback.
+  if (socket_) {
+    delete socket_;
+    socket_ = NULL;
+  }
+}
+
+buzz::AsyncSocket::State XmppSocketAdapter::state() {
+  return state_;
+}
+
+buzz::AsyncSocket::Error XmppSocketAdapter::error() {
+  return error_;
+}
+
+int XmppSocketAdapter::GetError() {
+  return wsa_error_;
+}
+
+bool XmppSocketAdapter::FreeState() {
+  int code = 0;
+
+  // Clean up the socket.
+  if (socket_ && !(state_ == STATE_CLOSED || state_ == STATE_CLOSING)) {
+    code = socket_->Close();
+  }
+
+  delete[] write_buffer_;
+  write_buffer_ = NULL;
+  write_buffer_length_ = 0;
+  write_buffer_capacity_ = 0;
+
+  if (code) {
+    SetWSAError(code);
+    return false;
+  }
+  return true;
+}
+
+bool XmppSocketAdapter::Connect(const talk_base::SocketAddress& addr) {
+  if (state_ != STATE_CLOSED) {
+    SetError(ERROR_WRONGSTATE);
+    return false;
+  }
+
+  VLOG(1) << "XmppSocketAdapter::Connect(" << addr.ToString() << ")";
+
+  // Clean up any previous socket - cannot delete socket on close because close
+  // happens during the child socket's stack callback.
+  if (socket_) {
+    delete socket_;
+    socket_ = NULL;
+  }
+
+  talk_base::AsyncSocket* socket =
+      talk_base::Thread::Current()->socketserver()->CreateAsyncSocket(
+          SOCK_STREAM);
+  if (!socket) {
+    SetWSAError(WSA_NOT_ENOUGH_MEMORY);
+    return false;
+  }
+
+  if (firewall_) {
+    // TODO(sync): Change this to make WSAAsyncSockets support current thread
+    // socket server.
+    talk_base::FirewallSocketServer* fw =
+        static_cast<talk_base::FirewallSocketServer*>(
+            talk_base::Thread::Current()->socketserver());
+    socket = fw->WrapSocket(socket, SOCK_STREAM);
+  }
+
+  if (proxy_.type) {
+    talk_base::AsyncSocket* proxy_socket = 0;
+    if (proxy_.type == talk_base::PROXY_SOCKS5) {
+      proxy_socket = new talk_base::AsyncSocksProxySocket(
+          socket, proxy_.address, proxy_.username, proxy_.password);
+    } else {
+      // Note: we are trying unknown proxies as HTTPS currently.
+      proxy_socket = new talk_base::AsyncHttpsProxySocket(socket,
+          "chromoting", proxy_.address, proxy_.username,
+          proxy_.password);
+    }
+    if (!proxy_socket) {
+      SetWSAError(WSA_NOT_ENOUGH_MEMORY);
+      delete socket;
+      return false;
+    }
+    socket = proxy_socket;  // For our purposes the proxy is now the socket.
+  }
+
+  if (protocol_ == cricket::PROTO_SSLTCP) {
+    talk_base::AsyncSocket *fake_ssl_socket =
+        new talk_base::AsyncSSLSocket(socket);
+    if (!fake_ssl_socket) {
+      SetWSAError(WSA_NOT_ENOUGH_MEMORY);
+      delete socket;
+      return false;
+    }
+    socket = fake_ssl_socket;  // For our purposes the SSL socket is the socket.
+  }
+
+#if defined(FEATURE_ENABLE_SSL)
+  talk_base::SSLAdapter* ssl_adapter = remoting::CreateSSLAdapter(socket);
+  socket = ssl_adapter;  // For our purposes the SSL adapter is the socket.
+#endif
+
+  socket->SignalReadEvent.connect(this, &XmppSocketAdapter::OnReadEvent);
+  socket->SignalWriteEvent.connect(this, &XmppSocketAdapter::OnWriteEvent);
+  socket->SignalConnectEvent.connect(this, &XmppSocketAdapter::OnConnectEvent);
+  socket->SignalCloseEvent.connect(this, &XmppSocketAdapter::OnCloseEvent);
+
+  // The linux implementation of socket::Connect returns an error when the
+  // connect didn't complete yet.  This can be distinguished from a failure
+  // because socket::IsBlocking is true.  Perhaps, the linux implementation
+  // should be made to behave like the windows version which doesn't do this,
+  // but it seems to be a pattern with these methods that they return an error
+  // if the operation didn't complete in a sync fashion and one has to check
+  // IsBlocking to tell if was a "real" error.
+  if (socket->Connect(addr) == SOCKET_ERROR && !socket->IsBlocking()) {
+    SetWSAError(socket->GetError());
+    delete socket;
+    return false;
+  }
+
+  socket_ = socket;
+  state_ = STATE_CONNECTING;
+  return true;
+}
+
+bool XmppSocketAdapter::Read(char* data, size_t len, size_t* len_read) {
+  if (len_read)
+    *len_read = 0;
+
+  if (state_ <= STATE_CLOSING) {
+    SetError(ERROR_WRONGSTATE);
+    return false;
+  }
+
+  DCHECK(socket_);
+
+  if (IsOpen()) {
+    int result = socket_->Recv(data, len);
+    if (result < 0) {
+      if (!socket_->IsBlocking()) {
+        SetWSAError(socket_->GetError());
+        return false;
+      }
+
+      result = 0;
+    }
+
+    if (len_read)
+      *len_read = result;
+  }
+
+  return true;
+}
+
+bool XmppSocketAdapter::Write(const char* data, size_t len) {
+  if (state_ <= STATE_CLOSING) {
+    // There may be data in a buffer that gets lost.  Too bad!
+    SetError(ERROR_WRONGSTATE);
+    return false;
+  }
+
+  DCHECK(socket_);
+
+  size_t sent = 0;
+
+  // Try an immediate write when there is no buffer and we aren't in SSL mode
+  // or opening the connection.
+  if (write_buffer_length_ == 0 && IsOpen()) {
+    int result = socket_->Send(data, len);
+    if (result < 0) {
+      if (!socket_->IsBlocking()) {
+        SetWSAError(socket_->GetError());
+        return false;
+      }
+      result = 0;
+    }
+
+    sent = static_cast<size_t>(result);
+  }
+
+  // Buffer what we didn't send.
+  if (sent < len) {
+    QueueWriteData(data + sent, len - sent);
+  }
+
+  // Service the socket right away to push the written data out in SSL mode.
+  return HandleWritable();
+}
+
+bool XmppSocketAdapter::Close() {
+  if (state_ == STATE_CLOSING) {
+    return false;  // Avoid recursion, but not unexpected.
+  }
+  if (state_ == STATE_CLOSED) {
+    // In theory should not be trying to re-InternalClose.
+    SetError(ERROR_WRONGSTATE);
+    return false;
+  }
+
+  // TODO(sync): deal with flushing close (flush, don't do reads, clean ssl).
+
+  // If we've gotten to the point where we really do have a socket underneath
+  // then close it.  It should call us back to tell us it is closed, and
+  // NotifyClose will be called.  We indicate "closing" state so that we
+  // do not recusively try to keep closing the socket.
+  if (socket_) {
+    state_ = STATE_CLOSING;
+    socket_->Close();
+  }
+
+  // If we didn't get the callback, then we better make sure we signal
+  // closed.
+  if (state_ != STATE_CLOSED) {
+    // The socket was closed manually, not directly due to error.
+    if (error_ != ERROR_NONE) {
+      VLOG(1) << "XmppSocketAdapter::Close - previous Error: " << error_
+              << " WSAError: " << wsa_error_;
+      error_ = ERROR_NONE;
+      wsa_error_ = 0;
+    }
+    NotifyClose();
+  }
+  return true;
+}
+
+void XmppSocketAdapter::NotifyClose() {
+  if (state_ == STATE_CLOSED) {
+    SetError(ERROR_WRONGSTATE);
+  } else {
+    VLOG(1) << "XmppSocketAdapter::NotifyClose - Error: " << error_
+            << " WSAError: " << wsa_error_;
+    state_ = STATE_CLOSED;
+    SignalClosed();
+    FreeState();
+  }
+}
+
+void XmppSocketAdapter::OnConnectEvent(talk_base::AsyncSocket *socket) {
+  if (state_ == STATE_CONNECTING) {
+    state_ = STATE_OPEN;
+    VLOG(1) << "XmppSocketAdapter::OnConnectEvent - STATE_OPEN";
+    SignalConnected();
+#if defined(FEATURE_ENABLE_SSL)
+  } else if (state_ == STATE_TLS_CONNECTING) {
+    state_ = STATE_TLS_OPEN;
+    VLOG(1) << "XmppSocketAdapter::OnConnectEvent - STATE_TLS_OPEN";
+    SignalSSLConnected();
+    if (write_buffer_length_ > 0) {
+      HandleWritable();
+    }
+#endif  // defined(FEATURE_ENABLE_SSL)
+  } else {
+    LOG(DFATAL) << "unexpected XmppSocketAdapter::OnConnectEvent state: "
+                << state_;
+  }
+}
+
+void XmppSocketAdapter::OnReadEvent(talk_base::AsyncSocket *socket) {
+  HandleReadable();
+}
+
+void XmppSocketAdapter::OnWriteEvent(talk_base::AsyncSocket *socket) {
+  HandleWritable();
+}
+
+void XmppSocketAdapter::OnCloseEvent(talk_base::AsyncSocket *socket,
+                                     int error) {
+  VLOG(1) << "XmppSocketAdapter::OnCloseEvent(" << error << ")";
+  SetWSAError(error);
+  if (error == SOCKET_EACCES) {
+    SignalAuthenticationError();  // Proxy needs authentication.
+  }
+  NotifyClose();
+}
+
+#if defined(FEATURE_ENABLE_SSL)
+bool XmppSocketAdapter::StartTls(const std::string& verify_host_name) {
+  if (state_ != STATE_OPEN) {
+    SetError(ERROR_WRONGSTATE);
+    return false;
+  }
+
+  state_ = STATE_TLS_CONNECTING;
+
+  DCHECK_EQ(write_buffer_length_, 0U);
+
+  talk_base::SSLAdapter* ssl_adapter =
+      static_cast<talk_base::SSLAdapter*>(socket_);
+
+  if (allow_unverified_certs_) {
+    ssl_adapter->set_ignore_bad_cert(true);
+  }
+
+  if (ssl_adapter->StartSSL(verify_host_name.c_str(), false) != 0) {
+    state_ = STATE_OPEN;
+    SetError(ERROR_SSL);
+    return false;
+  }
+
+  return true;
+}
+#endif  // defined(FEATURE_ENABLE_SSL)
+
+void XmppSocketAdapter::QueueWriteData(const char* data, size_t len) {
+  // Expand buffer if needed.
+  if (write_buffer_length_ + len > write_buffer_capacity_) {
+    size_t new_capacity = 1024;
+    while (new_capacity < write_buffer_length_ + len) {
+      new_capacity = new_capacity * 2;
+    }
+    char* new_buffer = new char[new_capacity];
+    DCHECK_LE(write_buffer_length_, 64000U);
+    memcpy(new_buffer, write_buffer_, write_buffer_length_);
+    delete[] write_buffer_;
+    write_buffer_ = new_buffer;
+    write_buffer_capacity_ = new_capacity;
+  }
+
+  // Copy data into the end of buffer.
+  memcpy(write_buffer_ + write_buffer_length_, data, len);
+  write_buffer_length_ += len;
+}
+
+void XmppSocketAdapter::FlushWriteQueue(Error* error, int* wsa_error) {
+  DCHECK(error);
+  DCHECK(wsa_error);
+
+  size_t flushed = 0;
+  while (flushed < write_buffer_length_) {
+    int sent = socket_->Send(write_buffer_ + flushed,
+                             static_cast<int>(write_buffer_length_ - flushed));
+    if (sent < 0) {
+      if (!socket_->IsBlocking()) {
+        *error = ERROR_WINSOCK;
+        *wsa_error = socket_->GetError();
+      }
+      break;
+    }
+    flushed += static_cast<size_t>(sent);
+  }
+
+  // Remove flushed memory.
+  write_buffer_length_ -= flushed;
+  memmove(write_buffer_, write_buffer_ + flushed, write_buffer_length_);
+
+  // When everything is flushed, deallocate the buffer if it's gotten big.
+  if (write_buffer_length_ == 0) {
+    if (write_buffer_capacity_ > 8192) {
+      delete[] write_buffer_;
+      write_buffer_ = NULL;
+      write_buffer_capacity_ = 0;
+    }
+  }
+}
+
+void XmppSocketAdapter::SetError(Error error) {
+  if (error_ == ERROR_NONE) {
+    error_ = error;
+  }
+}
+
+void XmppSocketAdapter::SetWSAError(int error) {
+  if (error_ == ERROR_NONE && error != 0) {
+    error_ = ERROR_WINSOCK;
+    wsa_error_ = error;
+  }
+}
+
+bool XmppSocketAdapter::HandleReadable() {
+  if (!IsOpen())
+    return false;
+
+  SignalRead();
+  return true;
+}
+
+bool XmppSocketAdapter::HandleWritable() {
+  if (!IsOpen())
+    return false;
+
+  Error error = ERROR_NONE;
+  int wsa_error = 0;
+  FlushWriteQueue(&error, &wsa_error);
+  if (error != ERROR_NONE) {
+    Close();
+    return false;
+  }
+  return true;
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/xmpp_socket_adapter.h b/remoting/jingle_glue/xmpp_socket_adapter.h
new file mode 100644
index 0000000..8ee58d4
--- /dev/null
+++ b/remoting/jingle_glue/xmpp_socket_adapter.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 REMOTING_JINGLE_GLUE_XMPP_SOCKET_ADAPTER_H_
+#define REMOTING_JINGLE_GLUE_XMPP_SOCKET_ADAPTER_H_
+
+#include <string>
+
+#include "base/basictypes.h"
+#include "base/compiler_specific.h"
+#include "third_party/libjingle/source/talk/base/asyncsocket.h"
+#include "third_party/libjingle/source/talk/xmpp/asyncsocket.h"
+#include "third_party/libjingle/source/talk/xmpp/xmppclientsettings.h"
+#include "third_party/libjingle/source/talk/xmpp/xmppengine.h"
+
+#ifndef _WIN32
+// Additional errors used by us from Win32 headers.
+#define SEC_E_CERT_EXPIRED static_cast<int>(0x80090328L)
+#define WSA_NOT_ENOUGH_MEMORY ENOMEM
+#endif
+
+namespace remoting {
+
+class XmppSocketAdapter : public buzz::AsyncSocket,
+                          public sigslot::has_slots<> {
+ public:
+  XmppSocketAdapter(const buzz::XmppClientSettings& xcs,
+                    bool allow_unverified_certs);
+  virtual ~XmppSocketAdapter();
+
+  virtual State state() OVERRIDE;
+  virtual Error error() OVERRIDE;
+  virtual int GetError() OVERRIDE;
+
+  void set_firewall(bool firewall) { firewall_ = firewall; }
+
+  virtual bool Connect(const talk_base::SocketAddress& addr) OVERRIDE;
+  virtual bool Read(char* data, size_t len, size_t* len_read) OVERRIDE;
+  virtual bool Write(const char* data, size_t len) OVERRIDE;
+  virtual bool Close() OVERRIDE;
+
+#if defined(FEATURE_ENABLE_SSL)
+  virtual bool StartTls(const std::string& domainname) OVERRIDE;
+  bool IsOpen() const { return state_ == STATE_OPEN
+                            || state_ == STATE_TLS_OPEN; }
+#else
+  bool IsOpen() const { return state_ == STATE_OPEN; }
+#endif
+
+  sigslot::signal0<> SignalAuthenticationError;
+
+ private:
+  // Return false if the socket is closed.
+  bool HandleReadable();
+  bool HandleWritable();
+
+  State state_;
+  Error error_;
+  int wsa_error_;
+
+  talk_base::AsyncSocket* socket_;
+  cricket::ProtocolType protocol_;
+  talk_base::ProxyInfo proxy_;
+  bool firewall_;
+  char* write_buffer_;
+  size_t write_buffer_length_;
+  size_t write_buffer_capacity_;
+  bool allow_unverified_certs_;
+
+  bool FreeState();
+  void NotifyClose();
+
+  void OnReadEvent(talk_base::AsyncSocket* socket);
+  void OnWriteEvent(talk_base::AsyncSocket* socket);
+  void OnConnectEvent(talk_base::AsyncSocket* socket);
+  void OnCloseEvent(talk_base::AsyncSocket* socket, int error);
+
+  void QueueWriteData(const char* data, size_t len);
+  void FlushWriteQueue(Error* error, int* wsa_error);
+
+  void SetError(Error error);
+  void SetWSAError(int error);
+  DISALLOW_COPY_AND_ASSIGN(XmppSocketAdapter);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_XMPP_SOCKET_ADAPTER_H_