Implemented basic support Chromotocol connection.

New code supports multiple PseudoTCP and UDP channels. Client and host still use old JjngleChannel for connection.
BUG=53986
TEST=None

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

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

9 files changed, 595 insertions, 47 deletions

diff --git a/remoting/jingle_glue/channel_socket_adapter.cc b/remoting/jingle_glue/channel_socket_adapter.cc
new file mode 100644
index 0000000..583eed1
--- /dev/null
+++ b/remoting/jingle_glue/channel_socket_adapter.cc
@@ -0,0 +1,171 @@
+// Copyright (c) 2010 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/channel_socket_adapter.h"
+
+#include <limits>
+
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "remoting/jingle_glue/utils.h"
+#include "third_party/libjingle/source/talk/p2p/base/transportchannel.h"
+
+namespace remoting {
+
+TransportChannelSocketAdapter::TransportChannelSocketAdapter(
+    cricket::TransportChannel* channel)
+    : channel_(channel),
+      read_pending_(false),
+      write_pending_(false),
+      closed_error_code_(net::OK) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(channel_);
+
+  channel_->SignalReadPacket.connect(
+      this, &TransportChannelSocketAdapter::OnNewPacket);
+  channel_->SignalWritableState.connect(
+      this, &TransportChannelSocketAdapter::OnWritableState);
+  channel_->SignalDestroyed.connect(
+      this, &TransportChannelSocketAdapter::OnChannelDestroyed);
+}
+
+TransportChannelSocketAdapter::~TransportChannelSocketAdapter() {
+}
+
+int TransportChannelSocketAdapter::Read(
+    net::IOBuffer* buf, int buffer_size, net::CompletionCallback* callback) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(buf);
+  CHECK(!read_pending_);
+
+  if (!channel_) {
+    DCHECK(closed_error_code_ != net::OK);
+    return closed_error_code_;
+  }
+
+  read_callback_ = callback;
+  read_buffer_ = buf;
+  read_buffer_size_ = buffer_size;
+  read_pending_ = true;
+
+  return net::ERR_IO_PENDING;
+}
+
+int TransportChannelSocketAdapter::Write(
+    net::IOBuffer* buffer, int buffer_size, net::CompletionCallback* callback) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(buffer);
+  CHECK(!write_pending_);
+
+  if (!channel_) {
+    DCHECK(closed_error_code_ != net::OK);
+    return closed_error_code_;
+  }
+
+  int result = channel_->SendPacket(buffer->data(), buffer_size);
+  if (result < 0) {
+    result = MapPosixToChromeError(channel_->GetError());
+    if (result == net::ERR_IO_PENDING) {
+      write_pending_ = true;
+      write_callback_ = callback;
+      write_buffer_ = buffer;
+      write_buffer_size_ = buffer_size;
+    }
+  }
+  return result;
+}
+
+bool TransportChannelSocketAdapter::SetReceiveBufferSize(int32 size) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+bool TransportChannelSocketAdapter::SetSendBufferSize(int32 size) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+void TransportChannelSocketAdapter::Close(int error_code) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+
+  if (!channel_)  // Already closed.
+    return;
+
+  DCHECK(error_code != net::OK);
+  closed_error_code_ = error_code;
+  channel_->SignalReadPacket.disconnect(this);
+  channel_->SignalDestroyed.disconnect(this);
+  channel_ = NULL;
+
+  if (read_pending_) {
+    net::CompletionCallback* callback = read_callback_;
+    read_pending_ = false;
+    read_buffer_ = NULL;
+    callback->Run(error_code);
+  }
+
+  if (write_pending_) {
+    net::CompletionCallback* callback = write_callback_;
+    write_pending_ = false;
+    write_buffer_ = NULL;
+    callback->Run(error_code);
+  }
+}
+
+void TransportChannelSocketAdapter::OnNewPacket(
+    cricket::TransportChannel* channel, const char* data, size_t data_size) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK_EQ(channel, channel_);
+  if (read_pending_) {
+    DCHECK(read_buffer_);
+    CHECK_LT(data_size, static_cast<size_t>(std::numeric_limits<int>::max()));
+
+    if (read_buffer_size_ < static_cast<int>(data_size)) {
+      LOG(WARNING) << "Data buffer is smaller than the received packet. "
+                   << "Dropping the data that doesn't fit.";
+      data_size = read_buffer_size_;
+    }
+
+    memcpy(read_buffer_->data(), data, data_size);
+
+    net::CompletionCallback* callback = read_callback_;
+    read_pending_ = false;
+    read_buffer_ = NULL;
+
+    callback->Run(data_size);
+  } else {
+    LOG(WARNING)
+        << "Data was received without a callback. Dropping the packet.";
+  }
+}
+
+void TransportChannelSocketAdapter::OnWritableState(
+    cricket::TransportChannel* channel) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  // Try to send the packet if there is a pending write.
+  if (write_pending_) {
+    int result = channel_->SendPacket(write_buffer_->data(),
+                                      write_buffer_size_);
+    if (result < 0)
+      result = MapPosixToChromeError(channel_->GetError());
+
+    if (result != net::ERR_IO_PENDING) {
+      net::CompletionCallback* callback = write_callback_;
+      write_pending_ = false;
+      write_buffer_ = NULL;
+      callback->Run(result);
+    }
+  }
+}
+
+void TransportChannelSocketAdapter::OnChannelDestroyed(
+    cricket::TransportChannel* channel) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK_EQ(channel, channel_);
+  Close(net::ERR_CONNECTION_ABORTED);
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/channel_socket_adapter.h b/remoting/jingle_glue/channel_socket_adapter.h
new file mode 100644
index 0000000..1b808f6
--- /dev/null
+++ b/remoting/jingle_glue/channel_socket_adapter.h
@@ -0,0 +1,67 @@
+// Copyright (c) 2010 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_CHANNEL_SOCKET_ADAPTER_H_
+#define REMOTING_JINGLE_GLUE_CHANNEL_SOCKET_ADAPTER_H_
+
+#include "net/socket/socket.h"
+#include "third_party/libjingle/source/talk/base/socketaddress.h"
+#include "third_party/libjingle/source/talk/base/sigslot.h"
+
+namespace cricket {
+class TransportChannel;
+}  // namespace cricket
+
+namespace remoting {
+
+// TransportChannelSocketAdapter implements net::Socket interface on
+// top of libjingle's TransportChannel. It is used by JingleChromotingConnection
+// to provide net::Socket interface for channels.
+class TransportChannelSocketAdapter : public net::Socket,
+                                      public sigslot::has_slots<> {
+ public:
+  // TransportChannel object is always owned by the corresponding session.
+  explicit TransportChannelSocketAdapter(cricket::TransportChannel* channel);
+  virtual ~TransportChannelSocketAdapter();
+
+  // Closes the stream. |error_code| specifies error code that will
+  // be returned by Read() and Write() after the stream is closed.
+  // Must be called before the session and the channel are destroyed.
+  void Close(int error_code);
+
+  // Socket interface.
+  virtual int Read(net::IOBuffer* buf, int buf_len,
+                   net::CompletionCallback* callback);
+  virtual int Write(net::IOBuffer* buf, int buf_len,
+                    net::CompletionCallback* callback);
+
+  virtual bool SetReceiveBufferSize(int32 size);
+  virtual bool SetSendBufferSize(int32 size);
+
+ private:
+  void OnNewPacket(cricket::TransportChannel* channel,
+                   const char* data, size_t data_size);
+  void OnWritableState(cricket::TransportChannel* channel);
+  void OnChannelDestroyed(cricket::TransportChannel* channel);
+
+  cricket::TransportChannel* channel_;
+
+  bool read_pending_;
+  net::CompletionCallback* read_callback_;  // Not owned.
+  scoped_refptr<net::IOBuffer> read_buffer_;
+  int read_buffer_size_;
+
+  bool write_pending_;
+  net::CompletionCallback* write_callback_;  // Not owned.
+  scoped_refptr<net::IOBuffer> write_buffer_;
+  int write_buffer_size_;
+
+  int closed_error_code_;
+
+  DISALLOW_COPY_AND_ASSIGN(TransportChannelSocketAdapter);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_CHANNEL_SOCKET_ADAPTER_H_
diff --git a/remoting/jingle_glue/jingle_client.cc b/remoting/jingle_glue/jingle_client.cc
index d3ca67e..c6206f7 100644
--- a/remoting/jingle_glue/jingle_client.cc
+++ b/remoting/jingle_glue/jingle_client.cc
@@ -200,6 +200,11 @@ MessageLoop* JingleClient::message_loop() {
   return thread_->message_loop();
 }
 
+cricket::SessionManager* JingleClient::session_manager() {
+  DCHECK_EQ(message_loop(), MessageLoop::current());
+  return session_manager_.get();
+}
+
 void JingleClient::OnConnectionStateChanged(buzz::XmppEngine::State state) {
   switch (state) {
     case buzz::XmppEngine::STATE_START:
@@ -209,10 +214,7 @@ void JingleClient::OnConnectionStateChanged(buzz::XmppEngine::State state) {
       UpdateState(CONNECTING);
       break;
     case buzz::XmppEngine::STATE_OPEN:
-      {
-        AutoLock auto_lock(full_jid_lock_);
-        full_jid_ = client_->jid().Str();
-      }
+      SetFullJid(client_->jid().Str());
       UpdateState(CONNECTED);
       break;
     case buzz::XmppEngine::STATE_CLOSED:
@@ -248,10 +250,17 @@ void JingleClient::OnIncomingTunnel(
   }
 }
 
+void JingleClient::SetFullJid(const std::string& full_jid) {
+  AutoLock auto_lock(full_jid_lock_);
+  full_jid_ = full_jid;
+}
+
 void JingleClient::UpdateState(State new_state) {
   if (new_state != state_) {
     state_ = new_state;
     {
+      // We have to have the lock held, otherwise we cannot be sure that
+      // the client hasn't been closed when we call the callback.
       AutoLock auto_lock(state_lock_);
       if (!closed_)
         callback_->OnStateChange(this, new_state);
diff --git a/remoting/jingle_glue/jingle_client.h b/remoting/jingle_glue/jingle_client.h
index 9d7b844..e9bf84d 100644
--- a/remoting/jingle_glue/jingle_client.h
+++ b/remoting/jingle_glue/jingle_client.h
@@ -7,7 +7,6 @@
 
 #include <string>
 
-#include "base/waitable_event.h"
 #include "remoting/jingle_glue/jingle_channel.h"
 #include "third_party/libjingle/source/talk/xmpp/xmppclient.h"
 
@@ -106,6 +105,10 @@ class JingleClient : public base::RefCountedThreadSafe<JingleClient>,
   // Message loop used by this object to execute tasks.
   MessageLoop* message_loop();
 
+  // The session manager used by this client. Must be called from the
+  // jingle thread only. Returns NULL if the client is not active.
+  cricket::SessionManager* session_manager();
+
  private:
   friend class HeartbeatSenderTest;
   friend class JingleClientTest;
@@ -127,6 +130,8 @@ class JingleClient : public base::RefCountedThreadSafe<JingleClient>,
   // Used by Close().
   void DoClose();
 
+  void SetFullJid(const std::string& full_jid);
+
   // Updates current state of the connection. Must be called only in
   // the jingle thread.
   void UpdateState(State new_state);
diff --git a/remoting/jingle_glue/ssl_socket_adapter.cc b/remoting/jingle_glue/ssl_socket_adapter.cc
index 3ea404b..6eca04b 100644
--- a/remoting/jingle_glue/ssl_socket_adapter.cc
+++ b/remoting/jingle_glue/ssl_socket_adapter.cc
@@ -12,47 +12,10 @@
 #include "net/base/sys_addrinfo.h"
 #include "net/socket/client_socket_factory.h"
 #include "net/url_request/url_request_context.h"
+#include "remoting/jingle_glue/utils.h"
 
 namespace remoting {
 
-namespace {
-
-// Convert values from <errno.h> to values from "net/base/net_errors.h"
-int MapPosixError(int err) {
-  // There are numerous posix error codes, but these are the ones we thus far
-  // find interesting.
-  switch (err) {
-    case EAGAIN:
-#if EWOULDBLOCK != EAGAIN
-    case EWOULDBLOCK:
-#endif
-      return net::ERR_IO_PENDING;
-    case ENETDOWN:
-      return net::ERR_INTERNET_DISCONNECTED;
-    case ETIMEDOUT:
-      return net::ERR_TIMED_OUT;
-    case ECONNRESET:
-    case ENETRESET:  // Related to keep-alive
-      return net::ERR_CONNECTION_RESET;
-    case ECONNABORTED:
-      return net::ERR_CONNECTION_ABORTED;
-    case ECONNREFUSED:
-      return net::ERR_CONNECTION_REFUSED;
-    case EHOSTUNREACH:
-    case ENETUNREACH:
-      return net::ERR_ADDRESS_UNREACHABLE;
-    case EADDRNOTAVAIL:
-      return net::ERR_ADDRESS_INVALID;
-    case 0:
-      return net::OK;
-    default:
-      LOG(WARNING) << "Unknown error " << err << " mapped to net::ERR_FAILED";
-      return net::ERR_FAILED;
-  }
-}
-
-}  // namespace
-
 SSLSocketAdapter* SSLSocketAdapter::Create(AsyncSocket* socket) {
   return new SSLSocketAdapter(socket);
 }
@@ -290,7 +253,7 @@ int TransportSocket::Read(net::IOBuffer* buf, int buf_len,
   DCHECK(!read_buffer_.get());
   int result = socket_->Recv(buf->data(), buf_len);
   if (result < 0) {
-    result = MapPosixError(socket_->GetError());
+    result = MapPosixToChromeError(socket_->GetError());
     if (result == net::ERR_IO_PENDING) {
       read_callback_ = callback;
       read_buffer_ = buf;
@@ -309,7 +272,7 @@ int TransportSocket::Write(net::IOBuffer* buf, int buf_len,
   DCHECK(!write_buffer_.get());
   int result = socket_->Send(buf->data(), buf_len);
   if (result < 0) {
-    result = MapPosixError(socket_->GetError());
+    result = MapPosixToChromeError(socket_->GetError());
     if (result == net::ERR_IO_PENDING) {
       write_callback_ = callback;
       write_buffer_ = buf;
@@ -344,7 +307,7 @@ void TransportSocket::OnReadEvent(talk_base::AsyncSocket* socket) {
 
     int result = socket_->Recv(buffer->data(), buffer_len);
     if (result < 0) {
-      result = MapPosixError(socket_->GetError());
+      result = MapPosixToChromeError(socket_->GetError());
       if (result == net::ERR_IO_PENDING) {
         read_callback_ = callback;
         read_buffer_ = buffer;
@@ -370,7 +333,7 @@ void TransportSocket::OnWriteEvent(talk_base::AsyncSocket* socket) {
 
     int result = socket_->Send(buffer->data(), buffer_len);
     if (result < 0) {
-      result = MapPosixError(socket_->GetError());
+      result = MapPosixToChromeError(socket_->GetError());
       if (result == net::ERR_IO_PENDING) {
         write_callback_ = callback;
         write_buffer_ = buffer;
diff --git a/remoting/jingle_glue/stream_socket_adapter.cc b/remoting/jingle_glue/stream_socket_adapter.cc
new file mode 100644
index 0000000..cb84941
--- /dev/null
+++ b/remoting/jingle_glue/stream_socket_adapter.cc
@@ -0,0 +1,195 @@
+// Copyright (c) 2010 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/stream_socket_adapter.h"
+
+#include "base/logging.h"
+#include "base/message_loop.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+#include "remoting/jingle_glue/utils.h"
+#include "third_party/libjingle/source/talk/base/stream.h"
+
+namespace remoting {
+
+StreamSocketAdapter::StreamSocketAdapter(talk_base::StreamInterface* stream)
+    : stream_(stream),
+      read_pending_(false),
+      write_pending_(false),
+      closed_error_code_(net::OK) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(stream);
+  stream_->SignalEvent.connect(this, &StreamSocketAdapter::OnStreamEvent);
+}
+
+StreamSocketAdapter::~StreamSocketAdapter() {
+}
+
+int StreamSocketAdapter::Read(
+    net::IOBuffer* buffer, int buffer_size, net::CompletionCallback* callback) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(buffer);
+  CHECK(!read_pending_);
+
+  if (!stream_.get()) {
+    DCHECK(closed_error_code_ != net::OK);
+    return closed_error_code_;
+  }
+
+  int result = ReadStream(buffer, buffer_size);
+  if (result == net::ERR_CONNECTION_CLOSED &&
+      stream_->GetState() == talk_base::SS_OPENING)
+    result = net::ERR_IO_PENDING;
+  if (result == net::ERR_IO_PENDING) {
+    read_pending_ = true;
+    read_callback_ = callback;
+    read_buffer_ = buffer;
+    read_buffer_size_ = buffer_size;
+  }
+  return result;
+}
+
+int StreamSocketAdapter::Write(
+    net::IOBuffer* buffer, int buffer_size, net::CompletionCallback* callback) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+  DCHECK(buffer);
+  CHECK(!write_pending_);
+
+  if (!stream_.get()) {
+    DCHECK(closed_error_code_ != net::OK);
+    return closed_error_code_;
+  }
+
+  int result = WriteStream(buffer, buffer_size);
+  if (result == net::ERR_CONNECTION_CLOSED &&
+      stream_->GetState() == talk_base::SS_OPENING)
+    result = net::ERR_IO_PENDING;
+  if (result == net::ERR_IO_PENDING) {
+    write_pending_ = true;
+    write_callback_ = callback;
+    write_buffer_ = buffer;
+    write_buffer_size_ = buffer_size;
+  }
+  return result;
+}
+
+bool StreamSocketAdapter::SetReceiveBufferSize(int32 size) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+bool StreamSocketAdapter::SetSendBufferSize(int32 size) {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+void StreamSocketAdapter::Close(int error_code) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+
+  if (!stream_.get())  // Already closed.
+    return;
+
+  DCHECK(error_code != net::OK);
+  closed_error_code_ = error_code;
+  stream_->SignalEvent.disconnect(this);
+  stream_->Close();
+  stream_.reset(NULL);
+
+  if (read_pending_) {
+    net::CompletionCallback* callback = read_callback_;
+    read_pending_ = false;
+    read_buffer_ = NULL;
+    callback->Run(error_code);
+  }
+
+  if (write_pending_) {
+    net::CompletionCallback* callback = write_callback_;
+    write_pending_ = false;
+    write_buffer_ = NULL;
+    callback->Run(error_code);
+  }
+}
+
+void StreamSocketAdapter::OnStreamEvent(
+    talk_base::StreamInterface* stream, int events, int error) {
+  DCHECK_EQ(MessageLoop::TYPE_IO, MessageLoop::current()->type());
+
+  if (events & talk_base::SE_WRITE)
+    DoWrite();
+
+  if (events & talk_base::SE_READ)
+    DoRead();
+}
+
+void StreamSocketAdapter::DoWrite() {
+  // Write if there is a pending read.
+  if (write_buffer_) {
+    int result = WriteStream(write_buffer_, write_buffer_size_);
+    if (result != net::ERR_IO_PENDING) {
+      net::CompletionCallback* callback = write_callback_;
+      write_pending_ = false;
+      write_buffer_ = NULL;
+      callback->Run(result);
+    }
+  }
+}
+
+void StreamSocketAdapter::DoRead() {
+  // Read if there is a pending read.
+  if (read_pending_) {
+    int result = ReadStream(read_buffer_, read_buffer_size_);
+    if (result != net::ERR_IO_PENDING) {
+      net::CompletionCallback* callback = read_callback_;\
+      read_pending_ = false;
+      read_buffer_ = NULL;
+      callback->Run(result);
+    }
+  }
+}
+
+int StreamSocketAdapter::ReadStream(net::IOBuffer* buffer, int buffer_size) {
+  size_t bytes_read;
+  int error;
+  talk_base::StreamResult result = stream_->Read(
+      buffer->data(), buffer_size, &bytes_read, &error);
+  switch (result) {
+    case talk_base::SR_SUCCESS:
+      return bytes_read;
+
+    case talk_base::SR_BLOCK:
+      return net::ERR_IO_PENDING;
+
+    case talk_base::SR_EOS:
+      return net::ERR_CONNECTION_CLOSED;
+
+    case talk_base::SR_ERROR:
+      return MapPosixToChromeError(error);
+  }
+  NOTREACHED();
+  return net::ERR_FAILED;
+}
+
+int StreamSocketAdapter::WriteStream(net::IOBuffer* buffer, int buffer_size) {
+  size_t bytes_written;
+  int error;
+  talk_base::StreamResult result = stream_->Write(
+      buffer->data(), buffer_size, &bytes_written, &error);
+  switch (result) {
+    case talk_base::SR_SUCCESS:
+      return bytes_written;
+
+    case talk_base::SR_BLOCK:
+      return net::ERR_IO_PENDING;
+
+    case talk_base::SR_EOS:
+      return net::ERR_CONNECTION_CLOSED;
+
+    case talk_base::SR_ERROR:
+      return MapPosixToChromeError(error);
+  }
+  NOTREACHED();
+  return net::ERR_FAILED;
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/stream_socket_adapter.h b/remoting/jingle_glue/stream_socket_adapter.h
new file mode 100644
index 0000000..53ac76b
--- /dev/null
+++ b/remoting/jingle_glue/stream_socket_adapter.h
@@ -0,0 +1,70 @@
+// Copyright (c) 2010 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_STREAM_SOCKET_ADAPTER_H_
+#define REMOTING_JINGLE_GLUE_STREAM_SOCKET_ADAPTER_H_
+
+#include "base/scoped_ptr.h"
+#include "net/socket/socket.h"
+#include "third_party/libjingle/source/talk/base/sigslot.h"
+
+namespace talk_base {
+class StreamInterface;
+}  // namespace talk_base
+
+namespace remoting {
+
+// StreamSocketAdapter implements net::Socket interface on top of
+// libjingle's StreamInterface. It is used by JingleChromotingConnection
+// to provide net::Socket interface for channels.
+class StreamSocketAdapter : public net::Socket,
+                            public sigslot::has_slots<> {
+ public:
+  // Ownership of the stream is passed to the adapter.
+  explicit StreamSocketAdapter(talk_base::StreamInterface* stream);
+  virtual ~StreamSocketAdapter();
+
+  // Closes the stream. |error_code| specifies error code that will
+  // be returned by Read() and Write() after the stream is closed.
+  void Close(int error_code);
+
+  // Socket interface.
+  virtual int Read(net::IOBuffer* buffer, int buffer_size,
+                   net::CompletionCallback* callback);
+  virtual int Write(net::IOBuffer* buffer, int buffer_size,
+                    net::CompletionCallback* callback);
+
+  virtual bool SetReceiveBufferSize(int32 size);
+  virtual bool SetSendBufferSize(int32 size);
+
+ private:
+  void OnStreamEvent(talk_base::StreamInterface* stream,
+                     int events, int error);
+
+  void DoWrite();
+  void DoRead();
+
+  int ReadStream(net::IOBuffer* buffer, int buffer_size);
+  int WriteStream(net::IOBuffer* buffer, int buffer_size);
+
+  scoped_ptr<talk_base::StreamInterface> stream_;
+
+  bool read_pending_;
+  net::CompletionCallback* read_callback_;
+  scoped_refptr<net::IOBuffer> read_buffer_;
+  int read_buffer_size_;
+
+  bool write_pending_;
+  net::CompletionCallback* write_callback_;
+  scoped_refptr<net::IOBuffer> write_buffer_;
+  int write_buffer_size_;
+
+  int closed_error_code_;
+
+  DISALLOW_COPY_AND_ASSIGN(StreamSocketAdapter);
+};
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_STREAM_SOCKET_ADAPTER_H_
diff --git a/remoting/jingle_glue/utils.cc b/remoting/jingle_glue/utils.cc
new file mode 100644
index 0000000..b37d734
--- /dev/null
+++ b/remoting/jingle_glue/utils.cc
@@ -0,0 +1,53 @@
+// Copyright (c) 2010 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/utils.h"
+
+#include "base/logging.h"
+#include "net/base/net_errors.h"
+#include "third_party/libjingle/source/talk/base/socket.h"
+
+namespace remoting {
+
+// TODO(sergeyu): This is a clone of MapPosixError() from
+// net/socket/tcp_client_socket_libevent.cc. Move MapPosixError() to
+// net/base/net_errors.cc and use it here.
+
+// Convert values from <errno.h> to values from "net/base/net_errors.h"
+int MapPosixToChromeError(int err) {
+  // There are numerous posix error codes, but these are the ones we thus far
+  // find interesting.
+  switch (err) {
+    case EAGAIN:
+#if EWOULDBLOCK != EAGAIN
+    case EWOULDBLOCK:
+#endif
+      return net::ERR_IO_PENDING;
+    case ENETDOWN:
+      return net::ERR_INTERNET_DISCONNECTED;
+    case ETIMEDOUT:
+      return net::ERR_TIMED_OUT;
+    case ENOTCONN:
+      return net::ERR_CONNECTION_CLOSED;
+    case ECONNRESET:
+    case ENETRESET:  // Related to keep-alive
+      return net::ERR_CONNECTION_RESET;
+    case ECONNABORTED:
+      return net::ERR_CONNECTION_ABORTED;
+    case ECONNREFUSED:
+      return net::ERR_CONNECTION_REFUSED;
+    case EHOSTUNREACH:
+    case ENETUNREACH:
+      return net::ERR_ADDRESS_UNREACHABLE;
+    case EADDRNOTAVAIL:
+      return net::ERR_ADDRESS_INVALID;
+    case 0:
+      return net::OK;
+    default:
+      LOG(WARNING) << "Unknown error " << err << " mapped to net::ERR_FAILED";
+      return net::ERR_FAILED;
+  }
+}
+
+}  // namespace remoting
diff --git a/remoting/jingle_glue/utils.h b/remoting/jingle_glue/utils.h
new file mode 100644
index 0000000..044a9ed
--- /dev/null
+++ b/remoting/jingle_glue/utils.h
@@ -0,0 +1,15 @@
+// Copyright (c) 2010 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_UTILS_H_
+#define REMOTING_JINGLE_GLUE_UTILS_H_
+
+namespace remoting {
+
+// Convert values from <errno.h> to values from "net/base/net_errors.h"
+int MapPosixToChromeError(int err);
+
+}  // namespace remoting
+
+#endif  // REMOTING_JINGLE_GLUE_UTILS_H_