Remove video_channel() from Session interface

BUG=None
TEST=Unittests.

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

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

26 files changed, 490 insertions, 177 deletions

diff --git a/remoting/base/constants.cc b/remoting/base/constants.cc
index 219fe14..90ea755 100644
--- a/remoting/base/constants.cc
+++ b/remoting/base/constants.cc
@@ -12,4 +12,8 @@ const char kChromotingTokenDefaultServiceName[] = "chromiumsync";
 
 const char kChromotingXmlNamespace[] = "google:remoting";
 
+const char kVideoChannelName[] = "video";
+const char kVideoRtpChannelName[] = "videortp";
+const char kVideoRtcpChannelName[] = "videortpc";
+
 }  // namespace remoting
diff --git a/remoting/base/constants.h b/remoting/base/constants.h
index d464fa4..6efa408 100644
--- a/remoting/base/constants.h
+++ b/remoting/base/constants.h
@@ -17,6 +17,11 @@ extern const char kChromotingTokenDefaultServiceName[];
 // Namespace used for chromoting XMPP stanzas.
 extern const char kChromotingXmlNamespace[];
 
+// Channel names.
+extern const char kVideoChannelName[];
+extern const char kVideoRtpChannelName[];
+extern const char kVideoRtcpChannelName[];
+
 }  // namespace remoting
 
 #endif  // REMOTING_BASE_CONSTANTS_H_
diff --git a/remoting/protocol/buffered_socket_writer.h b/remoting/protocol/buffered_socket_writer.h
index 46a3a7e..3bb03d3 100644
--- a/remoting/protocol/buffered_socket_writer.h
+++ b/remoting/protocol/buffered_socket_writer.h
@@ -41,7 +41,8 @@ class BufferedSocketWriterBase
 
   // Initializes the writer. Must be called on the thread that will be used
   // to access the socket in the future. |callback| will be called after each
-  // failed write.
+  // failed write. Caller retains ownership of |socket|.
+  // TODO(sergeyu): Change it so that it take ownership of |socket|.
   void Init(net::Socket* socket, WriteFailedCallback* callback);
 
   // Puts a new data chunk in the buffer. Returns false and doesn't enqueue
diff --git a/remoting/protocol/connection_to_client.cc b/remoting/protocol/connection_to_client.cc
index f5cf592e..f61aded 100644
--- a/remoting/protocol/connection_to_client.cc
+++ b/remoting/protocol/connection_to_client.cc
@@ -4,6 +4,7 @@
 
 #include "remoting/protocol/connection_to_client.h"
 
+#include "base/bind.h"
 #include "google/protobuf/message.h"
 #include "net/base/io_buffer.h"
 #include "remoting/protocol/client_control_sender.h"
@@ -96,20 +97,18 @@ void ConnectionToClient::OnSessionStateChange(protocol::Session::State state) {
       client_control_sender_.reset(
           new ClientControlSender(session_->control_channel()));
       video_writer_.reset(VideoWriter::Create(session_->config()));
-      video_writer_->Init(session_.get());
-
+      video_writer_->Init(
+          session_.get(), base::Bind(&ConnectionToClient::OnVideoInitialized,
+                                     base::Unretained(this)));
       dispatcher_.reset(new HostMessageDispatcher());
       dispatcher_->Initialize(this, host_stub_, input_stub_);
-
-      handler_->OnConnectionOpened(this);
       break;
     case protocol::Session::CLOSED:
       CloseChannels();
       handler_->OnConnectionClosed(this);
       break;
     case protocol::Session::FAILED:
-      CloseChannels();
-      handler_->OnConnectionFailed(this);
+      CloseOnError();
       break;
     default:
       // We shouldn't receive other states.
@@ -117,6 +116,20 @@ void ConnectionToClient::OnSessionStateChange(protocol::Session::State state) {
   }
 }
 
+void ConnectionToClient::OnVideoInitialized(bool successful) {
+  if (!successful) {
+    CloseOnError();
+    return;
+  }
+
+  handler_->OnConnectionOpened(this);
+}
+
+void ConnectionToClient::CloseOnError() {
+  CloseChannels();
+  handler_->OnConnectionFailed(this);
+}
+
 void ConnectionToClient::CloseChannels() {
   if (video_writer_.get())
     video_writer_->Close();
diff --git a/remoting/protocol/connection_to_client.h b/remoting/protocol/connection_to_client.h
index 296ce63..9d7e25f 100644
--- a/remoting/protocol/connection_to_client.h
+++ b/remoting/protocol/connection_to_client.h
@@ -90,6 +90,11 @@ class ConnectionToClient :
   // Callback for protocol Session.
   void OnSessionStateChange(Session::State state);
 
+  // Callback for VideoReader::Init().
+  void OnVideoInitialized(bool successful);
+
+  void CloseOnError();
+
   // Stops writing in the channels.
   void CloseChannels();
 
diff --git a/remoting/protocol/connection_to_client_unittest.cc b/remoting/protocol/connection_to_client_unittest.cc
index 3609511..8a8e1e8 100644
--- a/remoting/protocol/connection_to_client_unittest.cc
+++ b/remoting/protocol/connection_to_client_unittest.cc
@@ -4,6 +4,7 @@
 
 #include "base/message_loop.h"
 #include "remoting/base/base_mock_objects.h"
+#include "remoting/base/constants.h"
 #include "remoting/protocol/fake_session.h"
 #include "remoting/protocol/connection_to_client.h"
 #include "remoting/protocol/protocol_mock_objects.h"
@@ -60,7 +61,9 @@ TEST_F(ConnectionToClientTest, SendUpdateStream) {
 
   // Verify that something has been written.
   // TODO(sergeyu): Verify that the correct data has been written.
-  EXPECT_GT(session_->video_channel()->written_data().size(), 0u);
+  ASSERT_TRUE(session_->GetStreamChannel(kVideoChannelName));
+  EXPECT_GT(session_->GetStreamChannel(kVideoChannelName)->
+            written_data().size(), 0u);
 
   // And then close the connection to ConnectionToClient.
   viewer_->Disconnect();
diff --git a/remoting/protocol/connection_to_host.cc b/remoting/protocol/connection_to_host.cc
index 26dca8b..f00deed 100644
--- a/remoting/protocol/connection_to_host.cc
+++ b/remoting/protocol/connection_to_host.cc
@@ -176,8 +176,7 @@ void ConnectionToHost::OnSessionStateChange(
   switch (state) {
     case Session::FAILED:
       state_ = STATE_FAILED;
-      CloseChannels();
-      event_callback_->OnConnectionFailed(this);
+      CloseOnError();
       break;
 
     case Session::CLOSED:
@@ -190,7 +189,10 @@ void ConnectionToHost::OnSessionStateChange(
       state_ = STATE_CONNECTED;
       // Initialize reader and writer.
       video_reader_.reset(VideoReader::Create(session_->config()));
-      video_reader_->Init(session_.get(), video_stub_);
+      video_reader_->Init(
+          session_.get(), video_stub_,
+          base::Bind(&ConnectionToHost::OnVideoChannelInitialized,
+                     base::Unretained(this)));
       host_control_sender_.reset(
           new HostControlSender(session_->control_channel()));
       dispatcher_->Initialize(session_.get(), client_stub_);
@@ -203,12 +205,27 @@ void ConnectionToHost::OnSessionStateChange(
   }
 }
 
+void ConnectionToHost::OnVideoChannelInitialized(bool successful) {
+  if (!successful) {
+    CloseOnError();
+    return;
+  }
+}
+
+void ConnectionToHost::CloseOnError() {
+  state_ = STATE_FAILED;
+  CloseChannels();
+  event_callback_->OnConnectionFailed(this);
+}
+
 void ConnectionToHost::CloseChannels() {
   if (input_sender_.get())
     input_sender_->Close();
 
   if (host_control_sender_.get())
     host_control_sender_->Close();
+
+  video_reader_.reset();
 }
 
 void ConnectionToHost::OnClientAuthenticated() {
diff --git a/remoting/protocol/connection_to_host.h b/remoting/protocol/connection_to_host.h
index 29bbb54..65a4738 100644
--- a/remoting/protocol/connection_to_host.h
+++ b/remoting/protocol/connection_to_host.h
@@ -124,9 +124,14 @@ class ConnectionToHost : public SignalStrategy::StatusObserver,
   // Callback for |session_|.
   void OnSessionStateChange(Session::State state);
 
+  // Callback for VideoReader::Init().
+  void OnVideoChannelInitialized(bool successful);
+
   // Callback for |video_reader_|.
   void OnVideoPacket(VideoPacket* packet);
 
+  void CloseOnError();
+
   // Stops writing in the channels.
   void CloseChannels();
 
diff --git a/remoting/protocol/fake_session.cc b/remoting/protocol/fake_session.cc
index b23dbb1..a2496a7 100644
--- a/remoting/protocol/fake_session.cc
+++ b/remoting/protocol/fake_session.cc
@@ -70,6 +70,67 @@ bool FakeSocket::SetSendBufferSize(int32 size) {
   return false;
 }
 
+int FakeSocket::Connect(net::CompletionCallback* callback) {
+  return net::OK;
+}
+
+void FakeSocket::Disconnect() {
+  NOTIMPLEMENTED();
+}
+
+bool FakeSocket::IsConnected() const {
+  return true;
+}
+
+bool FakeSocket::IsConnectedAndIdle() const {
+  NOTIMPLEMENTED();
+  return false;
+}
+
+int FakeSocket::GetPeerAddress(
+    net::AddressList* address) const {
+  NOTIMPLEMENTED();
+  return net::ERR_FAILED;
+}
+
+int FakeSocket::GetLocalAddress(
+    net::IPEndPoint* address) const {
+  NOTIMPLEMENTED();
+  return net::ERR_FAILED;
+}
+
+const net::BoundNetLog& FakeSocket::NetLog() const {
+  return net_log_;
+}
+
+void FakeSocket::SetSubresourceSpeculation() {
+  NOTIMPLEMENTED();
+}
+
+void FakeSocket::SetOmniboxSpeculation() {
+  NOTIMPLEMENTED();
+}
+
+bool FakeSocket::WasEverUsed() const {
+  NOTIMPLEMENTED();
+  return true;
+}
+
+bool FakeSocket::UsingTCPFastOpen() const {
+  NOTIMPLEMENTED();
+  return true;
+}
+
+int64 FakeSocket::NumBytesRead() const {
+  NOTIMPLEMENTED();
+  return 0;
+}
+
+base::TimeDelta FakeSocket::GetConnectTimeMicros() const {
+  NOTIMPLEMENTED();
+  return base::TimeDelta();
+}
+
 FakeUdpSocket::FakeUdpSocket()
     : read_pending_(false),
       input_pos_(0) {
@@ -135,21 +196,31 @@ FakeSession::FakeSession()
 
 FakeSession::~FakeSession() { }
 
-void FakeSession::SetStateChangeCallback(
-    StateChangeCallback* callback) {
+FakeSocket* FakeSession::GetStreamChannel(const std::string& name) {
+  return stream_channels_[name];
+}
+
+FakeUdpSocket* FakeSession::GetDatagramChannel(const std::string& name) {
+  return datagram_channels_[name];
+}
+
+void FakeSession::SetStateChangeCallback(StateChangeCallback* callback) {
   callback_.reset(callback);
 }
 
 void FakeSession::CreateStreamChannel(
     const std::string& name, const StreamChannelCallback& callback) {
-  NOTIMPLEMENTED();
-  callback.Run(name, NULL);
+  LOG(ERROR) << " creating channel " << name;
+  FakeSocket* channel = new FakeSocket();
+  stream_channels_[name] = channel;
+  callback.Run(name, channel);
 }
 
 void FakeSession::CreateDatagramChannel(
     const std::string& name, const DatagramChannelCallback& callback) {
-  NOTIMPLEMENTED();
-  callback.Run(name, NULL);
+  FakeUdpSocket* channel = new FakeUdpSocket();
+  datagram_channels_[name] = channel;
+  callback.Run(name, channel);
 }
 
 FakeSocket* FakeSession::control_channel() {
@@ -160,18 +231,6 @@ FakeSocket* FakeSession::event_channel() {
   return &event_channel_;
 }
 
-FakeSocket* FakeSession::video_channel() {
-  return &video_channel_;
-}
-
-FakeUdpSocket* FakeSession::video_rtp_channel() {
-  return &video_rtp_channel_;
-}
-
-FakeUdpSocket* FakeSession::video_rtcp_channel() {
-  return &video_rtcp_channel_;
-}
-
 const std::string& FakeSession::jid() {
   return jid_;
 }
diff --git a/remoting/protocol/fake_session.h b/remoting/protocol/fake_session.h
index 912cfb0..f0d5828 100644
--- a/remoting/protocol/fake_session.h
+++ b/remoting/protocol/fake_session.h
@@ -5,11 +5,13 @@
 #ifndef REMOTING_PROTOCOL_FAKE_SESSION_H_
 #define REMOTING_PROTOCOL_FAKE_SESSION_H_
 
+#include <map>
 #include <string>
 #include <vector>
 
 #include "base/memory/scoped_ptr.h"
 #include "net/socket/socket.h"
+#include "net/socket/stream_socket.h"
 #include "remoting/protocol/session.h"
 
 namespace remoting {
@@ -22,7 +24,7 @@ extern const char kTestJid[];
 // Read() reads data from another buffer that can be set with AppendInputData().
 // Pending reads are supported, so if there is a pending read AppendInputData()
 // calls the read callback.
-class FakeSocket : public net::Socket {
+class FakeSocket : public net::StreamSocket {
  public:
   FakeSocket();
   virtual ~FakeSocket();
@@ -42,6 +44,21 @@ class FakeSocket : public net::Socket {
   virtual bool SetReceiveBufferSize(int32 size);
   virtual bool SetSendBufferSize(int32 size);
 
+  // net::StreamSocket interface.
+  virtual int Connect(net::CompletionCallback* callback) OVERRIDE;
+  virtual void Disconnect() OVERRIDE;
+  virtual bool IsConnected() const OVERRIDE;
+  virtual bool IsConnectedAndIdle() const OVERRIDE;
+  virtual int GetPeerAddress(net::AddressList* 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;
+
  private:
   bool read_pending_;
   scoped_refptr<net::IOBuffer> read_buffer_;
@@ -51,6 +68,10 @@ class FakeSocket : public net::Socket {
   std::string written_data_;
   std::string input_data_;
   int input_pos_;
+
+  net::BoundNetLog net_log_;
+
+  DISALLOW_COPY_AND_ASSIGN(FakeSocket);
 };
 
 // FakeUdpSocket is similar to FakeSocket but behaves as UDP socket. All written
@@ -86,6 +107,8 @@ class FakeUdpSocket : public net::Socket {
   std::vector<std::string> written_packets_;
   std::vector<std::string> input_packets_;
   int input_pos_;
+
+  DISALLOW_COPY_AND_ASSIGN(FakeUdpSocket);
 };
 
 // FakeSession is a dummy protocol::Session that uses FakeSocket for all
@@ -103,6 +126,10 @@ class FakeSession : public Session {
 
   bool is_closed() const { return closed_; }
 
+  FakeSocket* GetStreamChannel(const std::string& name);
+  FakeUdpSocket* GetDatagramChannel(const std::string& name);
+
+  // Session interface.
   virtual void SetStateChangeCallback(StateChangeCallback* callback);
 
   virtual void CreateStreamChannel(
@@ -112,10 +139,6 @@ class FakeSession : public Session {
 
   virtual FakeSocket* control_channel();
   virtual FakeSocket* event_channel();
-  virtual FakeSocket* video_channel();
-
-  virtual FakeUdpSocket* video_rtp_channel();
-  virtual FakeUdpSocket* video_rtcp_channel();
 
   virtual const std::string& jid();
 
@@ -140,9 +163,9 @@ class FakeSession : public Session {
   MessageLoop* message_loop_;
   FakeSocket control_channel_;
   FakeSocket event_channel_;
-  FakeSocket video_channel_;
-  FakeUdpSocket video_rtp_channel_;
-  FakeUdpSocket video_rtcp_channel_;
+
+  std::map<std::string, FakeSocket*> stream_channels_;
+  std::map<std::string, FakeUdpSocket*> datagram_channels_;
 
   std::string initiator_token_;
   std::string receiver_token_;
@@ -151,6 +174,8 @@ class FakeSession : public Session {
 
   std::string jid_;
   bool closed_;
+
+  DISALLOW_COPY_AND_ASSIGN(FakeSession);
 };
 
 }  // namespace protocol
diff --git a/remoting/protocol/jingle_session.cc b/remoting/protocol/jingle_session.cc
index 0fa7ccb..7980de4 100644
--- a/remoting/protocol/jingle_session.cc
+++ b/remoting/protocol/jingle_session.cc
@@ -33,7 +33,6 @@ namespace {
 
 const char kControlChannelName[] = "control";
 const char kEventChannelName[] = "event";
-const char kVideoChannelName[] = "video";
 
 const int kMasterKeyLength = 16;
 const int kChannelKeyLength = 16;
@@ -160,7 +159,6 @@ void JingleSession::CloseInternal(int result, bool failed) {
 
     control_channel_socket_.reset();
     event_channel_socket_.reset();
-    video_channel_socket_.reset();
     STLDeleteContainerPairSecondPointers(channel_connectors_.begin(),
                                          channel_connectors_.end());
 
@@ -224,23 +222,6 @@ net::Socket* JingleSession::event_channel() {
   return event_channel_socket_.get();
 }
 
-net::Socket* JingleSession::video_channel() {
-  DCHECK(CalledOnValidThread());
-  return video_channel_socket_.get();
-}
-
-net::Socket* JingleSession::video_rtp_channel() {
-  DCHECK(CalledOnValidThread());
-  NOTREACHED();
-  return NULL;
-}
-
-net::Socket* JingleSession::video_rtcp_channel() {
-  DCHECK(CalledOnValidThread());
-  NOTREACHED();
-  return NULL;
-}
-
 const std::string& JingleSession::jid() {
   // TODO(sergeyu): Fix ChromotingHost so that it doesn't call this
   // method on invalid thread and uncomment this DCHECK.
@@ -512,7 +493,6 @@ void JingleSession::CreateChannels() {
                  base::Unretained(this)));
   CreateStreamChannel(kControlChannelName, stream_callback);
   CreateStreamChannel(kEventChannelName, stream_callback);
-  CreateStreamChannel(kVideoChannelName, stream_callback);
 }
 
 void JingleSession::OnStreamChannelConnected(const std::string& name,
@@ -533,14 +513,11 @@ void JingleSession::OnChannelConnected(const std::string& name,
     control_channel_socket_.reset(socket);
   } else if (name == kEventChannelName) {
     event_channel_socket_.reset(socket);
-  } else if (name == kVideoChannelName) {
-    video_channel_socket_.reset(socket);
   } else {
     NOTREACHED();
   }
 
-  if (control_channel_socket_.get() && event_channel_socket_.get() &&
-      video_channel_socket_.get()) {
+  if (control_channel_socket_.get() && event_channel_socket_.get()) {
     // TODO(sergeyu): State should be set to CONNECTED in OnAccept
     // independent of the channels state.
     SetState(CONNECTED);
diff --git a/remoting/protocol/jingle_session.h b/remoting/protocol/jingle_session.h
index 6cb15f2..7ac6fac 100644
--- a/remoting/protocol/jingle_session.h
+++ b/remoting/protocol/jingle_session.h
@@ -38,9 +38,6 @@ class JingleSession : public protocol::Session,
       const DatagramChannelCallback& callback) OVERRIDE;
   virtual net::Socket* control_channel() OVERRIDE;
   virtual net::Socket* event_channel() OVERRIDE;
-  virtual net::Socket* video_channel() OVERRIDE;
-  virtual net::Socket* video_rtp_channel() OVERRIDE;
-  virtual net::Socket* video_rtcp_channel() OVERRIDE;
   virtual const std::string& jid() OVERRIDE;
   virtual const CandidateSessionConfig* candidate_config() OVERRIDE;
   virtual const SessionConfig* config() OVERRIDE;
@@ -185,7 +182,6 @@ class JingleSession : public protocol::Session,
 
   scoped_ptr<net::Socket> control_channel_socket_;
   scoped_ptr<net::Socket> event_channel_socket_;
-  scoped_ptr<net::Socket> video_channel_socket_;
 
   ScopedRunnableMethodFactory<JingleSession> task_factory_;
 
diff --git a/remoting/protocol/jingle_session_unittest.cc b/remoting/protocol/jingle_session_unittest.cc
index 1d09521..49f686d 100644
--- a/remoting/protocol/jingle_session_unittest.cc
+++ b/remoting/protocol/jingle_session_unittest.cc
@@ -13,6 +13,7 @@
 #include "net/base/io_buffer.h"
 #include "net/base/net_errors.h"
 #include "net/socket/socket.h"
+#include "net/socket/stream_socket.h"
 #include "remoting/protocol/jingle_session.h"
 #include "remoting/protocol/jingle_session_manager.h"
 #include "remoting/jingle_glue/jingle_thread.h"
@@ -53,6 +54,7 @@ const int kMessages = 100;
 const int kTestDataSize = kMessages * kMessageSize;
 const int kUdpWriteDelayMs = 10;
 const char kTestToken[] = "a_dummy_token";
+const char kChannelName[] = "test_channel";
 
 const char kHostJid[] = "host1@gmail.com/123";
 const char kClientJid[] = "host2@gmail.com/321";
@@ -292,14 +294,6 @@ class JingleSessionTest : public testing::Test {
 
 class ChannelTesterBase : public base::RefCountedThreadSafe<ChannelTesterBase> {
  public:
-  enum ChannelType {
-    CONTROL,
-    EVENT,
-    VIDEO,
-    VIDEO_RTP,
-    VIDEO_RTCP,
-  };
-
   ChannelTesterBase(Session* host_session,
                     Session* client_session)
       : host_session_(host_session),
@@ -309,10 +303,9 @@ class ChannelTesterBase : public base::RefCountedThreadSafe<ChannelTesterBase> {
 
   virtual ~ChannelTesterBase() { }
 
-  void Start(ChannelType channel) {
+  void Start() {
     MessageLoop::current()->PostTask(
-        FROM_HERE, NewRunnableMethod(this, &ChannelTesterBase::DoStart,
-                                     channel));
+        FROM_HERE, NewRunnableMethod(this, &ChannelTesterBase::DoStart));
   }
 
   bool WaitFinished() {
@@ -322,15 +315,12 @@ class ChannelTesterBase : public base::RefCountedThreadSafe<ChannelTesterBase> {
   virtual void CheckResults() = 0;
 
  protected:
-  void DoStart(ChannelType channel) {
-    socket_1_ = SelectChannel(host_session_, channel);
-    socket_2_ = SelectChannel(client_session_, channel);
-
-    InitBuffers();
-    DoRead();
-    DoWrite();
+  void DoStart() {
+    InitChannels();
   }
 
+  virtual void InitChannels() = 0;
+
   void Done() {
     done_ = true;
     MessageLoop::current()->PostTask(FROM_HERE, base::Bind(&QuitCurrentThread));
@@ -340,29 +330,9 @@ class ChannelTesterBase : public base::RefCountedThreadSafe<ChannelTesterBase> {
   virtual void DoWrite() = 0;
   virtual void DoRead() = 0;
 
-  net::Socket* SelectChannel(Session* session,
-                             ChannelType channel) {
-    switch (channel) {
-      case CONTROL:
-        return session->control_channel();
-      case EVENT:
-        return session->event_channel();
-      case VIDEO:
-        return session->video_channel();
-      case VIDEO_RTP:
-        return session->video_rtp_channel();
-      case VIDEO_RTCP:
-        return session->video_rtcp_channel();
-      default:
-        NOTREACHED();
-        return NULL;
-    }
-  }
-
   Session* host_session_;
   Session* client_session_;
-  net::Socket* socket_1_;
-  net::Socket* socket_2_;
+  scoped_ptr<net::Socket> sockets_[2];
   bool done_;
 };
 
@@ -400,6 +370,36 @@ class TCPChannelTester : public ChannelTesterBase {
   }
 
  protected:
+  virtual void InitChannels() OVERRIDE {
+    host_session_->CreateStreamChannel(
+        kChannelName,
+        base::Bind(&TCPChannelTester::OnChannelReady,
+                   base::Unretained(this), 0));
+    client_session_->CreateStreamChannel(
+        kChannelName,
+        base::Bind(&TCPChannelTester::OnChannelReady,
+                   base::Unretained(this), 1));
+  }
+
+  void OnChannelReady(int id, const std::string name,
+                      net::StreamSocket* socket) {
+    ASSERT_TRUE(socket);
+    ASSERT_EQ(name, kChannelName);
+    if (!socket) {
+      Done();
+      return;
+    }
+
+    DCHECK(id >= 0 && id < 2);
+    sockets_[id].reset(socket);
+
+    if (sockets_[0].get() && sockets_[1].get()) {
+      InitBuffers();
+      DoRead();
+      DoWrite();
+    }
+  }
+
   virtual void InitBuffers() {
     output_buffer_ = new net::DrainableIOBuffer(
         new net::IOBuffer(test_data_size_), test_data_size_);
@@ -415,7 +415,7 @@ class TCPChannelTester : public ChannelTesterBase {
         break;
       int bytes_to_write = std::min(output_buffer_->BytesRemaining(),
                                     message_size_);
-      result = socket_1_->Write(output_buffer_, bytes_to_write, &write_cb_);
+      result = sockets_[0]->Write(output_buffer_, bytes_to_write, &write_cb_);
       HandleWriteResult(result);
     };
   }
@@ -439,7 +439,7 @@ class TCPChannelTester : public ChannelTesterBase {
     int result = 1;
     while (result > 0) {
       input_buffer_->SetCapacity(input_buffer_->offset() + message_size_);
-      result = socket_2_->Read(input_buffer_, message_size_, &read_cb_);
+      result = sockets_[1]->Read(input_buffer_, message_size_, &read_cb_);
       HandleReadResult(result);
     };
   }
@@ -536,6 +536,36 @@ class UDPChannelTester : public ChannelTesterBase {
   }
 
  protected:
+  virtual void InitChannels() OVERRIDE {
+    host_session_->CreateDatagramChannel(
+        kChannelName,
+        base::Bind(&UDPChannelTester::OnChannelReady,
+                   base::Unretained(this), 0));
+    client_session_->CreateDatagramChannel(
+        kChannelName,
+        base::Bind(&UDPChannelTester::OnChannelReady,
+                   base::Unretained(this), 1));
+  }
+
+  void OnChannelReady(int id, const std::string name, net::Socket* socket) {
+    ASSERT_TRUE(socket);
+    ASSERT_EQ(name, kChannelName);
+    if (!socket) {
+      Done();
+      return;
+    }
+
+    DCHECK(id >= 0 && id < 2);
+    sockets_[id].reset(socket);
+
+    if (sockets_[0].get() && sockets_[1].get()) {
+      InitBuffers();
+      DoRead();
+      DoWrite();
+    }
+  }
+
+
   virtual void InitBuffers() {
   }
 
@@ -551,7 +581,7 @@ class UDPChannelTester : public ChannelTesterBase {
     // Put index of this packet in the beginning of the packet body.
     memcpy(packet->data(), &packets_sent_, sizeof(packets_sent_));
 
-    int result = socket_1_->Write(packet, kMessageSize, &write_cb_);
+    int result = sockets_[0]->Write(packet, kMessageSize, &write_cb_);
     HandleWriteResult(result);
   }
 
@@ -579,7 +609,7 @@ class UDPChannelTester : public ChannelTesterBase {
       int kReadSize = kMessageSize * 2;
       read_buffer_ = new net::IOBuffer(kReadSize);
 
-      result = socket_2_->Read(read_buffer_, kReadSize, &read_cb_);
+      result = sockets_[1]->Read(read_buffer_, kReadSize, &read_cb_);
       HandleReadResult(result);
     };
   }
@@ -678,43 +708,13 @@ TEST_F(JingleSessionTest, ConnectBadChannelAuth) {
 }
 
 // Verify that data can be transmitted over the event channel.
-TEST_F(JingleSessionTest, TestControlChannel) {
+TEST_F(JingleSessionTest, TestTcpChannel) {
   CreateServerPair();
   ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
   scoped_refptr<TCPChannelTester> tester(
       new TCPChannelTester(host_session_.get(), client_session_.get(),
                            kMessageSize, kMessages));
-  tester->Start(ChannelTesterBase::CONTROL);
-  ASSERT_TRUE(tester->WaitFinished());
-  tester->CheckResults();
-
-  // Connections must be closed while |tester| still exists.
-  CloseSessions();
-}
-
-// Verify that data can be transmitted over the video channel.
-TEST_F(JingleSessionTest, TestVideoChannel) {
-  CreateServerPair();
-  ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
-  scoped_refptr<TCPChannelTester> tester(
-      new TCPChannelTester(host_session_.get(), client_session_.get(),
-                           kMessageSize, kMessageSize));
-  tester->Start(ChannelTesterBase::VIDEO);
-  ASSERT_TRUE(tester->WaitFinished());
-  tester->CheckResults();
-
-  // Connections must be closed while |tester| still exists.
-  CloseSessions();
-}
-
-// Verify that data can be transmitted over the event channel.
-TEST_F(JingleSessionTest, TestEventChannel) {
-  CreateServerPair();
-  ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
-  scoped_refptr<TCPChannelTester> tester(
-      new TCPChannelTester(host_session_.get(), client_session_.get(),
-                           kMessageSize, kMessageSize));
-  tester->Start(ChannelTesterBase::EVENT);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   tester->CheckResults();
 
@@ -723,13 +723,12 @@ TEST_F(JingleSessionTest, TestEventChannel) {
 }
 
 // Verify that data can be transmitted over the video RTP channel.
-// Disabled because RTP support is disabled, see crbug.com/91538 .
-TEST_F(JingleSessionTest, DISABLED_TestVideoRtpChannel) {
+TEST_F(JingleSessionTest, TestUdpChannel) {
   CreateServerPair();
   ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
   scoped_refptr<UDPChannelTester> tester(
       new UDPChannelTester(host_session_.get(), client_session_.get()));
-  tester->Start(ChannelTesterBase::VIDEO_RTP);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   tester->CheckResults();
 
@@ -746,28 +745,37 @@ TEST_F(JingleSessionTest, TestSpeed) {
 
   tester = new ChannelSpeedTester(host_session_.get(),
                                   client_session_.get(), 512);
-  tester->Start(ChannelTesterBase::VIDEO);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   LOG(INFO) << "Time for 512 bytes "
             << tester->GetElapsedTime().InMilliseconds() << " ms.";
 
+  CloseSessions();
+  ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
+
   tester = new ChannelSpeedTester(host_session_.get(),
                                   client_session_.get(), 1024);
-  tester->Start(ChannelTesterBase::VIDEO);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   LOG(INFO) << "Time for 1024 bytes "
             << tester->GetElapsedTime().InMilliseconds() << " ms.";
 
+  CloseSessions();
+  ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
+
   tester = new ChannelSpeedTester(host_session_.get(),
                                   client_session_.get(), 51200);
-  tester->Start(ChannelTesterBase::VIDEO);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   LOG(INFO) << "Time for 50k bytes "
             << tester->GetElapsedTime().InMilliseconds() << " ms.";
 
+  CloseSessions();
+  ASSERT_TRUE(InitiateConnection(kTestSharedSecret));
+
   tester = new ChannelSpeedTester(host_session_.get(),
                                   client_session_.get(), 512000);
-  tester->Start(ChannelTesterBase::VIDEO);
+  tester->Start();
   ASSERT_TRUE(tester->WaitFinished());
   LOG(INFO) << "Time for 500k bytes "
             << tester->GetElapsedTime().InMilliseconds() << " ms.";
diff --git a/remoting/protocol/protobuf_video_reader.cc b/remoting/protocol/protobuf_video_reader.cc
index a2b62388..f93f2fb 100644
--- a/remoting/protocol/protobuf_video_reader.cc
+++ b/remoting/protocol/protobuf_video_reader.cc
@@ -4,7 +4,10 @@
 
 #include "remoting/protocol/protobuf_video_reader.h"
 
+#include "base/bind.h"
 #include "base/task.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/session.h"
 
@@ -19,11 +22,28 @@ ProtobufVideoReader::ProtobufVideoReader(VideoPacketFormat::Encoding encoding)
 ProtobufVideoReader::~ProtobufVideoReader() { }
 
 void ProtobufVideoReader::Init(protocol::Session* session,
-                               VideoStub* video_stub) {
-  reader_.Init(
-      session->video_channel(),
-      NewCallback(this, &ProtobufVideoReader::OnNewData));
+                               VideoStub* video_stub,
+                               const InitializedCallback& callback) {
+  initialized_callback_ = callback;
   video_stub_ = video_stub;
+
+  session->CreateStreamChannel(
+      kVideoChannelName,
+      base::Bind(&ProtobufVideoReader::OnChannelReady, base::Unretained(this)));
+}
+
+void ProtobufVideoReader::OnChannelReady(const std::string& name,
+                                         net::StreamSocket* socket) {
+  DCHECK_EQ(name, std::string(kVideoChannelName));
+  if (!socket) {
+    initialized_callback_.Run(false);
+    return;
+  }
+
+  DCHECK(!channel_.get());
+  channel_.reset(socket);
+  reader_.Init(socket, NewCallback(this, &ProtobufVideoReader::OnNewData));
+  initialized_callback_.Run(true);
 }
 
 void ProtobufVideoReader::OnNewData(VideoPacket* packet, Task* done_task) {
diff --git a/remoting/protocol/protobuf_video_reader.h b/remoting/protocol/protobuf_video_reader.h
index 8e8ce42..165a3a8 100644
--- a/remoting/protocol/protobuf_video_reader.h
+++ b/remoting/protocol/protobuf_video_reader.h
@@ -1,14 +1,19 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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_PROTOCOL_PROTOBUF_VIDEO_READER_H_
 #define REMOTING_PROTOCOL_PROTOBUF_VIDEO_READER_H_
 
+#include "base/compiler_specific.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/message_reader.h"
 #include "remoting/protocol/video_reader.h"
 
+namespace net {
+class StreamSocket;
+}  // namespace net
+
 namespace remoting {
 namespace protocol {
 
@@ -20,13 +25,21 @@ class ProtobufVideoReader : public VideoReader {
   virtual ~ProtobufVideoReader();
 
   // VideoReader interface.
-  virtual void Init(protocol::Session* session, VideoStub* video_stub);
+  virtual void Init(protocol::Session* session,
+                    VideoStub* video_stub,
+                    const InitializedCallback& callback) OVERRIDE;
 
  private:
+  void OnChannelReady(const std::string& name, net::StreamSocket* socket);
   void OnNewData(VideoPacket* packet, Task* done_task);
 
+  InitializedCallback initialized_callback_;
+
   VideoPacketFormat::Encoding encoding_;
 
+  // TODO(sergeyu): Remove |channel_| and let |reader_| own it.
+  scoped_ptr<net::StreamSocket> channel_;
+
   ProtobufMessageReader<VideoPacket> reader_;
 
   // The stub that processes all received packets.
diff --git a/remoting/protocol/protobuf_video_writer.cc b/remoting/protocol/protobuf_video_writer.cc
index a03a4e6..2d37f65 100644
--- a/remoting/protocol/protobuf_video_writer.cc
+++ b/remoting/protocol/protobuf_video_writer.cc
@@ -1,10 +1,13 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
 #include "remoting/protocol/protobuf_video_writer.h"
 
+#include "base/bind.h"
 #include "base/task.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/rtp_writer.h"
 #include "remoting/protocol/session.h"
@@ -17,14 +20,35 @@ ProtobufVideoWriter::ProtobufVideoWriter() { }
 
 ProtobufVideoWriter::~ProtobufVideoWriter() { }
 
-void ProtobufVideoWriter::Init(protocol::Session* session) {
+void ProtobufVideoWriter::Init(protocol::Session* session,
+                               const InitializedCallback& callback) {
+  initialized_callback_ = callback;
+
+  session->CreateStreamChannel(
+      kVideoChannelName,
+      base::Bind(&ProtobufVideoWriter::OnChannelReady, base::Unretained(this)));
+}
+
+void ProtobufVideoWriter::OnChannelReady(const std::string& name,
+                                         net::StreamSocket* socket) {
+  DCHECK_EQ(name, std::string(kVideoChannelName));
+  if (!socket) {
+    initialized_callback_.Run(false);
+    return;
+  }
+
+  DCHECK(!channel_.get());
+  channel_.reset(socket);
   buffered_writer_ = new BufferedSocketWriter();
   // TODO(sergeyu): Provide WriteFailedCallback for the buffered writer.
-  buffered_writer_->Init(session->video_channel(), NULL);
+  buffered_writer_->Init(socket, NULL);
+
+  initialized_callback_.Run(true);
 }
 
 void ProtobufVideoWriter::Close() {
   buffered_writer_->Close();
+  channel_.reset();
 }
 
 void ProtobufVideoWriter::ProcessVideoPacket(const VideoPacket* packet,
diff --git a/remoting/protocol/protobuf_video_writer.h b/remoting/protocol/protobuf_video_writer.h
index 189eacf..a5bb4fc 100644
--- a/remoting/protocol/protobuf_video_writer.h
+++ b/remoting/protocol/protobuf_video_writer.h
@@ -5,10 +5,17 @@
 #ifndef REMOTING_PROTOCOL_PROTOBUF_VIDEO_WRITER_H_
 #define REMOTING_PROTOCOL_PROTOBUF_VIDEO_WRITER_H_
 
+#include <string>
+
 #include "base/compiler_specific.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/scoped_ptr.h"
 #include "remoting/protocol/video_writer.h"
 
+namespace net {
+class StreamSocket;
+}  // namespace net
+
 namespace remoting {
 namespace protocol {
 
@@ -21,7 +28,8 @@ class ProtobufVideoWriter : public VideoWriter {
   virtual ~ProtobufVideoWriter();
 
   // VideoWriter interface.
-  virtual void Init(protocol::Session* session) OVERRIDE;
+  virtual void Init(protocol::Session* session,
+                    const InitializedCallback& callback) OVERRIDE;
   virtual void Close() OVERRIDE;
 
   // VideoStub interface.
@@ -30,6 +38,13 @@ class ProtobufVideoWriter : public VideoWriter {
   virtual int GetPendingPackets() OVERRIDE;
 
  private:
+  void OnChannelReady(const std::string& name, net::StreamSocket* socket);
+
+  InitializedCallback initialized_callback_;
+
+  // TODO(sergeyu): Remove |channel_| and let |buffered_writer_| own it.
+  scoped_ptr<net::StreamSocket> channel_;
+
   scoped_refptr<BufferedSocketWriter> buffered_writer_;
 
   DISALLOW_COPY_AND_ASSIGN(ProtobufVideoWriter);
diff --git a/remoting/protocol/rtp_video_reader.cc b/remoting/protocol/rtp_video_reader.cc
index b6d9e0b..9b97961 100644
--- a/remoting/protocol/rtp_video_reader.cc
+++ b/remoting/protocol/rtp_video_reader.cc
@@ -4,7 +4,9 @@
 
 #include "remoting/protocol/rtp_video_reader.h"
 
+#include "base/bind.h"
 #include "base/task.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/session.h"
 
@@ -22,7 +24,8 @@ RtpVideoReader::PacketsQueueEntry::PacketsQueueEntry()
 }
 
 RtpVideoReader::RtpVideoReader()
-    : last_sequence_number_(0),
+    : initialized_(false),
+      last_sequence_number_(0),
       video_stub_(NULL) {
 }
 
@@ -30,11 +33,47 @@ RtpVideoReader::~RtpVideoReader() {
   ResetQueue();
 }
 
-void RtpVideoReader::Init(protocol::Session* session, VideoStub* video_stub) {
-  rtp_reader_.Init(session->video_rtp_channel(),
-                   NewCallback(this, &RtpVideoReader::OnRtpPacket));
-  rtcp_writer_.Init(session->video_rtcp_channel());
+void RtpVideoReader::Init(protocol::Session* session,
+                          VideoStub* video_stub,
+                          const InitializedCallback& callback) {
+  initialized_callback_ = callback;
   video_stub_ = video_stub;
+
+  session->CreateDatagramChannel(
+      kVideoRtpChannelName,
+      base::Bind(&RtpVideoReader::OnChannelReady, base::Unretained(this)));
+  session->CreateDatagramChannel(
+      kVideoRtcpChannelName,
+      base::Bind(&RtpVideoReader::OnChannelReady, base::Unretained(this)));
+}
+
+void RtpVideoReader::OnChannelReady(const std::string& name,
+                                    net::Socket* socket) {
+  if (!socket) {
+    if (!initialized_) {
+      initialized_ = true;
+      initialized_callback_.Run(false);
+    }
+    return;
+  }
+
+  if (name == kVideoRtpChannelName) {
+    DCHECK(!rtp_channel_.get());
+    rtp_channel_.reset(socket);
+    rtp_reader_.Init(socket, NewCallback(this, &RtpVideoReader::OnRtpPacket));
+  } else if (name == kVideoRtcpChannelName) {
+    DCHECK(!rtcp_channel_.get());
+    rtcp_channel_.reset(socket);
+    rtcp_writer_.Init(socket);
+  } else {
+    NOTREACHED();
+  }
+
+  if (rtp_channel_.get() && rtcp_channel_.get()) {
+    DCHECK(!initialized_);
+    initialized_ = true;
+    initialized_callback_.Run(true);
+  }
 }
 
 void RtpVideoReader::ResetQueue() {
diff --git a/remoting/protocol/rtp_video_reader.h b/remoting/protocol/rtp_video_reader.h
index 8b0d3bd..e2eaf20e 100644
--- a/remoting/protocol/rtp_video_reader.h
+++ b/remoting/protocol/rtp_video_reader.h
@@ -5,7 +5,9 @@
 #ifndef REMOTING_PROTOCOL_RTP_VIDEO_READER_H_
 #define REMOTING_PROTOCOL_RTP_VIDEO_READER_H_
 
+#include "base/compiler_specific.h"
 #include "base/time.h"
+#include "base/memory/scoped_ptr.h"
 #include "remoting/protocol/rtcp_writer.h"
 #include "remoting/protocol/rtp_reader.h"
 #include "remoting/protocol/video_reader.h"
@@ -13,6 +15,8 @@
 namespace remoting {
 namespace protocol {
 
+class RtcpWriter;
+class RtpReader;
 class Session;
 
 class RtpVideoReader : public VideoReader {
@@ -21,7 +25,9 @@ class RtpVideoReader : public VideoReader {
   virtual ~RtpVideoReader();
 
   // VideoReader interface.
-  virtual void Init(protocol::Session* session, VideoStub* video_stub);
+  virtual void Init(protocol::Session* session,
+                    VideoStub* video_stub,
+                    const InitializedCallback& callback) OVERRIDE;
 
  private:
   friend class RtpVideoReaderTest;
@@ -44,6 +50,8 @@ class RtpVideoReader : public VideoReader {
 
   typedef std::deque<PacketsQueueEntry> PacketsQueue;
 
+  void OnChannelReady(const std::string& name, net::Socket* socket);
+
   void OnRtpPacket(const RtpPacket* rtp_packet);
   void CheckFullPacket(const PacketsQueue::iterator& pos);
   void RebuildVideoPacket(const PacketsQueue::iterator& from,
@@ -56,7 +64,12 @@ class RtpVideoReader : public VideoReader {
   // |kReceiverReportsIntervalMs|.
   void SendReceiverReportIf();
 
+  bool initialized_;
+  InitializedCallback initialized_callback_;
+
+  scoped_ptr<net::Socket> rtp_channel_;
   RtpReader rtp_reader_;
+  scoped_ptr<net::Socket> rtcp_channel_;
   RtcpWriter rtcp_writer_;
 
   PacketsQueue packets_queue_;
diff --git a/remoting/protocol/rtp_video_reader_unittest.cc b/remoting/protocol/rtp_video_reader_unittest.cc
index 2475d6a..3c6d5ef 100644
--- a/remoting/protocol/rtp_video_reader_unittest.cc
+++ b/remoting/protocol/rtp_video_reader_unittest.cc
@@ -4,6 +4,7 @@
 
 #include <vector>
 
+#include "base/bind.h"
 #include "base/message_loop.h"
 #include "base/string_number_conversions.h"
 #include "net/base/io_buffer.h"
@@ -65,10 +66,16 @@ class RtpVideoReaderTest : public testing::Test,
   void Reset() {
     session_.reset(new FakeSession());
     reader_.reset(new RtpVideoReader());
-    reader_->Init(session_.get(), this);
+    reader_->Init(session_.get(), this,
+                  base::Bind(&RtpVideoReaderTest::OnReaderInitialized,
+                             base::Unretained(this)));
     received_packets_.clear();
   }
 
+  void OnReaderInitialized(bool success) {
+    ASSERT_TRUE(success);
+  }
+
   void InitData(int size) {
     data_.resize(size);
     for (int i = 0; i < size; ++i) {
diff --git a/remoting/protocol/rtp_video_writer.cc b/remoting/protocol/rtp_video_writer.cc
index d5d4b75..2a0f950 100644
--- a/remoting/protocol/rtp_video_writer.cc
+++ b/remoting/protocol/rtp_video_writer.cc
@@ -4,9 +4,11 @@
 
 #include "remoting/protocol/rtp_video_writer.h"
 
+#include "base/bind.h"
 #include "base/task.h"
 #include "net/base/io_buffer.h"
 #include "remoting/base/compound_buffer.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/rtp_writer.h"
 #include "remoting/protocol/session.h"
@@ -18,18 +20,56 @@ namespace {
 const int kMtu = 1200;
 }  // namespace
 
-RtpVideoWriter::RtpVideoWriter() { }
+RtpVideoWriter::RtpVideoWriter()
+    : initialized_(false) {
+}
 
 RtpVideoWriter::~RtpVideoWriter() {
   Close();
 }
 
-void RtpVideoWriter::Init(protocol::Session* session) {
-  rtp_writer_.Init(session->video_rtp_channel());
+void RtpVideoWriter::Init(protocol::Session* session,
+                          const InitializedCallback& callback) {
+  initialized_callback_ = callback;
+  session->CreateDatagramChannel(
+      kVideoRtpChannelName,
+      base::Bind(&RtpVideoWriter::OnChannelReady, base::Unretained(this)));
+  session->CreateDatagramChannel(
+      kVideoRtcpChannelName,
+      base::Bind(&RtpVideoWriter::OnChannelReady, base::Unretained(this)));
+}
+
+void RtpVideoWriter::OnChannelReady(const std::string& name,
+                                    net::Socket* socket) {
+  if (!socket) {
+    if (!initialized_) {
+      initialized_ = true;
+      initialized_callback_.Run(false);
+    }
+    return;
+  }
+
+  if (name == kVideoRtpChannelName) {
+    DCHECK(!rtp_channel_.get());
+    rtp_channel_.reset(socket);
+    rtp_writer_.Init(socket);
+  } else if (name == kVideoRtcpChannelName) {
+    DCHECK(!rtcp_channel_.get());
+    rtcp_channel_.reset(socket);
+    // TODO(sergeyu): Use RTCP channel somehow.
+  }
+
+  if (rtp_channel_.get() && rtcp_channel_.get()) {
+    DCHECK(!initialized_);
+    initialized_ = true;
+    initialized_callback_.Run(true);
+  }
 }
 
 void RtpVideoWriter::Close() {
   rtp_writer_.Close();
+  rtp_channel_.reset();
+  rtcp_channel_.reset();
 }
 
 void RtpVideoWriter::ProcessVideoPacket(const VideoPacket* packet, Task* done) {
diff --git a/remoting/protocol/rtp_video_writer.h b/remoting/protocol/rtp_video_writer.h
index d54d721..fd7349b2 100644
--- a/remoting/protocol/rtp_video_writer.h
+++ b/remoting/protocol/rtp_video_writer.h
@@ -5,6 +5,7 @@
 #ifndef REMOTING_PROTOCOL_RTP_VIDEO_WRITER_H_
 #define REMOTING_PROTOCOL_RTP_VIDEO_WRITER_H_
 
+#include "base/memory/scoped_ptr.h"
 #include "remoting/protocol/rtp_writer.h"
 #include "remoting/protocol/video_writer.h"
 
@@ -19,7 +20,8 @@ class RtpVideoWriter : public VideoWriter {
   virtual ~RtpVideoWriter();
 
   // VideoWriter interface.
-  virtual void Init(protocol::Session* session) OVERRIDE;
+  virtual void Init(protocol::Session* session,
+                    const InitializedCallback& callback) OVERRIDE;
   virtual void Close() OVERRIDE;
 
   // VideoStub interface.
@@ -28,7 +30,14 @@ class RtpVideoWriter : public VideoWriter {
   virtual int GetPendingPackets() OVERRIDE;
 
  private:
+  void OnChannelReady(const std::string& name, net::Socket* socket);
+
+  bool initialized_;
+  InitializedCallback initialized_callback_;
+
+  scoped_ptr<net::Socket> rtp_channel_;
   RtpWriter rtp_writer_;
+  scoped_ptr<net::Socket> rtcp_channel_;
 
   DISALLOW_COPY_AND_ASSIGN(RtpVideoWriter);
 };
diff --git a/remoting/protocol/rtp_video_writer_unittest.cc b/remoting/protocol/rtp_video_writer_unittest.cc
index 2724927..118c300 100644
--- a/remoting/protocol/rtp_video_writer_unittest.cc
+++ b/remoting/protocol/rtp_video_writer_unittest.cc
@@ -5,8 +5,10 @@
 #include <string>
 #include <vector>
 
+#include "base/bind.h"
 #include "base/message_loop.h"
 #include "base/string_number_conversions.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/video.pb.h"
 #include "remoting/protocol/fake_session.h"
 #include "remoting/protocol/rtp_reader.h"
@@ -57,7 +59,13 @@ class RtpVideoWriterTest : public testing::Test {
 
   virtual void SetUp() {
     session_.reset(new FakeSession());
-    writer_.Init(session_.get());
+    writer_.Init(session_.get(),
+                 base::Bind(&RtpVideoWriterTest::OnWriterInitialized,
+                            base::Unretained(this)));
+  }
+
+  void OnWriterInitialized(bool success) {
+    ASSERT_TRUE(success);
   }
 
   void InitData(int size) {
@@ -88,7 +96,7 @@ class RtpVideoWriterTest : public testing::Test {
   void VerifyResult(const ExpectedPacket expected[],
                     int count) {
     const vector<string>& rtp_packets =
-        session_->video_rtp_channel()->written_packets();
+        session_->GetDatagramChannel(kVideoRtpChannelName)->written_packets();
     ASSERT_EQ(count, static_cast<int>(rtp_packets.size()));
     int pos = 0;
     for (int i = 0; i < count; ++i) {
diff --git a/remoting/protocol/session.h b/remoting/protocol/session.h
index fc40a6d..89cc266 100644
--- a/remoting/protocol/session.h
+++ b/remoting/protocol/session.h
@@ -64,9 +64,6 @@ class Session : public base::NonThreadSafe {
   // instead.
   virtual net::Socket* control_channel() = 0;
   virtual net::Socket* event_channel() = 0;
-  virtual net::Socket* video_channel() = 0;
-  virtual net::Socket* video_rtp_channel() = 0;
-  virtual net::Socket* video_rtcp_channel() = 0;
 
   // JID of the other side.
   virtual const std::string& jid() = 0;
diff --git a/remoting/protocol/video_reader.h b/remoting/protocol/video_reader.h
index ba39be3..e539c0d 100644
--- a/remoting/protocol/video_reader.h
+++ b/remoting/protocol/video_reader.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -25,12 +25,17 @@ class VideoReader {
  public:
   static VideoReader* Create(const SessionConfig* config);
 
+  // The callback is called when initialization is finished. The
+  // parameter is set to true on success.
+  typedef base::Callback<void(bool)> InitializedCallback;
+
   virtual ~VideoReader();
 
   // Initializies the reader. Doesn't take ownership of either |connection|
   // or |video_stub|.
   virtual void Init(Session* session,
-                    VideoStub* video_stub) = 0;
+                    VideoStub* video_stub,
+                    const InitializedCallback& callback) = 0;
 
  protected:
   VideoReader() { }
diff --git a/remoting/protocol/video_writer.h b/remoting/protocol/video_writer.h
index be4c715..68b19fd 100644
--- a/remoting/protocol/video_writer.h
+++ b/remoting/protocol/video_writer.h
@@ -1,4 +1,4 @@
-// Copyright (c) 2010 The Chromium Authors. All rights reserved.
+// 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.
 
@@ -11,6 +11,7 @@
 #define REMOTING_PROTOCOL_VIDEO_WRITER_H_
 
 #include "base/basictypes.h"
+#include "base/callback.h"
 #include "remoting/protocol/video_stub.h"
 
 namespace remoting {
@@ -23,10 +24,14 @@ class VideoWriter : public VideoStub {
  public:
   virtual ~VideoWriter();
 
+  // The callback is called when initialization is finished. The
+  // parameter is set to true on success.
+  typedef base::Callback<void(bool)> InitializedCallback;
+
   static VideoWriter* Create(const SessionConfig* config);
 
   // Initializes the writer.
-  virtual void Init(Session* session) = 0;
+  virtual void Init(Session* session, const InitializedCallback& callback) = 0;
 
   // Stops writing. Must be called on the network thread before this
   // object is destroyed.