Remove event_channel() and control_channel() from Session interface.

Now all channels are created using CreateStreamChannel() as they should be!

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

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

remoting/protocol/channel_dispatcher_base.cc

+// 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/channel_dispatcher_base.h"
+
+#include "base/bind.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/protocol/session.h"
+
+namespace remoting {
+namespace protocol {
+
+ChannelDispatcherBase::ChannelDispatcherBase(const char* channel_name)
+    : channel_name_(channel_name),
+      session_(NULL) {
+}
+
+ChannelDispatcherBase::~ChannelDispatcherBase() {
+  if (session_)
+    session_->CancelChannelCreation(channel_name_);
+}
+
+void ChannelDispatcherBase::Init(Session* session,
+                                 const InitializedCallback& callback) {
+  DCHECK(session);
+  session_ = session;
+  initialized_callback_ = callback;
+
+  session_->CreateStreamChannel(channel_name_, base::Bind(
+      &ChannelDispatcherBase::OnChannelReady, base::Unretained(this)));
+}
+
+void ChannelDispatcherBase::OnChannelReady(net::StreamSocket* socket) {
+  if (!socket) {
+    initialized_callback_.Run(false);
+    return;
+  }
+
+  channel_.reset(socket);
+
+  OnInitialized();
+
+  initialized_callback_.Run(true);
+}
+
+}  // namespace protocol
+}  // namespace remoting

remoting/protocol/channel_dispatcher_base.h

+// 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_CHANNEL_DISPATCHER_BASE_H_
+#define REMOTING_PROTOCOL_CHANNEL_DISPATCHER_BASE_H_
+
+#include <string>
+
+#include "base/basictypes.h"
+#include "base/callback.h"
+#include "base/memory/scoped_ptr.h"
+
+namespace net {
+class StreamSocket;
+}  // namespace net
+
+namespace remoting {
+namespace protocol {
+
+class Session;
+
+// Base class for channel message dispatchers. It's responsible for
+// creating the named channel. Derived dispatchers then dispatch
+// incoming messages on this channel as well as send outgoing
+// messages.
+class ChannelDispatcherBase {
+ public:
+  // The callback is called when initialization is finished. The
+  // parameter is set to true on success.
+  typedef base::Callback<void(bool)> InitializedCallback;
+
+  virtual ~ChannelDispatcherBase();
+
+  // Creates and connects the channel in the specified
+  // |session|. Caller retains ownership of the Session.
+  void Init(Session* session, const InitializedCallback& callback);
+
+  // Returns true if the channel is currently connected.
+  bool is_connected() { return channel() != NULL; }
+
+ protected:
+  explicit ChannelDispatcherBase(const char* channel_name);
+
+  net::StreamSocket* channel() { return channel_.get(); }
+
+  // Called when channel is initialized. Must be overriden in the
+  // child classes. Should not delete the dispatcher.
+  virtual void OnInitialized() = 0;
+
+ private:
+  void OnChannelReady(net::StreamSocket* socket);
+
+  std::string channel_name_;
+  Session* session_;
+  InitializedCallback initialized_callback_;
+  scoped_ptr<net::StreamSocket> channel_;
+
+  DISALLOW_COPY_AND_ASSIGN(ChannelDispatcherBase);
+};
+
+}  // namespace protocol
+}  // namespace remoting
+
+#endif  // REMOTING_PROTOCOL_CHANNEL_DISPATCHER_BASE_H_

remoting/protocol/client_control_dispatcher.cc

@@ -4,22 +4,20 @@
 
 #include "remoting/protocol/client_control_dispatcher.h"
 
-#include "base/memory/ref_counted.h"
 #include "base/message_loop_proxy.h"
-#include "net/base/io_buffer.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/control.pb.h"
-#include "remoting/proto/event.pb.h"
 #include "remoting/proto/internal.pb.h"
+#include "remoting/protocol/buffered_socket_writer.h"
 #include "remoting/protocol/client_stub.h"
-#include "remoting/protocol/input_stub.h"
-#include "remoting/protocol/message_reader.h"
-#include "remoting/protocol/session.h"
 
 namespace remoting {
 namespace protocol {
 
 ClientControlDispatcher::ClientControlDispatcher()
-    : client_stub_(NULL),
+    : ChannelDispatcherBase(kControlChannelName),
+      client_stub_(NULL),
       writer_(new BufferedSocketWriter(base::MessageLoopProxy::current())) {
 }
 
@@ -27,13 +25,10 @@ ClientControlDispatcher::~ClientControlDispatcher() {
   writer_->Close();
 }
 
-void ClientControlDispatcher::Init(protocol::Session* session) {
-  DCHECK(session);
-
+void ClientControlDispatcher::OnInitialized() {
   // TODO(garykac): Set write failed callback.
-  writer_->Init(session->control_channel(),
-                BufferedSocketWriter::WriteFailedCallback());
-  reader_.Init(session->control_channel(), base::Bind(
+  writer_->Init(channel(), BufferedSocketWriter::WriteFailedCallback());
+  reader_.Init(channel(), base::Bind(
       &ClientControlDispatcher::OnMessageReceived, base::Unretained(this)));
 }
 

remoting/protocol/client_control_dispatcher.h

@@ -5,8 +5,8 @@
 #ifndef REMOTING_PROTOCOL_CLIENT_CONTROL_DISPATCHER_H_
 #define REMOTING_PROTOCOL_CLIENT_CONTROL_DISPATCHER_H_
 
-#include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
+#include "remoting/protocol/channel_dispatcher_base.h"
 #include "remoting/protocol/host_stub.h"
 #include "remoting/protocol/message_reader.h"
 
@@ -25,20 +25,20 @@ class Session;
 // ClientControlDispatcher dispatches incoming messages on the control
 // channel to ClientStub, and also implements HostStub for outgoing
 // messages.
-class ClientControlDispatcher : public HostStub {
+class ClientControlDispatcher : public ChannelDispatcherBase, public HostStub {
  public:
   ClientControlDispatcher();
   virtual ~ClientControlDispatcher();
 
-  // Initialize the control channel and the dispatcher for the
-  // |session|. Doesn't take ownership of |session|.
-  void Init(protocol::Session* session);
-
   // Sets ClientStub that will be called for each incoming control
   // message. Doesn't take ownership of |client_stub|. It must outlive
   // this dispatcher.
   void set_client_stub(ClientStub* client_stub) { client_stub_ = client_stub; }
 
+ protected:
+  // ChannelDispatcherBase overrides.
+  virtual void OnInitialized() OVERRIDE;
+
  private:
   void OnMessageReceived(ControlMessage* message,
                          const base::Closure& done_task);

remoting/protocol/client_event_dispatcher.cc

@@ -6,28 +6,28 @@
 
 #include "base/message_loop_proxy.h"
 #include "base/time.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/event.pb.h"
 #include "remoting/proto/internal.pb.h"
 #include "remoting/protocol/buffered_socket_writer.h"
-#include "remoting/protocol/session.h"
 #include "remoting/protocol/util.h"
 
 namespace remoting {
 namespace protocol {
 
 ClientEventDispatcher::ClientEventDispatcher()
-    : writer_(new BufferedSocketWriter(base::MessageLoopProxy::current())) {
+    : ChannelDispatcherBase(kEventChannelName),
+      writer_(new BufferedSocketWriter(base::MessageLoopProxy::current())) {
 }
 
 ClientEventDispatcher::~ClientEventDispatcher() {
   writer_->Close();
 }
 
-void ClientEventDispatcher::Init(Session* session) {
-  DCHECK(session);
-
+void ClientEventDispatcher::OnInitialized() {
   // TODO(garykac): Set write failed callback.
-  writer_->Init(session->event_channel(),
+  writer_->Init(channel(),
                 BufferedSocketWriter::WriteFailedCallback());
 }
 

remoting/protocol/client_event_dispatcher.h

@@ -5,35 +5,30 @@
 #ifndef REMOTING_PROTOCOL_CLIENT_INPUT_DISPATCHER_H_
 #define REMOTING_PROTOCOL_CLIENT_INPUT_DISPATCHER_H_
 
-#include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
+#include "remoting/protocol/channel_dispatcher_base.h"
 #include "remoting/protocol/input_stub.h"
 
-namespace base {
-class MessageLoopProxy;
-}  // namespace base
-
 namespace remoting {
 namespace protocol {
 
 class BufferedSocketWriter;
-class Session;
 
 // ClientEventDispatcher manages the event channel on the client
 // side. It implements InputStub for outgoing input messages.
-class ClientEventDispatcher : public InputStub {
+class ClientEventDispatcher : public ChannelDispatcherBase, public InputStub {
  public:
   ClientEventDispatcher();
   virtual ~ClientEventDispatcher();
 
-  // Initialize the event channel and the dispatcher for the
-  // |session|.
-  void Init(Session* session);
-
   // InputStub implementation.
   virtual void InjectKeyEvent(const KeyEvent& event) OVERRIDE;
   virtual void InjectMouseEvent(const MouseEvent& event) OVERRIDE;
 
+ protected:
+  // ChannelDispatcherBase overrides.
+  virtual void OnInitialized() OVERRIDE;
+
  private:
   scoped_refptr<BufferedSocketWriter> writer_;
 

remoting/protocol/connection_to_client.cc

@@ -21,10 +21,7 @@ ConnectionToClient::ConnectionToClient(protocol::Session* session)
     : handler_(NULL),
       host_stub_(NULL),
       input_stub_(NULL),
-      session_(session),
-      control_connected_(false),
-      input_connected_(false),
-      video_connected_(false) {
+      session_(session) {
   session_->SetStateChangeCallback(
       base::Bind(&ConnectionToClient::OnSessionStateChange,
                  base::Unretained(this)));
@@ -101,27 +98,24 @@ void ConnectionToClient::OnSessionStateChange(protocol::Session::State state) {
       break;
 
     case protocol::Session::CONNECTED:
-      video_writer_.reset(
-          VideoWriter::Create(base::MessageLoopProxy::current(),
-                              session_->config()));
-      video_writer_->Init(
-          session_.get(), base::Bind(&ConnectionToClient::OnVideoInitialized,
-                                     base::Unretained(this)));
-      break;
-
-    case protocol::Session::CONNECTED_CHANNELS:
+      // Initialize channels.
       control_dispatcher_.reset(new HostControlDispatcher());
-      control_dispatcher_->Init(session_.get());
+      control_dispatcher_->Init(session_.get(), base::Bind(
+          &ConnectionToClient::OnChannelInitialized, base::Unretained(this)));
       control_dispatcher_->set_host_stub(host_stub_);
-      input_dispatcher_.reset(new HostEventDispatcher());
-      input_dispatcher_->Init(session_.get());
-      input_dispatcher_->set_input_stub(input_stub_);
-      input_dispatcher_->set_sequence_number_callback(base::Bind(
+
+      event_dispatcher_.reset(new HostEventDispatcher());
+      event_dispatcher_->Init(session_.get(), base::Bind(
+          &ConnectionToClient::OnChannelInitialized, base::Unretained(this)));
+      event_dispatcher_->set_input_stub(input_stub_);
+      event_dispatcher_->set_sequence_number_callback(base::Bind(
           &ConnectionToClient::UpdateSequenceNumber, base::Unretained(this)));
 
-      control_connected_ = true;
-      input_connected_ = true;
-      NotifyIfChannelsReady();
+      video_writer_.reset(VideoWriter::Create(
+          base::MessageLoopProxy::current(), session_->config()));
+      video_writer_->Init(session_.get(), base::Bind(
+          &ConnectionToClient::OnChannelInitialized, base::Unretained(this)));
+
       break;
 
     case protocol::Session::CLOSED:
@@ -139,24 +133,26 @@ void ConnectionToClient::OnSessionStateChange(protocol::Session::State state) {
   }
 }
 
-void ConnectionToClient::OnVideoInitialized(bool successful) {
+void ConnectionToClient::OnChannelInitialized(bool successful) {
   DCHECK(CalledOnValidThread());
 
   if (!successful) {
-    LOG(ERROR) << "Failed to connect video channel";
+    LOG(ERROR) << "Failed to connect a channel";
     CloseOnError();
     return;
   }
 
-  video_connected_ = true;
   NotifyIfChannelsReady();
 }
 
 void ConnectionToClient::NotifyIfChannelsReady() {
   DCHECK(CalledOnValidThread());
 
-  if (control_connected_ && input_connected_ && video_connected_)
+  if (control_dispatcher_.get() && control_dispatcher_->is_connected() &&
+      event_dispatcher_.get() && event_dispatcher_->is_connected() &&
+      video_writer_.get() && video_writer_->is_connected()) {
     handler_->OnConnectionOpened(this);
+  }
 }
 
 void ConnectionToClient::CloseOnError() {
@@ -166,7 +162,7 @@ void ConnectionToClient::CloseOnError() {
 
 void ConnectionToClient::CloseChannels() {
   control_dispatcher_.reset();
-  input_dispatcher_.reset();
+  event_dispatcher_.reset();
   video_writer_.reset();
 }
 

remoting/protocol/connection_to_client.h

@@ -80,8 +80,8 @@ class ConnectionToClient : public base::NonThreadSafe {
   // Callback for protocol Session.
   void OnSessionStateChange(Session::State state);
 
-  // Callback for VideoReader::Init().
-  void OnVideoInitialized(bool successful);
+  // Callback for channel initialization.
+  void OnChannelInitialized(bool successful);
 
   void NotifyIfChannelsReady();
 
@@ -100,14 +100,9 @@ class ConnectionToClient : public base::NonThreadSafe {
   // The libjingle channel used to send and receive data from the remote client.
   scoped_ptr<Session> session_;
 
-  scoped_ptr<VideoWriter> video_writer_;
   scoped_ptr<HostControlDispatcher> control_dispatcher_;
-  scoped_ptr<HostEventDispatcher> input_dispatcher_;
-
-  // State of the channels.
-  bool control_connected_;
-  bool input_connected_;
-  bool video_connected_;
+  scoped_ptr<HostEventDispatcher> event_dispatcher_;
+  scoped_ptr<VideoWriter> video_writer_;
 
   DISALLOW_COPY_AND_ASSIGN(ConnectionToClient);
 };

remoting/protocol/connection_to_client_unittest.cc

@@ -37,8 +37,6 @@ class ConnectionToClientTest : public testing::Test {
     EXPECT_CALL(handler_, OnConnectionOpened(viewer_.get()));
     session_->state_change_callback().Run(
         protocol::Session::CONNECTED);
-    session_->state_change_callback().Run(
-        protocol::Session::CONNECTED_CHANNELS);
     message_loop_.RunAllPending();
   }
 

remoting/protocol/connection_to_host.cc

@@ -35,17 +35,14 @@ ConnectionToHost::ConnectionToHost(
       client_stub_(NULL),
       video_stub_(NULL),
       state_(CONNECTING),
-      error_(OK),
-      control_connected_(false),
-      input_connected_(false),
-      video_connected_(false) {
+      error_(OK) {
 }
 
 ConnectionToHost::~ConnectionToHost() {
 }
 
 InputStub* ConnectionToHost::input_stub() {
-  return input_dispatcher_.get();
+  return event_dispatcher_.get();
 }
 
 HostStub* ConnectionToHost::host_stub() {
@@ -190,24 +187,19 @@ void ConnectionToHost::OnSessionStateChange(
       break;
 
     case Session::CONNECTED:
-      video_reader_.reset(
-          VideoReader::Create(message_loop_, session_->config()));
-      video_reader_->Init(
-          session_.get(), video_stub_,
-          base::Bind(&ConnectionToHost::OnVideoChannelInitialized,
-                     base::Unretained(this)));
-      break;
+      video_reader_.reset(VideoReader::Create(
+          message_loop_, session_->config()));
+      video_reader_->Init(session_.get(), video_stub_, base::Bind(
+          &ConnectionToHost::OnChannelInitialized, base::Unretained(this)));
 
-    case Session::CONNECTED_CHANNELS:
       control_dispatcher_.reset(new ClientControlDispatcher());
-      control_dispatcher_->Init(session_.get());
+      control_dispatcher_->Init(session_.get(), base::Bind(
+          &ConnectionToHost::OnChannelInitialized, base::Unretained(this)));
       control_dispatcher_->set_client_stub(client_stub_);
-      input_dispatcher_.reset(new ClientEventDispatcher());
-      input_dispatcher_->Init(session_.get());
 
-      control_connected_ = true;
-      input_connected_ = true;
-      NotifyIfChannelsReady();
+      event_dispatcher_.reset(new ClientEventDispatcher());
+      event_dispatcher_->Init(session_.get(), base::Bind(
+          &ConnectionToHost::OnChannelInitialized, base::Unretained(this)));
       break;
 
     default:
@@ -216,19 +208,20 @@ void ConnectionToHost::OnSessionStateChange(
   }
 }
 
-void ConnectionToHost::OnVideoChannelInitialized(bool successful) {
+void ConnectionToHost::OnChannelInitialized(bool successful) {
   if (!successful) {
     LOG(ERROR) << "Failed to connect video channel";
     CloseOnError(NETWORK_FAILURE);
     return;
   }
 
-  video_connected_ = true;
   NotifyIfChannelsReady();
 }
 
 void ConnectionToHost::NotifyIfChannelsReady() {
-  if (control_connected_ && input_connected_ && video_connected_ &&
+  if (control_dispatcher_.get() && control_dispatcher_->is_connected() &&
+      event_dispatcher_.get() && event_dispatcher_->is_connected() &&
+      video_reader_.get() && video_reader_->is_connected() &&
       state_ == CONNECTING) {
     SetState(CONNECTED, OK);
     SetState(AUTHENTICATED, OK);
@@ -242,7 +235,7 @@ void ConnectionToHost::CloseOnError(Error error) {
 
 void ConnectionToHost::CloseChannels() {
   control_dispatcher_.reset();
-  input_dispatcher_.reset();
+  event_dispatcher_.reset();
   video_reader_.reset();
 }
 

remoting/protocol/connection_to_host.h

@@ -117,8 +117,8 @@ class ConnectionToHost : public SignalStrategy::StatusObserver,
   // Callback for |session_|.
   void OnSessionStateChange(Session::State state);
 
-  // Callback for VideoReader::Init().
-  void OnVideoChannelInitialized(bool successful);
+  // Callbacks for channel initialization
+  void OnChannelInitialized(bool successful);
 
   void NotifyIfChannelsReady();
 
@@ -153,17 +153,12 @@ class ConnectionToHost : public SignalStrategy::StatusObserver,
 
   scoped_ptr<VideoReader> video_reader_;
   scoped_ptr<ClientControlDispatcher> control_dispatcher_;
-  scoped_ptr<ClientEventDispatcher> input_dispatcher_;
+  scoped_ptr<ClientEventDispatcher> event_dispatcher_;
 
   // Internal state of the connection.
   State state_;
   Error error_;
 
-  // State of the channels.
-  bool control_connected_;
-  bool input_connected_;
-  bool video_connected_;
-
  private:
   DISALLOW_COPY_AND_ASSIGN(ConnectionToHost);
 };

remoting/protocol/fake_session.cc

@@ -245,14 +245,6 @@ void FakeSession::CreateDatagramChannel(
 void FakeSession::CancelChannelCreation(const std::string& name) {
 }
 
-FakeSocket* FakeSession::control_channel() {
-  return &control_channel_;
-}
-
-FakeSocket* FakeSession::event_channel() {
-  return &event_channel_;
-}
-
 const std::string& FakeSession::jid() {
   return jid_;
 }

remoting/protocol/fake_session.h

@@ -150,9 +150,6 @@ class FakeSession : public Session {
       const DatagramChannelCallback& callback) OVERRIDE;
   virtual void CancelChannelCreation(const std::string& name) OVERRIDE;
 
-  virtual FakeSocket* control_channel() OVERRIDE;
-  virtual FakeSocket* event_channel() OVERRIDE;
-
   virtual const std::string& jid() OVERRIDE;
 
   virtual const CandidateSessionConfig* candidate_config() OVERRIDE;
@@ -174,8 +171,6 @@ class FakeSession : public Session {
   scoped_ptr<const CandidateSessionConfig> candidate_config_;
   SessionConfig config_;
   MessageLoop* message_loop_;
-  FakeSocket control_channel_;
-  FakeSocket event_channel_;
 
   std::map<std::string, FakeSocket*> stream_channels_;
   std::map<std::string, FakeUdpSocket*> datagram_channels_;

remoting/protocol/host_control_dispatcher.cc

@@ -5,18 +5,20 @@
 #include "remoting/protocol/host_control_dispatcher.h"
 
 #include "base/message_loop_proxy.h"
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/control.pb.h"
 #include "remoting/proto/internal.pb.h"
 #include "remoting/protocol/buffered_socket_writer.h"
 #include "remoting/protocol/host_stub.h"
-#include "remoting/protocol/session.h"
 #include "remoting/protocol/util.h"
 
 namespace remoting {
 namespace protocol {
 
 HostControlDispatcher::HostControlDispatcher()
-    : host_stub_(NULL),
+    : ChannelDispatcherBase(kControlChannelName),
+      host_stub_(NULL),
       writer_(new BufferedSocketWriter(base::MessageLoopProxy::current())) {
 }
 
@@ -24,13 +26,10 @@ HostControlDispatcher::~HostControlDispatcher() {
   writer_->Close();
 }
 
-void HostControlDispatcher::Init(Session* session) {
-  DCHECK(session);
-
-  reader_.Init(session->control_channel(), base::Bind(
+void HostControlDispatcher::OnInitialized() {
+  reader_.Init(channel(), base::Bind(
       &HostControlDispatcher::OnMessageReceived, base::Unretained(this)));
-  writer_->Init(session->control_channel(),
-                BufferedSocketWriter::WriteFailedCallback());
+  writer_->Init(channel(), BufferedSocketWriter::WriteFailedCallback());
 
   // Write legacy BeginSession message.
   // TODO(sergeyu): Remove it. See http://crbug.com/104670 .

remoting/protocol/host_control_dispatcher.h

@@ -5,8 +5,8 @@
 #ifndef REMOTING_PROTOCOL_HOST_CONTROL_DISPATCHER_H_
 #define REMOTING_PROTOCOL_HOST_CONTROL_DISPATCHER_H_
 
-#include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
+#include "remoting/protocol/channel_dispatcher_base.h"
 #include "remoting/protocol/client_stub.h"
 #include "remoting/protocol/message_reader.h"
 
@@ -14,6 +14,10 @@ namespace base {
 class MessageLoopProxy;
 }  // namespace base
 
+namespace net {
+class StreamSocket;
+}  // namespace net
+
 namespace remoting {
 namespace protocol {
 
@@ -25,22 +29,21 @@ class Session;
 // HostControlDispatcher dispatches incoming messages on the control
 // channel to HostStub, and also implements ClientStub for outgoing
 // messages.
-class HostControlDispatcher : public ClientStub {
+class HostControlDispatcher : public ChannelDispatcherBase, public ClientStub {
  public:
   HostControlDispatcher();
   virtual ~HostControlDispatcher();
 
-  // Initialize the control channel and the dispatcher for the
-  // |session|. Doesn't take ownership of |session|.
-  void Init(Session* session);
-
   // Sets HostStub that will be called for each incoming control
   // message. Doesn't take ownership of |host_stub|. It must outlive
   // this dispatcher.
   void set_host_stub(HostStub* host_stub) { host_stub_ = host_stub; }
 
+ protected:
+  // ChannelDispatcherBase overrides.
+  virtual void OnInitialized() OVERRIDE;
+
  private:
-  // This method is called by |reader_| when a message is received.
   void OnMessageReceived(ControlMessage* message,
                          const base::Closure& done_task);
 

remoting/protocol/host_event_dispatcher.cc

@@ -4,24 +4,25 @@
 
 #include "remoting/protocol/host_event_dispatcher.h"
 
+#include "net/socket/stream_socket.h"
+#include "remoting/base/constants.h"
 #include "remoting/proto/event.pb.h"
 #include "remoting/proto/internal.pb.h"
 #include "remoting/protocol/input_stub.h"
-#include "remoting/protocol/session.h"
 
 namespace remoting {
 namespace protocol {
 
 HostEventDispatcher::HostEventDispatcher()
-    : input_stub_(NULL) {
+    : ChannelDispatcherBase(kEventChannelName),
+      input_stub_(NULL) {
 }
 
 HostEventDispatcher::~HostEventDispatcher() {
 }
 
-void HostEventDispatcher::Init(Session* session) {
-  DCHECK(session);
-  reader_.Init(session->event_channel(), base::Bind(
+void HostEventDispatcher::OnInitialized() {
+  reader_.Init(channel(), base::Bind(
       &HostEventDispatcher::OnMessageReceived, base::Unretained(this)));
 }
 

remoting/protocol/host_event_dispatcher.h

@@ -5,7 +5,7 @@
 #ifndef REMOTING_PROTOCOL_HOST_EVENT_DISPATCHER_H_
 #define REMOTING_PROTOCOL_HOST_EVENT_DISPATCHER_H_
 
-#include "base/basictypes.h"
+#include "remoting/protocol/channel_dispatcher_base.h"
 #include "remoting/protocol/message_reader.h"
 
 namespace remoting {
@@ -13,21 +13,16 @@ namespace protocol {
 
 class EventMessage;
 class InputStub;
-class Session;
 
 // HostEventDispatcher dispatches incoming messages on the event
 // channel to InputStub.
-class HostEventDispatcher {
+class HostEventDispatcher : public ChannelDispatcherBase {
  public:
   typedef base::Callback<void(int64)> SequenceNumberCallback;
 
   HostEventDispatcher();
   virtual ~HostEventDispatcher();
 
-  // Initialize the event channel and the dispatcher for the
-  // |session|. Caller retains ownership of |session|.
-  void Init(Session* session);
-
   // Set InputStub that will be called for each incoming input
   // message. Doesn't take ownership of |input_stub|. It must outlive
   // the dispatcher.
@@ -39,8 +34,11 @@ class HostEventDispatcher {
     sequence_number_callback_ = value;
   }
 
+ protected:
+  // ChannelDispatcherBase overrides.
+  virtual void OnInitialized() OVERRIDE;
+
  private:
-  // This method is called by |reader_| when a message is received.
   void OnMessageReceived(EventMessage* message,
                          const base::Closure& done_task);
 

remoting/protocol/jingle_session.cc

@@ -88,8 +88,6 @@ void JingleSession::CloseInternal(int result, Error error) {
   if (state_ != FAILED && state_ != CLOSED && !closing_) {
     closing_ = true;
 
-    control_channel_socket_.reset();
-    event_channel_socket_.reset();
     STLDeleteContainerPairSecondPointers(channel_connectors_.begin(),
                                          channel_connectors_.end());
 
@@ -180,16 +178,6 @@ void JingleSession::CancelChannelCreation(const std::string& name) {
   }
 }
 
-net::Socket* JingleSession::control_channel() {
-  DCHECK(CalledOnValidThread());
-  return control_channel_socket_.get();
-}
-
-net::Socket* JingleSession::event_channel() {
-  DCHECK(CalledOnValidThread());
-  return event_channel_socket_.get();
-}
-
 const std::string& JingleSession::jid() {
   DCHECK(CalledOnValidThread());
   return jid_;
@@ -388,8 +376,6 @@ void JingleSession::OnAccept() {
     }
   }
 
-  CreateChannels();
-
   SetState(CONNECTED);
 }
 
@@ -443,33 +429,6 @@ void JingleSession::OnChannelConnectorFinished(
   channel_connectors_.erase(name);
 }
 
-void JingleSession::CreateChannels() {
-  CreateStreamChannel(
-      kControlChannelName,
-      base::Bind(&JingleSession::OnChannelConnected,
-                 base::Unretained(this), &control_channel_socket_));
-  CreateStreamChannel(
-      kEventChannelName,
-      base::Bind(&JingleSession::OnChannelConnected,
-                 base::Unretained(this), &event_channel_socket_));
-}
-
-void JingleSession::OnChannelConnected(
-    scoped_ptr<net::Socket>* socket_container,
-    net::StreamSocket* socket) {
-  if (!socket) {
-    LOG(ERROR) << "Failed to connect control or events channel. "
-               << "Terminating connection";
-    CloseInternal(net::ERR_CONNECTION_CLOSED, CHANNEL_CONNECTION_ERROR);
-    return;
-  }
-
-  socket_container->reset(socket);
-
-  if (control_channel_socket_.get() && event_channel_socket_.get())
-    SetState(CONNECTED_CHANNELS);
-}
-
 const cricket::ContentInfo* JingleSession::GetContentInfo() const {
   const cricket::SessionDescription* session_description;
   // If we initiate the session, we get to specify the content name. When

remoting/protocol/jingle_session.h

@@ -37,8 +37,6 @@ class JingleSession : public protocol::Session,
       const std::string& name,
       const DatagramChannelCallback& callback) OVERRIDE;
   virtual void CancelChannelCreation(const std::string& name) OVERRIDE;
-  virtual net::Socket* control_channel() OVERRIDE;
-  virtual net::Socket* event_channel() OVERRIDE;
   virtual const std::string& jid() OVERRIDE;
   virtual const CandidateSessionConfig* candidate_config() OVERRIDE;
   virtual const SessionConfig& config() OVERRIDE;
@@ -121,16 +119,6 @@ class JingleSession : public protocol::Session,
   void OnChannelConnectorFinished(const std::string& name,
                                   JingleChannelConnector* connector);
 
-  // Creates channels after session has been accepted.
-  // TODO(sergeyu): Don't create channels in JingleSession.
-  void CreateChannels();
-
-  // Callbacks for the channels created in JingleSession.
-  // TODO(sergeyu): Remove this method once *_channel() methods are
-  // removed from Session interface.
-  void OnChannelConnected(scoped_ptr<net::Socket>* socket_container,
-                          net::StreamSocket* socket);
-
   const cricket::ContentInfo* GetContentInfo() const;
 
   void SetState(State new_state);
@@ -180,9 +168,6 @@ class JingleSession : public protocol::Session,
   // Channels that are currently being connected.
   ChannelConnectorsMap channel_connectors_;
 
-  scoped_ptr<net::Socket> control_channel_socket_;
-  scoped_ptr<net::Socket> event_channel_socket_;
-
   ScopedRunnableMethodFactory<JingleSession> task_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(JingleSession);

remoting/protocol/jingle_session_unittest.cc

@@ -219,32 +219,14 @@ class JingleSessionTest : public testing::Test {
     {
       InSequence dummy;
 
-      if (shared_secret == kTestSharedSecret) {
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::CONNECTED))
-            .Times(1);
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::CONNECTED_CHANNELS))
-            .Times(1)
-            .WillOnce(QuitThreadOnCounter(&not_connected_peers));
-        // Expect that the connection will be closed eventually.
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::CLOSED))
-            .Times(AtMost(1));
-      } else {
-        // Might pass through the CONNECTED state.
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::CONNECTED))
-            .Times(AtMost(1));
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::CONNECTED_CHANNELS))
-            .Times(AtMost(1));
-        // Expect that the connection will fail.
-        EXPECT_CALL(host_connection_callback_,
-                    OnStateChange(Session::FAILED))
-            .Times(1)
-            .WillOnce(InvokeWithoutArgs(&QuitCurrentThread));
-      }
+      EXPECT_CALL(host_connection_callback_,
+                  OnStateChange(Session::CONNECTED))
+          .Times(1)
+          .WillOnce(QuitThreadOnCounter(&not_connected_peers));
+      // Expect that the connection will be closed eventually.
+      EXPECT_CALL(host_connection_callback_,
+                  OnStateChange(Session::CLOSED))
+          .Times(AtMost(1));
     }
 
     {
@@ -253,22 +235,10 @@ class JingleSessionTest : public testing::Test {
       EXPECT_CALL(client_connection_callback_,
                   OnStateChange(Session::CONNECTING))
           .Times(1);
-      if (shared_secret == kTestSharedSecret) {
-        EXPECT_CALL(client_connection_callback_,
-                    OnStateChange(Session::CONNECTED))
-            .Times(1);
-        EXPECT_CALL(client_connection_callback_,
-                    OnStateChange(Session::CONNECTED_CHANNELS))
-            .Times(1)
-            .WillOnce(QuitThreadOnCounter(&not_connected_peers));
-      } else {
-        EXPECT_CALL(client_connection_callback_,
-                    OnStateChange(Session::CONNECTED))
-            .Times(AtMost(1));
-        EXPECT_CALL(client_connection_callback_,
-                    OnStateChange(Session::CONNECTED_CHANNELS))
-            .Times(AtMost(1));
-      }
+      EXPECT_CALL(client_connection_callback_,
+                  OnStateChange(Session::CONNECTED))
+          .Times(1)
+          .WillOnce(QuitThreadOnCounter(&not_connected_peers));
       // Expect that the connection will be closed eventually.
       EXPECT_CALL(client_connection_callback_,
                   OnStateChange(Session::CLOSED))
@@ -368,7 +338,13 @@ class TCPChannelTester : public ChannelTesterBase {
 
   virtual ~TCPChannelTester() { }
 
+  virtual bool did_initialization_fail() {
+    return !sockets_[0].get() || !sockets_[1].get();
+  }
+
   virtual void CheckResults() {
+    ASSERT_FALSE(did_initialization_fail());
+
     EXPECT_EQ(0, write_errors_);
     EXPECT_EQ(0, read_errors_);
 
@@ -394,7 +370,6 @@ class TCPChannelTester : public ChannelTesterBase {
   }
 
   void OnChannelReady(int id, net::StreamSocket* socket) {
-    ASSERT_TRUE(socket);
     if (!socket) {
       Done();
       return;
@@ -714,6 +689,14 @@ TEST_F(JingleSessionTest, Connect) {
 TEST_F(JingleSessionTest, ConnectBadChannelAuth) {
   CreateServerPair();
   ASSERT_TRUE(InitiateConnection(kTestSharedSecretBad));
+  scoped_refptr<TCPChannelTester> tester(
+      new TCPChannelTester(host_session_.get(), client_session_.get(),
+                           kMessageSize, kMessages));
+  tester->Start();
+  ASSERT_TRUE(tester->WaitFinished());
+  EXPECT_TRUE(tester->did_initialization_fail());
+
+  CloseSessions();
 }
 
 // Verify that data can be transmitted over the event channel.

remoting/protocol/pepper_session.cc

@@ -38,8 +38,6 @@ PepperSession::PepperSession(PepperSessionManager* session_manager)
 }
 
 PepperSession::~PepperSession() {
-  control_channel_socket_.reset();
-  event_channel_socket_.reset();
   STLDeleteContainerPairSecondPointers(channels_.begin(), channels_.end());
   session_manager_->SessionDestroyed(this);
 }
@@ -134,16 +132,6 @@ void PepperSession::CancelChannelCreation(const std::string& name) {
   }
 }
 
-net::Socket* PepperSession::control_channel() {
-  DCHECK(CalledOnValidThread());
-  return control_channel_socket_.get();
-}
-
-net::Socket* PepperSession::event_channel() {
-  DCHECK(CalledOnValidThread());
-  return event_channel_socket_.get();
-}
-
 const std::string& PepperSession::jid() {
   DCHECK(CalledOnValidThread());
   return peer_jid_;
@@ -199,8 +187,7 @@ const std::string& PepperSession::shared_secret() {
 void PepperSession::Close() {
   DCHECK(CalledOnValidThread());
 
-  if (state_ == CONNECTING || state_ == CONNECTED ||
-      state_ == CONNECTED_CHANNELS) {
+  if (state_ == CONNECTING || state_ == CONNECTED) {
     // Send session-terminate message.
     JingleMessage message(peer_jid_, JingleMessage::SESSION_TERMINATE,
                           session_id_);
@@ -252,7 +239,6 @@ void PepperSession::OnAccept(const JingleMessage& message,
     return;
   }
 
-  CreateChannels();
   SetState(CONNECTED);
 
    // In case there is transport information in the accept message.
@@ -292,16 +278,16 @@ void PepperSession::OnTerminate(const JingleMessage& message,
     return;
   }
 
-  // TODO(sergeyu): We should return CHANNEL_CONNECTION_ERROR only in
-  // case when |message.reason| is set GENERAL_ERROR, but some legacy
-  // hosts may sent terminate messages with reason set to SUCCESS.
   if (state_ == CONNECTED) {
-    // Session was connected, but we failed to connect channels.
-    OnError(CHANNEL_CONNECTION_ERROR);
+    if (message.reason == JingleMessage::GENERAL_ERROR) {
+      OnError(CHANNEL_CONNECTION_ERROR);
+    } else {
+      CloseInternal(false);
+    }
     return;
   }
 
-  CloseInternal(false);
+  LOG(WARNING) << "Received unexpected session-terminate message.";
 }
 
 bool PepperSession::InitializeConfigFromDescription(
@@ -369,40 +355,11 @@ void PepperSession::SendTransportInfo() {
           base::Unretained(this))));
 }
 
-void PepperSession::CreateChannels() {
-   CreateStreamChannel(
-      kControlChannelName,
-      base::Bind(&PepperSession::OnChannelConnected,
-                 base::Unretained(this), &control_channel_socket_));
-  CreateStreamChannel(
-      kEventChannelName,
-      base::Bind(&PepperSession::OnChannelConnected,
-                 base::Unretained(this), &event_channel_socket_));
-}
-
-void PepperSession::OnChannelConnected(
-    scoped_ptr<net::Socket>* socket_container,
-    net::StreamSocket* socket) {
-  if (!socket) {
-    LOG(ERROR) << "Failed to connect control or events channel. "
-               << "Terminating connection";
-    OnError(CHANNEL_CONNECTION_ERROR);
-    return;
-  }
-
-  socket_container->reset(socket);
-
-  if (control_channel_socket_.get() && event_channel_socket_.get())
-    SetState(CONNECTED_CHANNELS);
-}
 
 void PepperSession::CloseInternal(bool failed) {
   DCHECK(CalledOnValidThread());
 
   if (state_ != FAILED && state_ != CLOSED) {
-    control_channel_socket_.reset();
-    event_channel_socket_.reset();
-
     if (failed)
       SetState(FAILED);
     else

remoting/protocol/pepper_session.h

@@ -49,8 +49,6 @@ class PepperSession : public Session {
       const std::string& name,
       const DatagramChannelCallback& callback) OVERRIDE;
   virtual void CancelChannelCreation(const std::string& name) OVERRIDE;
-  virtual net::Socket* control_channel() OVERRIDE;
-  virtual net::Socket* event_channel() OVERRIDE;
   virtual const std::string& jid() OVERRIDE;
   virtual const CandidateSessionConfig* candidate_config() OVERRIDE;
   virtual const SessionConfig& config() OVERRIDE;
@@ -104,12 +102,6 @@ class PepperSession : public Session {
   void SendTransportInfo();
   void OnTransportInfoResponse(const buzz::XmlElement* response);
 
-  // Helper methods to create event and control channels.
-  // TODO(sergeyu): Remove these methods.
-  void CreateChannels();
-  void OnChannelConnected(scoped_ptr<net::Socket>* socket_container,
-                          net::StreamSocket* socket);
-
   // Close all the channels and terminate the session.
   void CloseInternal(bool failed);
 
@@ -138,9 +130,6 @@ class PepperSession : public Session {
 
   ChannelsMap channels_;
 
-  scoped_ptr<net::Socket> control_channel_socket_;
-  scoped_ptr<net::Socket> event_channel_socket_;
-
   base::OneShotTimer<PepperSession> transport_infos_timer_;
   std::list<cricket::Candidate> pending_candidates_;
 

remoting/protocol/protobuf_video_reader.cc

@@ -37,6 +37,10 @@ void ProtobufVideoReader::Init(protocol::Session* session,
       base::Bind(&ProtobufVideoReader::OnChannelReady, base::Unretained(this)));
 }
 
+bool ProtobufVideoReader::is_connected() {
+  return channel_.get() != NULL;
+}
+
 void ProtobufVideoReader::OnChannelReady(net::StreamSocket* socket) {
   if (!socket) {
     initialized_callback_.Run(false);

remoting/protocol/protobuf_video_reader.h

@@ -28,6 +28,7 @@ class ProtobufVideoReader : public VideoReader {
   virtual void Init(protocol::Session* session,
                     VideoStub* video_stub,
                     const InitializedCallback& callback) OVERRIDE;
+  virtual bool is_connected() OVERRIDE;
 
  private:
   void OnChannelReady(net::StreamSocket* socket);

remoting/protocol/protobuf_video_writer.cc

@@ -57,6 +57,10 @@ void ProtobufVideoWriter::Close() {
   }
 }
 
+bool ProtobufVideoWriter::is_connected() {
+  return channel_.get() != NULL;
+}
+
 void ProtobufVideoWriter::ProcessVideoPacket(const VideoPacket* packet,
                                              const base::Closure& done) {
   buffered_writer_->Write(SerializeAndFrameMessage(*packet), done);

remoting/protocol/protobuf_video_writer.h

@@ -35,6 +35,7 @@ class ProtobufVideoWriter : public VideoWriter {
   virtual void Init(protocol::Session* session,
                     const InitializedCallback& callback) OVERRIDE;
   virtual void Close() OVERRIDE;
+  virtual bool is_connected() OVERRIDE;
 
   // VideoStub interface.
   virtual void ProcessVideoPacket(const VideoPacket* packet,

remoting/protocol/rtp_video_reader.cc

@@ -56,6 +56,10 @@ void RtpVideoReader::Init(protocol::Session* session,
                  base::Unretained(this), false));
 }
 
+bool RtpVideoReader::is_connected() {
+  return rtp_channel_.get() && rtcp_channel_.get();
+}
+
 void RtpVideoReader::OnChannelReady(bool rtp, net::Socket* socket) {
   if (!socket) {
     if (!initialized_) {

remoting/protocol/rtp_video_reader.h

@@ -32,6 +32,7 @@ class RtpVideoReader : public VideoReader {
   virtual void Init(protocol::Session* session,
                     VideoStub* video_stub,
                     const InitializedCallback& callback) OVERRIDE;
+  virtual bool is_connected() OVERRIDE;
 
  private:
   friend class RtpVideoReaderTest;

remoting/protocol/rtp_video_writer.cc

@@ -81,6 +81,10 @@ void RtpVideoWriter::Close() {
   }
 }
 
+bool RtpVideoWriter::is_connected() {
+  return rtp_channel_.get() && rtcp_channel_.get();
+}
+
 void RtpVideoWriter::ProcessVideoPacket(const VideoPacket* packet,
                                         const base::Closure& done) {
   CHECK(packet->format().encoding() == VideoPacketFormat::ENCODING_VP8)

remoting/protocol/rtp_video_writer.h

@@ -27,6 +27,7 @@ class RtpVideoWriter : public VideoWriter {
   virtual void Init(Session* session,
                     const InitializedCallback& callback) OVERRIDE;
   virtual void Close() OVERRIDE;
+  virtual bool is_connected() OVERRIDE;
 
   // VideoStub interface.
   virtual void ProcessVideoPacket(const VideoPacket* packet,

remoting/protocol/session.h

@@ -41,10 +41,6 @@ class Session : public base::NonThreadSafe {
     // Session has been accepted, but channels are connected yet.
     CONNECTED,
 
-    // Video and control channels are connected.
-    // TODO(sergeyu): Remove this state.
-    CONNECTED_CHANNELS,
-
     // Session has been closed.
     CLOSED,
 
@@ -95,11 +91,6 @@ class Session : public base::NonThreadSafe {
   // completed then cancelling it has no effect.
   virtual void CancelChannelCreation(const std::string& name) = 0;
 
-  // TODO(sergeyu): Remove these methods, and use CreateChannel()
-  // instead.
-  virtual net::Socket* control_channel() = 0;
-  virtual net::Socket* event_channel() = 0;
-
   // JID of the other side.
   virtual const std::string& jid() = 0;
 

remoting/protocol/video_reader.h

@@ -38,6 +38,7 @@ class VideoReader {
   virtual void Init(Session* session,
                     VideoStub* video_stub,
                     const InitializedCallback& callback) = 0;
+  virtual bool is_connected() = 0;
 
  protected:
   VideoReader() { }

remoting/protocol/video_writer.h

@@ -42,6 +42,9 @@ class VideoWriter : public VideoStub {
   // object is destroyed.
   virtual void Close() = 0;
 
+  // Returns true if the channel is connected.
+  virtual bool is_connected() = 0;
+
  protected:
   VideoWriter() { }
 

remoting/remoting.gyp

@@ -742,6 +742,8 @@
         'protocol/buffered_socket_writer.h',
         'protocol/channel_authenticator.cc',
         'protocol/channel_authenticator.h',
+        'protocol/channel_dispatcher_base.cc',
+        'protocol/channel_dispatcher_base.h',
         'protocol/client_control_dispatcher.cc',
         'protocol/client_control_dispatcher.h',
         'protocol/client_event_dispatcher.cc',