Chromoting protocol layers to receive and send login messages

BUG=None
TEST=None

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

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

20 files changed, 297 insertions, 113 deletions

diff --git a/remoting/client/chromoting_client.cc b/remoting/client/chromoting_client.cc
index 4e4b9a3..299a2cf 100644
--- a/remoting/client/chromoting_client.cc
+++ b/remoting/client/chromoting_client.cc
@@ -217,9 +217,13 @@ void ChromotingClient::Initialize() {
 ////////////////////////////////////////////////////////////////////////////
 // ClientStub control channel interface.
 void ChromotingClient::NotifyResolution(
-    const protocol::NotifyResolutionRequest* msg,
-    Task* done) {
-  // TODO(garykac): Implement this.
+    const protocol::NotifyResolutionRequest* msg, Task* done) {
+  NOTIMPLEMENTED();
+}
+
+void ChromotingClient::BeginSessionResponse(
+    const protocol::LocalLoginStatus* msg, Task* done) {
+  NOTIMPLEMENTED();
 }
 
 }  // namespace remoting
diff --git a/remoting/client/chromoting_client.h b/remoting/client/chromoting_client.h
index 441c3b3..b90dc41 100644
--- a/remoting/client/chromoting_client.h
+++ b/remoting/client/chromoting_client.h
@@ -22,6 +22,7 @@ class MessageLoop;
 namespace remoting {
 
 namespace protocol {
+class LocalLoginStatus;
 class NotifyResolutionRequest;
 }  // namespace protocol
 
@@ -67,6 +68,8 @@ class ChromotingClient : public protocol::ConnectionToHost::HostEventCallback,
   // ClientStub implementation.
   virtual void NotifyResolution(const protocol::NotifyResolutionRequest* msg,
                                 Task* done);
+  virtual void BeginSessionResponse(const protocol::LocalLoginStatus* msg,
+                                    Task* done);
 
   // VideoStub implementation.
   virtual void ProcessVideoPacket(const VideoPacket* packet, Task* done);
diff --git a/remoting/host/host_stub_fake.cc b/remoting/host/host_stub_fake.cc
index 179dd34..2e3fc76 100644
--- a/remoting/host/host_stub_fake.cc
+++ b/remoting/host/host_stub_fake.cc
@@ -13,4 +13,10 @@ void HostStubFake::SuggestResolution(
   delete done;
 }
 
+void HostStubFake::BeginSessionRequest(
+    const protocol::LocalLoginCredentials* credentials, Task* done) {
+  done->Run();
+  delete done;
+}
+
 }  // namespace remoting
diff --git a/remoting/host/host_stub_fake.h b/remoting/host/host_stub_fake.h
index 5757b36..218e15d 100644
--- a/remoting/host/host_stub_fake.h
+++ b/remoting/host/host_stub_fake.h
@@ -19,6 +19,8 @@ class HostStubFake : public protocol::HostStub {
 
   virtual void SuggestResolution(
       const protocol::SuggestResolutionRequest* msg, Task* done);
+  virtual void BeginSessionRequest(
+      const protocol::LocalLoginCredentials* credentials, Task* done);
 
  private:
   DISALLOW_COPY_AND_ASSIGN(HostStubFake);
diff --git a/remoting/protocol/client_control_sender.cc b/remoting/protocol/client_control_sender.cc
index c16d235..e5cda49 100644
--- a/remoting/protocol/client_control_sender.cc
+++ b/remoting/protocol/client_control_sender.cc
@@ -31,5 +31,13 @@ void ClientControlSender::NotifyResolution(
   buffered_writer_->Write(SerializeAndFrameMessage(message), done);
 }
 
+void ClientControlSender::BeginSessionResponse(const LocalLoginStatus* msg,
+                                               Task* done) {
+  protocol::ControlMessage message;
+  message.mutable_begin_session_response()->mutable_login_status()->CopyFrom(
+      *msg);
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
+}
+
 }  // namespace protocol
 }  // namespace remoting
diff --git a/remoting/protocol/client_control_sender.h b/remoting/protocol/client_control_sender.h
index 6d9b06f..cd039b8 100644
--- a/remoting/protocol/client_control_sender.h
+++ b/remoting/protocol/client_control_sender.h
@@ -35,6 +35,9 @@ class ClientControlSender : public ClientStub {
 
   virtual void NotifyResolution(const NotifyResolutionRequest* msg,
                                 Task* done);
+  virtual void BeginSessionResponse(const LocalLoginStatus* msg,
+                                    Task* done);
+
  private:
   // Buffered socket writer holds the serialized message and send it on the
   // right thread.
diff --git a/remoting/protocol/client_message_dispatcher.cc b/remoting/protocol/client_message_dispatcher.cc
new file mode 100644
index 0000000..c70588a
--- /dev/null
+++ b/remoting/protocol/client_message_dispatcher.cc
@@ -0,0 +1,58 @@
+// 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 "base/ref_counted.h"
+#include "net/base/io_buffer.h"
+#include "remoting/proto/control.pb.h"
+#include "remoting/proto/event.pb.h"
+#include "remoting/proto/internal.pb.h"
+#include "remoting/protocol/client_message_dispatcher.h"
+#include "remoting/protocol/client_stub.h"
+#include "remoting/protocol/input_stub.h"
+#include "remoting/protocol/message_reader.h"
+#include "remoting/protocol/ref_counted_message.h"
+#include "remoting/protocol/session.h"
+
+namespace remoting {
+namespace protocol {
+
+ClientMessageDispatcher::ClientMessageDispatcher() : client_stub_(NULL) {
+}
+
+ClientMessageDispatcher::~ClientMessageDispatcher() {
+}
+
+void ClientMessageDispatcher::Initialize(
+    protocol::Session* session, ClientStub* client_stub) {
+  if (!session || !client_stub || !session->control_channel()) {
+    return;
+  }
+
+  control_message_reader_.reset(new MessageReader());
+  client_stub_ = client_stub;
+
+  control_message_reader_->Init<ControlMessage>(
+      session->control_channel(),
+      NewCallback(this, &ClientMessageDispatcher::OnControlMessageReceived));
+  return;
+}
+
+void ClientMessageDispatcher::OnControlMessageReceived(
+    ControlMessage* message) {
+  scoped_refptr<RefCountedMessage<ControlMessage> > ref_msg =
+      new RefCountedMessage<ControlMessage>(message);
+  if (message->has_notify_resolution()) {
+    client_stub_->NotifyResolution(
+        &message->notify_resolution(), NewDeleteTask(ref_msg));
+  } else if (message->has_begin_session_response()) {
+    client_stub_->BeginSessionResponse(
+        &message->begin_session_response().login_status(),
+        NewDeleteTask(ref_msg));
+  } else {
+    NOTREACHED() << "Invalid control message received";
+  }
+}
+
+}  // namespace protocol
+}  // namespace remoting
diff --git a/remoting/protocol/client_message_dispatcher.h b/remoting/protocol/client_message_dispatcher.h
new file mode 100644
index 0000000..cad48e4
--- /dev/null
+++ b/remoting/protocol/client_message_dispatcher.h
@@ -0,0 +1,61 @@
+// 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_PROTOCOL_CLIENT_MESSAGE_DISPATCHER_H_
+#define REMOTING_PROTOCOL_CLIENT_MESSAGE_DISPATCHER_H_
+
+#include "base/basictypes.h"
+#include "base/scoped_ptr.h"
+#include "base/task.h"
+
+namespace remoting {
+
+class EventMessage;
+
+namespace protocol {
+
+class ClientStub;
+class ControlMessage;
+class InputStub;
+class MessageReader;
+class Session;
+
+// A message dispatcher used to listen for messages received in
+// protocol::Session. It dispatches messages to the corresponding
+// handler.
+//
+// Internally it contains an EventStreamReader that decodes data on
+// communications channels into protocol buffer messages.
+// EventStreamReader is registered with protocol::Session given to it.
+//
+// Object of this class is owned by ConnectionToHost.
+class ClientMessageDispatcher {
+ public:
+  // Construct a message dispatcher.
+  ClientMessageDispatcher();
+  virtual ~ClientMessageDispatcher();
+
+  // Initialize the message dispatcher with the given connection and
+  // message handlers.
+  void Initialize(protocol::Session* session, ClientStub* client_stub);
+
+ private:
+  void OnControlMessageReceived(ControlMessage* message);
+
+  // MessageReader that runs on the control channel. It runs a loop
+  // that parses data on the channel and then calls the corresponding handler
+  // in this class.
+  scoped_ptr<MessageReader> control_message_reader_;
+
+  // Stubs for client and input. These objects are not owned.
+  // They are called on the thread there data is received, i.e. jingle thread.
+  ClientStub* client_stub_;
+
+  DISALLOW_COPY_AND_ASSIGN(ClientMessageDispatcher);
+};
+
+}  // namespace protocol
+}  // namespace remoting
+
+#endif  // REMOTING_PROTOCOL_CLIENT_MESSAGE_DISPATCHER_H_
diff --git a/remoting/protocol/client_stub.h b/remoting/protocol/client_stub.h
index 2662e95..4b73789 100644
--- a/remoting/protocol/client_stub.h
+++ b/remoting/protocol/client_stub.h
@@ -17,6 +17,7 @@ class Task;
 namespace remoting {
 namespace protocol {
 
+class LocalLoginStatus;
 class NotifyResolutionRequest;
 
 class ClientStub {
@@ -26,6 +27,8 @@ class ClientStub {
 
   virtual void NotifyResolution(const NotifyResolutionRequest* msg,
                                 Task* done) = 0;
+  virtual void BeginSessionResponse(const LocalLoginStatus* msg,
+                                    Task* done) = 0;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(ClientStub);
diff --git a/remoting/protocol/client_stub_impl.cc b/remoting/protocol/client_stub_impl.cc
deleted file mode 100644
index 2c8fedb..0000000
--- a/remoting/protocol/client_stub_impl.cc
+++ /dev/null
@@ -1,38 +0,0 @@
-// 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.
-
-// This stub is thread safe because of the use of BufferedSocketWriter.
-// BufferedSocketWriter buffers messages and send them on them right thread.
-
-#include "remoting/protocol/client_stub_impl.h"
-
-#include "base/task.h"
-#include "remoting/protocol/buffered_socket_writer.h"
-#include "remoting/proto/control.pb.h"
-#include "remoting/protocol/util.h"
-
-namespace remoting {
-namespace protocol {
-
-ClientStubImpl::ClientStubImpl(net::Socket* socket)
-    : buffered_writer_(new BufferedSocketWriter()) {
-  buffered_writer_->Init(socket, NULL);
-}
-
-ClientStubImpl::~ClientStubImpl() {
-}
-
-void ClientStubImpl::NotifyResolution(
-    const NotifyResolutionRequest& msg, Task* done) {
-  ControlMessage message;
-  message.mutable_notify_resolution()->CopyFrom(msg);
-  buffered_writer_->Write(SerializeAndFrameMessage(message));
-  if (done) {
-    done->Run();
-    delete done;
-  }
-}
-
-}  // namespace protocol
-}  // namespace remoting
diff --git a/remoting/protocol/host_control_sender.cc b/remoting/protocol/host_control_sender.cc
new file mode 100644
index 0000000..3f27e80
--- /dev/null
+++ b/remoting/protocol/host_control_sender.cc
@@ -0,0 +1,43 @@
+// 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.
+
+// This stub is thread safe because of the use of BufferedSocketWriter.
+// BufferedSocketWriter buffers messages and send them on them right thread.
+
+#include "remoting/protocol/host_control_sender.h"
+
+#include "base/task.h"
+#include "remoting/protocol/buffered_socket_writer.h"
+#include "remoting/proto/control.pb.h"
+#include "remoting/proto/internal.pb.h"
+#include "remoting/protocol/util.h"
+
+namespace remoting {
+namespace protocol {
+
+HostControlSender::HostControlSender(net::Socket* socket)
+    : buffered_writer_(new BufferedSocketWriter()) {
+  buffered_writer_->Init(socket, NULL);
+}
+
+HostControlSender::~HostControlSender() {
+}
+
+void HostControlSender::SuggestResolution(
+    const SuggestResolutionRequest* msg, Task* done) {
+  protocol::ControlMessage message;
+  message.mutable_suggest_resolution()->CopyFrom(*msg);
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
+}
+
+void HostControlSender::BeginSessionRequest(const LocalLoginCredentials* msg,
+                                            Task* done) {
+  protocol::ControlMessage message;
+  message.mutable_begin_session_request()->mutable_credentials()->CopyFrom(
+      *msg);
+  buffered_writer_->Write(SerializeAndFrameMessage(message), done);
+}
+
+}  // namespace protocol
+}  // namespace remoting
diff --git a/remoting/protocol/client_stub_impl.h b/remoting/protocol/host_control_sender.h
index d953be7..148b519 100644
--- a/remoting/protocol/client_stub_impl.h
+++ b/remoting/protocol/host_control_sender.h
@@ -2,19 +2,19 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-// Implementation of ClientStub using sockets created from jingle connection.
+// Implementation of HostStub using sockets created from jingle connection.
 // It sends messages through the socket after serializing it.
 //
-// Object of this class can only be created by ConnectionToClient.
+// Object of this class can only be created by ConnectionToHost.
 //
 // This class can be used on any thread.
 
-#ifndef REMOTING_PROTOCOL_CLIENT_STUB_IMPL_H_
-#define REMOTING_PROTOCOL_CLIENT_STUB_IMPL_H_
+#ifndef REMOTING_PROTOCOL_HOST_STUB_IMPL_H_
+#define REMOTING_PROTOCOL_HOST_STUB_IMPL_H_
 
 #include "base/basictypes.h"
 #include "base/ref_counted.h"
-#include "remoting/protocol/client_stub.h"
+#include "remoting/protocol/host_stub.h"
 
 class Task;
 
@@ -27,23 +27,26 @@ namespace protocol {
 
 class BufferedSocketWriter;
 
-class ClientStubImpl : public ClientStub {
+class HostControlSender : public HostStub {
  public:
   // Create a stub using a socket.
-  explicit ClientStubImpl(net::Socket* socket);
-  virtual ~ClientStubImpl();
+  explicit HostControlSender(net::Socket* socket);
+  virtual ~HostControlSender();
+
+  virtual void SuggestResolution(
+      const SuggestResolutionRequest* msg, Task* done);
+  virtual void BeginSessionRequest(
+      const LocalLoginCredentials* credentials, Task* done);
 
-  virtual void NotifyResolution(const NotifyResolutionRequest& msg,
-                                Task* done) = 0;
  private:
   // Buffered socket writer holds the serialized message and send it on the
   // right thread.
   scoped_refptr<BufferedSocketWriter> buffered_writer_;
 
-  DISALLOW_COPY_AND_ASSIGN(ClientStubImpl);
+  DISALLOW_COPY_AND_ASSIGN(HostControlSender);
 };
 
 }  // namespace protocol
 }  // namespace remoting
 
-#endif  // REMOTING_PROTOCOL_CLIENT_STUB_IMPL_H_
+#endif  // REMOTING_PROTOCOL_HOST_STUB_IMPL_H_
diff --git a/remoting/protocol/host_message_dispatcher.cc b/remoting/protocol/host_message_dispatcher.cc
index eec734c..b557171 100644
--- a/remoting/protocol/host_message_dispatcher.cc
+++ b/remoting/protocol/host_message_dispatcher.cc
@@ -11,35 +11,9 @@
 #include "remoting/protocol/host_stub.h"
 #include "remoting/protocol/input_stub.h"
 #include "remoting/protocol/message_reader.h"
+#include "remoting/protocol/ref_counted_message.h"
 #include "remoting/protocol/session.h"
 
-namespace {
-
-// A single protobuf can contain multiple messages that will be handled by
-// different message handlers.  We use this wrapper to ensure that the
-// protobuf is only deleted after all the handlers have finished executing.
-template <typename T>
-class RefCountedMessage : public base::RefCounted<RefCountedMessage<T> > {
- public:
-  RefCountedMessage(T* message) : message_(message) { }
-
-  T* message() { return message_.get(); }
-
- private:
-  scoped_ptr<T> message_;
-};
-
-// Dummy methods to destroy messages.
-template <class T>
-static void DeleteMessage(scoped_refptr<T> message) { }
-
-template <class T>
-static Task* NewDeleteTask(scoped_refptr<T> message) {
-  return NewRunnableFunction(&DeleteMessage<T>, message);
-}
-
-}  // namespace
-
 namespace remoting {
 namespace protocol {
 
@@ -51,12 +25,12 @@ HostMessageDispatcher::HostMessageDispatcher() :
 HostMessageDispatcher::~HostMessageDispatcher() {
 }
 
-bool HostMessageDispatcher::Initialize(
+void HostMessageDispatcher::Initialize(
     protocol::Session* session,
     HostStub* host_stub, InputStub* input_stub) {
   if (!session || !host_stub || !input_stub ||
       !session->event_channel() || !session->control_channel()) {
-    return false;
+    return;
   }
 
   control_message_reader_.reset(new MessageReader());
@@ -71,7 +45,6 @@ bool HostMessageDispatcher::Initialize(
   control_message_reader_->Init<ControlMessage>(
       session->control_channel(),
       NewCallback(this, &HostMessageDispatcher::OnControlMessageReceived));
-  return true;
 }
 
 void HostMessageDispatcher::OnControlMessageReceived(ControlMessage* message) {
@@ -80,6 +53,12 @@ void HostMessageDispatcher::OnControlMessageReceived(ControlMessage* message) {
   if (message->has_suggest_resolution()) {
     host_stub_->SuggestResolution(
         &message->suggest_resolution(), NewDeleteTask(ref_msg));
+  } else if (message->has_begin_session_request()) {
+    host_stub_->BeginSessionRequest(
+        &message->begin_session_request().credentials(),
+        NewDeleteTask(ref_msg));
+  } else {
+    NOTREACHED() << "Invalid control message received";
   }
 }
 
diff --git a/remoting/protocol/host_message_dispatcher.h b/remoting/protocol/host_message_dispatcher.h
index 0a9aa5e..6274168 100644
--- a/remoting/protocol/host_message_dispatcher.h
+++ b/remoting/protocol/host_message_dispatcher.h
@@ -29,7 +29,7 @@ class Session;
 // communications channels into protocol buffer messages.
 // EventStreamReader is registered with protocol::Session given to it.
 //
-// Object of this class is owned by ChromotingHost to dispatch messages
+// Object of this class is owned by ConnectionToClient to dispatch messages
 // to itself.
 class HostMessageDispatcher {
  public:
@@ -39,8 +39,7 @@ class HostMessageDispatcher {
 
   // Initialize the message dispatcher with the given connection and
   // message handlers.
-  // Return true if initalization was successful.
-  bool Initialize(protocol::Session* session,
+  void Initialize(protocol::Session* session,
                   HostStub* host_stub, InputStub* input_stub);
 
  private:
diff --git a/remoting/protocol/host_stub.h b/remoting/protocol/host_stub.h
index cf14daa..1a26225 100644
--- a/remoting/protocol/host_stub.h
+++ b/remoting/protocol/host_stub.h
@@ -17,6 +17,7 @@ namespace remoting {
 namespace protocol {
 
 class SuggestResolutionRequest;
+class LocalLoginCredentials;
 
 class HostStub {
  public:
@@ -25,6 +26,8 @@ class HostStub {
 
   virtual void SuggestResolution(
       const SuggestResolutionRequest* msg, Task* done) = 0;
+  virtual void BeginSessionRequest(
+      const LocalLoginCredentials* credentials, Task* done) = 0;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(HostStub);
diff --git a/remoting/protocol/jingle_connection_to_host.cc b/remoting/protocol/jingle_connection_to_host.cc
index a551f17..f786ef8 100644
--- a/remoting/protocol/jingle_connection_to_host.cc
+++ b/remoting/protocol/jingle_connection_to_host.cc
@@ -9,6 +9,7 @@
 #include "remoting/base/constants.h"
 #include "remoting/jingle_glue/jingle_thread.h"
 #include "remoting/proto/auth.pb.h"
+#include "remoting/protocol/client_message_dispatcher.h"
 #include "remoting/protocol/client_stub.h"
 #include "remoting/protocol/input_sender.h"
 #include "remoting/protocol/jingle_session_manager.h"
@@ -21,7 +22,8 @@ namespace protocol {
 
 JingleConnectionToHost::JingleConnectionToHost(JingleThread* thread)
     : thread_(thread),
-      event_callback_(NULL) {
+      event_callback_(NULL),
+      dispatcher_(new ClientMessageDispatcher()) {
 }
 
 JingleConnectionToHost::~JingleConnectionToHost() {
@@ -34,6 +36,7 @@ void JingleConnectionToHost::Connect(const std::string& username,
                                      ClientStub* client_stub,
                                      VideoStub* video_stub) {
   event_callback_ = event_callback;
+  client_stub_ = client_stub;
   video_stub_ = video_stub;
 
   // Initialize |jingle_client_|.
@@ -65,18 +68,12 @@ void JingleConnectionToHost::Disconnect() {
   }
 }
 
-void JingleConnectionToHost::OnControlMessage(ControlMessage* msg) {
-  // TODO(sergeyu): Remove this method and pass ClientStub to the control
-  // stream dispatcher.
-  delete msg;
-}
-
 void JingleConnectionToHost::InitSession() {
   DCHECK_EQ(message_loop(), MessageLoop::current());
 
   // Initialize chromotocol |session_manager_|.
-  protocol::JingleSessionManager* session_manager =
-      new protocol::JingleSessionManager(thread_);
+  JingleSessionManager* session_manager =
+      new JingleSessionManager(thread_);
   // TODO(ajwong): Make this a command switch when we're more stable.
   session_manager->set_allow_local_ips(true);
   session_manager->Init(
@@ -144,35 +141,33 @@ void JingleConnectionToHost::OnStateChange(JingleClient* client,
   }
 }
 
-void JingleConnectionToHost::OnNewSession(protocol::Session* session,
-    protocol::SessionManager::IncomingSessionResponse* response) {
+void JingleConnectionToHost::OnNewSession(Session* session,
+    SessionManager::IncomingSessionResponse* response) {
   DCHECK_EQ(message_loop(), MessageLoop::current());
   // Client always rejects incoming sessions.
-  *response = protocol::SessionManager::DECLINE;
+  *response = SessionManager::DECLINE;
 }
 
 void JingleConnectionToHost::OnSessionStateChange(
-    protocol::Session::State state) {
+    Session::State state) {
   DCHECK_EQ(message_loop(), MessageLoop::current());
   DCHECK(event_callback_);
 
   switch (state) {
-    case protocol::Session::FAILED:
+    case Session::FAILED:
       event_callback_->OnConnectionFailed(this);
       break;
 
-    case protocol::Session::CLOSED:
+    case Session::CLOSED:
       event_callback_->OnConnectionClosed(this);
       break;
 
-    case protocol::Session::CONNECTED:
+    case Session::CONNECTED:
       // Initialize reader and writer.
-      control_reader_.Init<ControlMessage>(
-          session_->control_channel(),
-          NewCallback(this, &JingleConnectionToHost::OnControlMessage));
       video_reader_.reset(VideoReader::Create(session_->config()));
       video_reader_->Init(session_, video_stub_);
       input_stub_.reset(new InputSender(session_->event_channel()));
+      dispatcher_->Initialize(session_.get(), client_stub_);
       event_callback_->OnConnectionOpened(this);
       break;
 
diff --git a/remoting/protocol/jingle_connection_to_host.h b/remoting/protocol/jingle_connection_to_host.h
index b8b64b4..8e90c5b 100644
--- a/remoting/protocol/jingle_connection_to_host.h
+++ b/remoting/protocol/jingle_connection_to_host.h
@@ -37,6 +37,7 @@ class VideoPacket;
 
 namespace protocol {
 
+class ClientMessageDispatcher;
 class VideoReader;
 class VideoStub;
 
@@ -62,11 +63,11 @@ class JingleConnectionToHost : public ConnectionToHost,
 
   // Callback for chromotocol SessionManager.
   void OnNewSession(
-      protocol::Session* connection,
-      protocol::SessionManager::IncomingSessionResponse* response);
+      Session* connection,
+      SessionManager::IncomingSessionResponse* response);
 
   // Callback for chromotocol Session.
-  void OnSessionStateChange(protocol::Session::State state);
+  void OnSessionStateChange(Session::State state);
 
  private:
   // The message loop for the jingle thread this object works on.
@@ -76,9 +77,6 @@ class JingleConnectionToHost : public ConnectionToHost,
   // P2P connection to the host.
   void InitSession();
 
-  // Callback for |control_reader_|.
-  void OnControlMessage(ControlMessage* msg);
-
   // Callback for |video_reader_|.
   void OnVideoPacket(VideoPacket* packet);
 
@@ -90,8 +88,8 @@ class JingleConnectionToHost : public ConnectionToHost,
   JingleThread* thread_;
 
   scoped_refptr<JingleClient> jingle_client_;
-  scoped_refptr<protocol::SessionManager> session_manager_;
-  scoped_refptr<protocol::Session> session_;
+  scoped_refptr<SessionManager> session_manager_;
+  scoped_refptr<Session> session_;
 
   MessageReader control_reader_;
   scoped_ptr<VideoReader> video_reader_;
@@ -101,6 +99,8 @@ class JingleConnectionToHost : public ConnectionToHost,
 
   std::string host_jid_;
 
+  scoped_ptr<ClientMessageDispatcher> dispatcher_;
+
   DISALLOW_COPY_AND_ASSIGN(JingleConnectionToHost);
 };
 
diff --git a/remoting/protocol/mock_objects.h b/remoting/protocol/mock_objects.h
index 0c2cc76..045a414 100644
--- a/remoting/protocol/mock_objects.h
+++ b/remoting/protocol/mock_objects.h
@@ -59,6 +59,8 @@ class MockHostStub : public HostStub {
 
   MOCK_METHOD2(SuggestResolution, void(const SuggestResolutionRequest* msg,
                                        Task* done));
+  MOCK_METHOD2(BeginSessionRequest,
+               void(const LocalLoginCredentials* credentials, Task* done));
 
  private:
   DISALLOW_COPY_AND_ASSIGN(MockHostStub);
diff --git a/remoting/protocol/ref_counted_message.h b/remoting/protocol/ref_counted_message.h
new file mode 100644
index 0000000..24a8b52
--- /dev/null
+++ b/remoting/protocol/ref_counted_message.h
@@ -0,0 +1,45 @@
+// 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.
+
+// This is a wrapper class to help ref-counting a protobuf message.
+// This file should only be inclued on host_message_dispatcher.cc and
+// client_message_dispatche.cc.
+
+// A single protobuf can contain multiple messages that will be handled by
+// different message handlers.  We use this wrapper to ensure that the
+// protobuf is only deleted after all the handlers have finished executing.
+
+#ifndef REMOTING_PROTOCOL_REF_COUNTED_MESSAGE_H_
+#define REMOTING_PROTOCOL_REF_COUNTED_MESSAGE_H_
+
+#include "base/ref_counted.h"
+#include "base/task.h"
+
+namespace remoting {
+namespace protocol {
+
+template <typename T>
+class RefCountedMessage : public base::RefCounted<RefCountedMessage<T> > {
+ public:
+  RefCountedMessage(T* message) : message_(message) { }
+
+  T* message() { return message_.get(); }
+
+ private:
+  scoped_ptr<T> message_;
+};
+
+// Dummy methods to destroy messages.
+template <class T>
+static void DeleteMessage(scoped_refptr<T> message) { }
+
+template <class T>
+static Task* NewDeleteTask(scoped_refptr<T> message) {
+  return NewRunnableFunction(&DeleteMessage<T>, message);
+}
+
+}  // namespace protocol
+}  // namespace remoting
+
+#endif  // REMOTING_PROTOCOL_REF_COUNTED_MESSAGE_H_
diff --git a/remoting/remoting.gyp b/remoting/remoting.gyp
index d5e4b0f..8b6b5f1 100644
--- a/remoting/remoting.gyp
+++ b/remoting/remoting.gyp
@@ -362,15 +362,20 @@
         'protocol/buffered_socket_writer.h',
         'protocol/client_control_sender.cc',
         'protocol/client_control_Sender.h',
+        'protocol/client_message_dispatcher.cc',
+        'protocol/client_message_dispatcher.h',
         'protocol/client_stub.h',
         'protocol/connection_to_client.cc',
         'protocol/connection_to_client.h',
         'protocol/connection_to_host.cc',
         'protocol/connection_to_host.h',
         'protocol/host_control_message_handler.h',
+        'protocol/host_control_sender.cc',
+        'protocol/host_control_sender.h',
         'protocol/host_event_message_handler.h',
         'protocol/host_message_dispatcher.cc',
         'protocol/host_message_dispatcher.h',
+        'protocol/host_stub.h',
         'protocol/input_sender.cc',
         'protocol/input_sender.h',
         'protocol/jingle_connection_to_host.cc',