Use new Chromotocol code in host andclient.

1. ProtocolDecoder renamed to MessagesDecoder and moved to remoting/protocol.
2. base/protocol_util.[h|cc] split into base/util.[h|cc] and protocol/util.[h|cc].
3. Added StreamReader and StreamWriter classes for events and video channels.
4. Client and host changed to use the new protocol code.

BUG=None
TEST=Unittests

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

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

14 files changed, 207 insertions, 316 deletions

diff --git a/remoting/host/chromoting_host.cc b/remoting/host/chromoting_host.cc
index 656327fb..d0b7bb7 100644
--- a/remoting/host/chromoting_host.cc
+++ b/remoting/host/chromoting_host.cc
@@ -9,13 +9,14 @@
 #include "build/build_config.h"
 #include "remoting/base/constants.h"
 #include "remoting/base/encoder.h"
-#include "remoting/base/protocol_decoder.h"
 #include "remoting/host/chromoting_host_context.h"
 #include "remoting/host/capturer.h"
 #include "remoting/host/event_executor.h"
 #include "remoting/host/host_config.h"
 #include "remoting/host/session_manager.h"
 #include "remoting/jingle_glue/jingle_channel.h"
+#include "remoting/protocol/messages_decoder.h"
+#include "remoting/protocol/jingle_chromoting_server.h"
 
 namespace remoting {
 
@@ -54,9 +55,6 @@ void ChromotingHost::Start(Task* shutdown_task) {
     state_ = kStarted;
   }
 
-  // Get capturer to set up it's initial configuration.
-  capturer_->ScreenConfigurationChanged();
-
   // Save the shutdown task.
   shutdown_task_.reset(shutdown_task);
 
@@ -64,11 +62,12 @@ void ChromotingHost::Start(Task* shutdown_task) {
   std::string xmpp_auth_token;
   if (!config_->GetString(kXmppLoginConfigPath, &xmpp_login) ||
       !config_->GetString(kXmppAuthTokenConfigPath, &xmpp_auth_token)) {
-    LOG(ERROR) << "XMPP credentials are not defined in config.";
+    LOG(ERROR) << "XMPP credentials are not defined in the config.";
     return;
   }
 
-  access_verifier_.Init(config_);
+  if (!access_verifier_.Init(config_))
+    return;
 
   // Connect to the talk network with a JingleClient.
   jingle_client_ = new JingleClient(context_->jingle_thread());
@@ -94,8 +93,10 @@ void ChromotingHost::Shutdown() {
   // No-op if this object is not started yet.
   {
     AutoLock auto_lock(lock_);
-    if (state_ != kStarted)
+    if (state_ != kStarted) {
+      state_ = kStopped;
       return;
+    }
     state_ = kStopped;
   }
 
@@ -110,16 +111,22 @@ void ChromotingHost::Shutdown() {
     client_->Disconnect();
   }
 
-  // Disconnect from the talk network.
-  if (jingle_client_) {
-    jingle_client_->Close();
-  }
-
   // Stop the heartbeat sender.
   if (heartbeat_sender_) {
     heartbeat_sender_->Stop();
   }
 
+  // Stop chromotocol server.
+  if (chromotocol_server_) {
+    chromotocol_server_->Close(
+        NewRunnableMethod(this, &ChromotingHost::OnServerClosed));
+  }
+
+  // Disconnect from the talk network.
+  if (jingle_client_) {
+    jingle_client_->Close();
+  }
+
   // Lastly call the shutdown task.
   if (shutdown_task_.get()) {
     shutdown_task_->Run();
@@ -168,15 +175,14 @@ void ChromotingHost::OnClientDisconnected(ClientConnection* client) {
 
 ////////////////////////////////////////////////////////////////////////////
 // ClientConnection::EventHandler implementations
-void ChromotingHost::HandleMessages(ClientConnection* client,
-                                    ClientMessageList* messages) {
+void ChromotingHost::HandleMessage(ClientConnection* client,
+                                   ChromotingClientMessage* message) {
   DCHECK_EQ(context_->main_message_loop(), MessageLoop::current());
 
   // Delegate the messages to EventExecutor and delete the unhandled
   // messages.
   DCHECK(executor_.get());
-  executor_->HandleInputEvents(messages);
-  STLDeleteElements<ClientMessageList>(messages);
+  executor_->HandleInputEvent(message);
 }
 
 void ChromotingHost::OnConnectionOpened(ClientConnection* client) {
@@ -210,7 +216,15 @@ void ChromotingHost::OnStateChange(JingleClient* jingle_client,
     LOG(INFO) << "Host connected as "
               << jingle_client->GetFullJid() << "." << std::endl;
 
-    // Start heartbeating after we've connected.
+    // Create and start |chromotocol_server_|.
+    chromotocol_server_ =
+        new JingleChromotingServer(context_->jingle_thread()->message_loop());
+    chromotocol_server_->Init(
+        jingle_client->GetFullJid(),
+        jingle_client->session_manager(),
+        NewCallback(this, &ChromotingHost::OnNewClientConnection));
+
+    // Start heartbeating.
     heartbeat_sender_->Start();
   } else if (state == JingleClient::CLOSED) {
     LOG(INFO) << "Host disconnected from talk network." << std::endl;
@@ -224,45 +238,44 @@ void ChromotingHost::OnStateChange(JingleClient* jingle_client,
   }
 }
 
-bool ChromotingHost::OnAcceptConnection(
-    JingleClient* jingle_client, const std::string& jid,
-    JingleChannel::Callback** channel_callback) {
+void ChromotingHost::OnNewClientConnection(
+    ChromotingConnection* connection, bool* accept) {
   AutoLock auto_lock(lock_);
-  if (state_ != kStarted)
-    return false;
-
-  DCHECK_EQ(jingle_client_.get(), jingle_client);
-
   // TODO(hclam): Allow multiple clients to connect to the host.
-  if (client_.get())
-    return false;
+  if (client_.get() || state_ != kStarted) {
+    *accept = false;
+    return;
+  }
 
   // Check that the user has access to the host.
-  if (!access_verifier_.VerifyPermissions(jid))
-    return false;
+  if (!access_verifier_.VerifyPermissions(connection->jid())) {
+    *accept = false;
+    return;
+  }
 
-  LOG(INFO) << "Client connected: " << jid << std::endl;
+  *accept = true;
+
+  LOG(INFO) << "Client connected: " << connection->jid() << std::endl;
 
   // If we accept the connected then create a client object and set the
   // callback.
-  client_ = new ClientConnection(context_->main_message_loop(),
-                                 new ProtocolDecoder(), this);
-  *channel_callback = client_.get();
-  return true;
+  client_ = new ClientConnection(context_->main_message_loop(), this);
+  client_->Init(connection);
+}
+
+bool ChromotingHost::OnAcceptConnection(
+    JingleClient* jingle_client, const std::string& jid,
+    JingleChannel::Callback** channel_callback) {
+  return false;
 }
 
 void ChromotingHost::OnNewConnection(JingleClient* jingle_client,
                                      scoped_refptr<JingleChannel> channel) {
-  AutoLock auto_lock(lock_);
-  if (state_ != kStarted)
-    return;
-
-  DCHECK_EQ(jingle_client_.get(), jingle_client);
+  NOTREACHED();
+}
 
-  // Since the session manager has not started, it is still safe to access
-  // the client directly. Note that we give the ownership of the channel
-  // to the client.
-  client_->set_jingle_channel(channel);
+void ChromotingHost::OnServerClosed() {
+  // Don't need to do anything here.
 }
 
 }  // namespace remoting
diff --git a/remoting/host/chromoting_host.h b/remoting/host/chromoting_host.h
index ce3bdda..c1773cb 100644
--- a/remoting/host/chromoting_host.h
+++ b/remoting/host/chromoting_host.h
@@ -27,6 +27,7 @@ class Encoder;
 class EventExecutor;
 class MutableHostConfig;
 class SessionManager;
+class JingleChromotingServer;
 
 // A class to implement the functionality of a host process.
 //
@@ -83,8 +84,8 @@ class ChromotingHost : public base::RefCountedThreadSafe<ChromotingHost>,
 
   ////////////////////////////////////////////////////////////////////////////
   // ClientConnection::EventHandler implementations
-  virtual void HandleMessages(ClientConnection* client,
-                              ClientMessageList* messages);
+  virtual void HandleMessage(ClientConnection* client,
+                             ChromotingClientMessage* message);
   virtual void OnConnectionOpened(ClientConnection* client);
   virtual void OnConnectionClosed(ClientConnection* client);
   virtual void OnConnectionFailed(ClientConnection* client);
@@ -99,6 +100,9 @@ class ChromotingHost : public base::RefCountedThreadSafe<ChromotingHost>,
       JingleClient* jingle,
       scoped_refptr<JingleChannel> channel);
 
+  // Callback for ChromotingServer.
+  void OnNewClientConnection(ChromotingConnection* connection, bool* accept);
+
  private:
   enum State {
     kInitial,
@@ -106,6 +110,13 @@ class ChromotingHost : public base::RefCountedThreadSafe<ChromotingHost>,
     kStopped,
   };
 
+  // This method connects to the talk network and start listening for incoming
+  // connections.
+  void DoStart(Task* shutdown_task);
+
+  // Callback for ChromotingServer::Close().
+  void OnServerClosed();
+
   // The context that the chromoting host runs on.
   ChromotingHostContext* context_;
 
@@ -126,6 +137,8 @@ class ChromotingHost : public base::RefCountedThreadSafe<ChromotingHost>,
   // receive connection requests from chromoting client.
   scoped_refptr<JingleClient> jingle_client_;
 
+  scoped_refptr<JingleChromotingServer> chromotocol_server_;
+
   // Objects that takes care of sending heartbeats to the chromoting bot.
   scoped_refptr<HeartbeatSender> heartbeat_sender_;
 
diff --git a/remoting/host/client_connection.cc b/remoting/host/client_connection.cc
index 2cbbed6..8c8bc94 100644
--- a/remoting/host/client_connection.cc
+++ b/remoting/host/client_connection.cc
@@ -5,11 +5,9 @@
 #include "remoting/host/client_connection.h"
 
 #include "google/protobuf/message.h"
-#include "media/base/data_buffer.h"
-#include "remoting/base/protocol_decoder.h"
-#include "remoting/base/protocol_util.h"
-
-using media::DataBuffer;
+#include "net/base/io_buffer.h"
+#include "remoting/protocol/messages_decoder.h"
+#include "remoting/protocol/util.h"
 
 namespace remoting {
 
@@ -18,165 +16,100 @@ namespace remoting {
 static const size_t kAverageUpdateStream = 10;
 
 ClientConnection::ClientConnection(MessageLoop* message_loop,
-                                   ProtocolDecoder* decoder,
                                    EventHandler* handler)
     : loop_(message_loop),
-      decoder_(decoder),
-      size_in_queue_(0),
-      update_stream_size_(0),
       handler_(handler) {
   DCHECK(loop_);
-  DCHECK(decoder_.get());
   DCHECK(handler_);
 }
 
 ClientConnection::~ClientConnection() {
   // TODO(hclam): When we shut down the viewer we may have to close the
-  // jingle channel.
+  // connection.
 }
 
-// static
-scoped_refptr<media::DataBuffer>
-    ClientConnection::CreateWireFormatDataBuffer(
-        const ChromotingHostMessage* msg) {
-  // TODO(hclam): Instead of serializing |msg| create an DataBuffer
-  // object that wraps around it.
-  scoped_ptr<const ChromotingHostMessage> message_deleter(msg);
-  return SerializeAndFrameMessage(*msg);
+void ClientConnection::Init(ChromotingConnection* connection) {
+  DCHECK_EQ(connection->message_loop(), MessageLoop::current());
+
+  connection_ = connection;
+  connection_->SetStateChangeCallback(
+      NewCallback(this, &ClientConnection::OnConnectionStateChange));
 }
 
 void ClientConnection::SendInitClientMessage(int width, int height) {
   DCHECK_EQ(loop_, MessageLoop::current());
-  DCHECK(!update_stream_size_);
 
   // If we are disconnected then return.
-  if (!channel_)
+  if (!connection_)
     return;
 
   ChromotingHostMessage msg;
   msg.mutable_init_client()->set_width(width);
   msg.mutable_init_client()->set_height(height);
   DCHECK(msg.IsInitialized());
-  channel_->Write(SerializeAndFrameMessage(msg));
-}
-
-void ClientConnection::SendBeginUpdateStreamMessage() {
-  DCHECK_EQ(loop_, MessageLoop::current());
-
-  // If we are disconnected then return.
-  if (!channel_)
-    return;
-
-  ChromotingHostMessage msg;
-  msg.mutable_begin_update_stream();
-  DCHECK(msg.IsInitialized());
-
-  scoped_refptr<DataBuffer> data = SerializeAndFrameMessage(msg);
-  DCHECK(!update_stream_size_);
-  update_stream_size_ += data->GetDataSize();
-  channel_->Write(data);
+  video_writer_.SendMessage(msg);
 }
 
 void ClientConnection::SendUpdateStreamPacketMessage(
-    scoped_refptr<DataBuffer> data) {
+    const ChromotingHostMessage& message) {
   DCHECK_EQ(loop_, MessageLoop::current());
 
   // If we are disconnected then return.
-  if (!channel_)
+  if (!connection_)
     return;
 
-  update_stream_size_ += data->GetDataSize();
-  channel_->Write(data);
-}
-
-void ClientConnection::SendEndUpdateStreamMessage() {
-  DCHECK_EQ(loop_, MessageLoop::current());
-
-  // If we are disconnected then return.
-  if (!channel_)
-    return;
-
-  ChromotingHostMessage msg;
-  msg.mutable_end_update_stream();
-  DCHECK(msg.IsInitialized());
-
-  scoped_refptr<DataBuffer> data = SerializeAndFrameMessage(msg);
-  update_stream_size_ += data->GetDataSize();
-  channel_->Write(data);
-
-  // Here's some logic to help finding the average update stream size.
-  size_in_queue_ += update_stream_size_;
-  size_queue_.push_back(update_stream_size_);
-  if (size_queue_.size() > kAverageUpdateStream) {
-    size_in_queue_ -= size_queue_.front();
-    size_queue_.pop_front();
-    DCHECK_GE(size_in_queue_, 0);
-  }
-  update_stream_size_ = 0;
-}
-
-void ClientConnection::MarkEndOfUpdate() {
-  // This is some logic to help calculate the average update stream size.
-  size_in_queue_ += update_stream_size_;
-  size_queue_.push_back(update_stream_size_);
-  if (size_queue_.size() > kAverageUpdateStream) {
-    size_in_queue_ -= size_queue_.front();
-    size_queue_.pop_front();
-    DCHECK_GE(size_in_queue_, 0);
-  }
-  update_stream_size_ = 0;
+  video_writer_.SendMessage(message);
 }
 
 int ClientConnection::GetPendingUpdateStreamMessages() {
   DCHECK_EQ(loop_, MessageLoop::current());
-
-  if (!size_queue_.size())
-    return 0;
-  int average_size = size_in_queue_ / size_queue_.size();
-  if (!average_size)
-    return 0;
-  return channel_->write_buffer_size() / average_size;
+  return video_writer_.GetPendingMessages();
 }
 
 void ClientConnection::Disconnect() {
   DCHECK_EQ(loop_, MessageLoop::current());
 
   // If there is a channel then close it and release the reference.
-  if (channel_) {
-    channel_->Close();
-    channel_ = NULL;
+  if (connection_) {
+    connection_->Close(NewRunnableMethod(this, &ClientConnection::OnClosed));
+    connection_ = NULL;
   }
 }
 
-void ClientConnection::OnStateChange(JingleChannel* channel,
-                                     JingleChannel::State state) {
-  DCHECK(channel);
+void ClientConnection::OnConnectionStateChange(
+    ChromotingConnection::State state) {
+  if (state == ChromotingConnection::CONNECTED) {
+    events_reader_.Init(
+        connection_->GetEventsChannel(),
+        NewCallback(this, &ClientConnection::OnMessageReceived));
+    video_writer_.Init(connection_->GetVideoChannel());
+  }
+
   loop_->PostTask(FROM_HERE,
       NewRunnableMethod(this, &ClientConnection::StateChangeTask, state));
 }
 
-void ClientConnection::OnPacketReceived(JingleChannel* channel,
-                                        scoped_refptr<DataBuffer> data) {
-  DCHECK_EQ(channel_.get(), channel);
+void ClientConnection::OnMessageReceived(ChromotingClientMessage* message) {
   loop_->PostTask(FROM_HERE,
-      NewRunnableMethod(this, &ClientConnection::PacketReceivedTask, data));
+      NewRunnableMethod(this, &ClientConnection::MessageReceivedTask,
+                        message));
 }
 
-void ClientConnection::StateChangeTask(JingleChannel::State state) {
+void ClientConnection::StateChangeTask(ChromotingConnection::State state) {
   DCHECK_EQ(loop_, MessageLoop::current());
 
   DCHECK(handler_);
   switch(state) {
-    case JingleChannel::CONNECTING:
+    case ChromotingConnection::CONNECTING:
       break;
     // Don't care about this message.
-    case JingleChannel::OPEN:
+    case ChromotingConnection::CONNECTED:
       handler_->OnConnectionOpened(this);
       break;
-    case JingleChannel::CLOSED:
+    case ChromotingConnection::CLOSED:
       handler_->OnConnectionClosed(this);
       break;
-    case JingleChannel::FAILED:
+    case ChromotingConnection::FAILED:
       handler_->OnConnectionFailed(this);
       break;
     default:
@@ -185,17 +118,14 @@ void ClientConnection::StateChangeTask(JingleChannel::State state) {
   }
 }
 
-void ClientConnection::PacketReceivedTask(scoped_refptr<DataBuffer> data) {
+void ClientConnection::MessageReceivedTask(ChromotingClientMessage* message) {
   DCHECK_EQ(loop_, MessageLoop::current());
-
-  // Use the decoder to parse incoming data.
-  DCHECK(decoder_.get());
-  ClientMessageList list;
-  decoder_->ParseClientMessages(data, &list);
-
-  // Then submit the messages to the handler.
   DCHECK(handler_);
-  handler_->HandleMessages(this, &list);
+  handler_->HandleMessage(this, message);
+}
+
+// OnClosed() is used as a callback for ChromotingConnection::Close().
+void ClientConnection::OnClosed() {
 }
 
 }  // namespace remoting
diff --git a/remoting/host/client_connection.h b/remoting/host/client_connection.h
index 0ea2b26..44bb761 100644
--- a/remoting/host/client_connection.h
+++ b/remoting/host/client_connection.h
@@ -11,9 +11,10 @@
 #include "base/message_loop.h"
 #include "base/ref_counted.h"
 #include "base/scoped_ptr.h"
-#include "remoting/base/protocol_decoder.h"
 #include "remoting/base/protocol/chromotocol.pb.h"
-#include "remoting/jingle_glue/jingle_channel.h"
+#include "remoting/protocol/chromoting_connection.h"
+#include "remoting/protocol/stream_reader.h"
+#include "remoting/protocol/stream_writer.h"
 
 namespace media {
 
@@ -28,8 +29,7 @@ namespace remoting {
 // screen updates and other messages to the remote viewer. It is also
 // responsible for receiving and parsing data from the remote viewer and
 // delegating events to the event handler.
-class ClientConnection : public base::RefCountedThreadSafe<ClientConnection>,
-                         public JingleChannel::Callback {
+class ClientConnection : public base::RefCountedThreadSafe<ClientConnection> {
  public:
   class EventHandler {
    public:
@@ -39,8 +39,8 @@ class ClientConnection : public base::RefCountedThreadSafe<ClientConnection>,
     // ClientMessages in ClientMessageList and needs to delete them.
     // Note that the sender of messages will not reference messages
     // again so it is okay to clear |messages| in this method.
-    virtual void HandleMessages(ClientConnection* viewer,
-                                ClientMessageList* messages) = 0;
+    virtual void HandleMessage(ClientConnection* viewer,
+                               ChromotingClientMessage* message) = 0;
 
     // Called when the network connection is opened.
     virtual void OnConnectionOpened(ClientConnection* viewer) = 0;
@@ -57,45 +57,21 @@ class ClientConnection : public base::RefCountedThreadSafe<ClientConnection>,
   // a libjingle channel, these events are delegated to |handler|.
   // It is guranteed that |handler| is called only on the |message_loop|.
   ClientConnection(MessageLoop* message_loop,
-                   ProtocolDecoder* decoder,
                    EventHandler* handler);
 
   virtual ~ClientConnection();
 
-  // Creates a DataBuffer object that wraps around ChromotingHostMessage. The
-  // DataBuffer object will be responsible for serializing and framing the
-  // message. DataBuffer will also own |msg| after this call.
-  static scoped_refptr<media::DataBuffer> CreateWireFormatDataBuffer(
-      const ChromotingHostMessage* msg);
+  virtual void Init(ChromotingConnection* connection);
 
-  virtual void set_jingle_channel(JingleChannel* channel) {
-    channel_ = channel;
-  }
-
-  // Returns the channel in use.
-  virtual JingleChannel* jingle_channel() { return channel_; }
+  // Returns the connection in use.
+  virtual ChromotingConnection* connection() { return connection_; }
 
   // Send information to the client for initialization.
   virtual void SendInitClientMessage(int width, int height);
 
-  // Notifies the viewer the start of an update stream.
-  virtual void SendBeginUpdateStreamMessage();
-
   // Send encoded update stream data to the viewer.
-  //
-  // |data| is the actual bytes in wire format. That means it is fully framed
-  // and serialized from a ChromotingHostMessage. This is a special case only
-  // for UpdateStreamPacket to reduce the amount of memory copies.
-  //
-  // |data| should be created by calling to
-  // CreateWireFormatDataBuffer(ChromotingHostMessage).
   virtual void SendUpdateStreamPacketMessage(
-      scoped_refptr<media::DataBuffer> data);
-
-  // Notifies the viewer the update stream has ended.
-  virtual void SendEndUpdateStreamMessage();
-
-  virtual void MarkEndOfUpdate();
+      const ChromotingHostMessage& message);
 
   // Gets the number of update stream messages not yet transmitted.
   // Note that the value returned is an estimate using average size of the
@@ -109,43 +85,34 @@ class ClientConnection : public base::RefCountedThreadSafe<ClientConnection>,
   // After this method is called all the send method calls will be ignored.
   virtual void Disconnect();
 
-  /////////////////////////////////////////////////////////////////////////////
-  // JingleChannel::Callback implmentations
-  virtual void OnStateChange(JingleChannel* channel,
-                             JingleChannel::State state);
-  virtual void OnPacketReceived(JingleChannel* channel,
-                                scoped_refptr<media::DataBuffer> data);
-
  protected:
   // Protected constructor used by unit test.
   ClientConnection() {}
 
  private:
+  // Callback for ChromotingConnection.
+  void OnConnectionStateChange(ChromotingConnection::State state);
+
+  // Callback for EventsStreamReader.
+  void OnMessageReceived(ChromotingClientMessage* message);
+
   // Process a libjingle state change event on the |loop_|.
-  void StateChangeTask(JingleChannel::State state);
+  void StateChangeTask(ChromotingConnection::State state);
 
   // Process a data buffer received from libjingle.
-  void PacketReceivedTask(scoped_refptr<media::DataBuffer> data);
+  void MessageReceivedTask(ChromotingClientMessage* message);
+
+  void OnClosed();
 
   // The libjingle channel used to send and receive data from the remote client.
-  scoped_refptr<JingleChannel> channel_;
+  scoped_refptr<ChromotingConnection> connection_;
+
+  EventsStreamReader events_reader_;
+  VideoStreamWriter video_writer_;
 
   // The message loop that this object runs on.
   MessageLoop* loop_;
 
-  // An object used by the ClientConnection to decode data received from the
-  // network.
-  scoped_ptr<ProtocolDecoder> decoder_;
-
-  // A queue to count the sizes of the last 10 update streams.
-  std::deque<int> size_queue_;
-
-  // Count the sum of sizes in the queue.
-  int size_in_queue_;
-
-  // Measure the number of bytes of the current upstream stream.
-  int update_stream_size_;
-
   // Event handler for handling events sent from this object.
   EventHandler* handler_;
 
diff --git a/remoting/host/client_connection_unittest.cc b/remoting/host/client_connection_unittest.cc
index 9c3bcb3..317d7fa 100644
--- a/remoting/host/client_connection_unittest.cc
+++ b/remoting/host/client_connection_unittest.cc
@@ -7,7 +7,7 @@
 #include "remoting/base/mock_objects.h"
 #include "remoting/host/client_connection.h"
 #include "remoting/host/mock_objects.h"
-#include "remoting/jingle_glue/mock_objects.h"
+#include "remoting/protocol/fake_connection.h"
 #include "testing/gmock/include/gmock/gmock.h"
 
 using ::testing::_;
@@ -23,86 +23,67 @@ class ClientConnectionTest : public testing::Test {
 
  protected:
   virtual void SetUp() {
-    decoder_ = new MockProtocolDecoder();
-    channel_ = new StrictMock<MockJingleChannel>();
-
-    // Allocate a ClientConnection object with the mock objects. we give the
-    // ownership of decoder to the viewer.
-    viewer_ = new ClientConnection(&message_loop_,
-                                   decoder_,
-                                   &handler_);
-
-    viewer_->set_jingle_channel(channel_.get());
+    connection_ = new FakeChromotingConnection();
+    connection_->set_message_loop(&message_loop_);
+
+    // Allocate a ClientConnection object with the mock objects.
+    viewer_ = new ClientConnection(&message_loop_, &handler_);
+    viewer_->Init(connection_);
+    EXPECT_CALL(handler_, OnConnectionOpened(viewer_.get()));
+    connection_->get_state_change_callback()->Run(
+        ChromotingConnection::CONNECTED);
+    message_loop_.RunAllPending();
   }
 
   MessageLoop message_loop_;
-  MockProtocolDecoder* decoder_;
   MockClientConnectionEventHandler handler_;
   scoped_refptr<ClientConnection> viewer_;
 
-  // |channel_| is wrapped with StrictMock because we limit strictly the calls
-  // to it.
-  scoped_refptr<StrictMock<MockJingleChannel> > channel_;
+  scoped_refptr<FakeChromotingConnection> connection_;
 
  private:
   DISALLOW_COPY_AND_ASSIGN(ClientConnectionTest);
 };
 
 TEST_F(ClientConnectionTest, SendUpdateStream) {
-  // Tell the viewer we are starting an update stream.
-  EXPECT_CALL(*channel_, Write(_));
-  viewer_->SendBeginUpdateStreamMessage();
-
   // Then send the actual data.
-  EXPECT_CALL(*channel_, Write(_));
-  scoped_refptr<media::DataBuffer> data = new media::DataBuffer(10);
-  viewer_->SendUpdateStreamPacketMessage(data);
-
-  // Send the end of update message.
-  EXPECT_CALL(*channel_, Write(_));
-  viewer_->SendEndUpdateStreamMessage();
+  ChromotingHostMessage message;
+  viewer_->SendUpdateStreamPacketMessage(message);
 
   // And then close the connection to ClientConnection.
-  EXPECT_CALL(*channel_, Close());
   viewer_->Disconnect();
-}
 
-TEST_F(ClientConnectionTest, StateChange) {
-  EXPECT_CALL(handler_, OnConnectionOpened(viewer_.get()));
-  viewer_->OnStateChange(channel_.get(), JingleChannel::OPEN);
   message_loop_.RunAllPending();
 
+  // Verify that something has been written.
+  // TODO(sergeyu): Verify that the correct data has been written.
+  EXPECT_GT(connection_->GetVideoChannel()->written_data().size(), 0u);
+}
+
+TEST_F(ClientConnectionTest, StateChange) {
   EXPECT_CALL(handler_, OnConnectionClosed(viewer_.get()));
-  viewer_->OnStateChange(channel_.get(), JingleChannel::CLOSED);
+  connection_->get_state_change_callback()->Run(ChromotingConnection::CLOSED);
   message_loop_.RunAllPending();
 
   EXPECT_CALL(handler_, OnConnectionFailed(viewer_.get()));
-  viewer_->OnStateChange(channel_.get(), JingleChannel::FAILED);
-  message_loop_.RunAllPending();
-}
-
-TEST_F(ClientConnectionTest, ParseMessages) {
-  scoped_refptr<media::DataBuffer> data;
-
-  // Give the data to the ClientConnection, it will use ProtocolDecoder to
-  // decode the messages.
-  EXPECT_CALL(*decoder_, ParseClientMessages(data, NotNull()));
-  EXPECT_CALL(handler_, HandleMessages(viewer_.get(), NotNull()));
-
-  viewer_->OnPacketReceived(channel_.get(), data);
+  connection_->get_state_change_callback()->Run(ChromotingConnection::FAILED);
   message_loop_.RunAllPending();
 }
 
 // Test that we can close client connection more than once and operations
 // after the connection is closed has no effect.
 TEST_F(ClientConnectionTest, Close) {
-  EXPECT_CALL(*channel_, Close());
   viewer_->Disconnect();
+  message_loop_.RunAllPending();
+  EXPECT_TRUE(connection_->is_closed());
 
-  scoped_refptr<media::DataBuffer> data = new media::DataBuffer(10);
-  viewer_->SendUpdateStreamPacketMessage(data);
-  viewer_->SendEndUpdateStreamMessage();
+  ChromotingHostMessage message;
+  viewer_->SendUpdateStreamPacketMessage(message);
   viewer_->Disconnect();
+  message_loop_.RunAllPending();
+
+  // Verify that nothing has been written.
+  EXPECT_EQ(connection_->GetVideoChannel()->written_data().size(), 0u);
 }
 
 }  // namespace remoting
diff --git a/remoting/host/event_executor.h b/remoting/host/event_executor.h
index bfdf38f..3f87ea7 100644
--- a/remoting/host/event_executor.h
+++ b/remoting/host/event_executor.h
@@ -7,11 +7,12 @@
 
 #include <vector>
 
-#include "remoting/base/protocol_decoder.h"
+#include "base/basictypes.h"
 
 namespace remoting {
 
 class Capturer;
+class ChromotingClientMessage;
 
 // An interface that defines the behavior of an event executor object.
 // An event executor is to perform actions on the host machine. For example
@@ -19,14 +20,14 @@ class Capturer;
 // clipboards.
 class EventExecutor {
  public:
-  EventExecutor(Capturer* capturer)
+  explicit EventExecutor(Capturer* capturer)
     : capturer_(capturer) {
   }
   virtual ~EventExecutor() {}
 
   // Handles input events from ClientMessageList and removes them from the
   // list.
-  virtual void HandleInputEvents(ClientMessageList* messages) = 0;
+  virtual void HandleInputEvent(ChromotingClientMessage* message) = 0;
   // TODO(hclam): Define actions for clipboards.
 
  protected:
diff --git a/remoting/host/event_executor_linux.cc b/remoting/host/event_executor_linux.cc
index d2e4a13..44c4c55 100644
--- a/remoting/host/event_executor_linux.cc
+++ b/remoting/host/event_executor_linux.cc
@@ -4,6 +4,8 @@
 
 #include "remoting/host/event_executor_linux.h"
 
+#include "remoting/protocol/messages_decoder.h"
+
 namespace remoting {
 
 EventExecutorLinux::EventExecutorLinux(Capturer* capturer)
@@ -13,7 +15,8 @@ EventExecutorLinux::EventExecutorLinux(Capturer* capturer)
 EventExecutorLinux::~EventExecutorLinux() {
 }
 
-void EventExecutorLinux::HandleInputEvents(ClientMessageList* messages) {
+void EventExecutorLinux::HandleInputEvent(ChromotingClientMessage* message) {
+  delete message;
 }
 
 }  // namespace remoting
diff --git a/remoting/host/event_executor_linux.h b/remoting/host/event_executor_linux.h
index 44b8209..bdb3e73 100644
--- a/remoting/host/event_executor_linux.h
+++ b/remoting/host/event_executor_linux.h
@@ -17,7 +17,7 @@ class EventExecutorLinux : public EventExecutor {
   EventExecutorLinux(Capturer* capturer);
   virtual ~EventExecutorLinux();
 
-  virtual void HandleInputEvents(ClientMessageList* messages);
+  virtual void HandleInputEvent(ChromotingClientMessage* message);
 
  private:
   DISALLOW_COPY_AND_ASSIGN(EventExecutorLinux);
diff --git a/remoting/host/event_executor_mac.cc b/remoting/host/event_executor_mac.cc
index 3a1256b..89acece 100644
--- a/remoting/host/event_executor_mac.cc
+++ b/remoting/host/event_executor_mac.cc
@@ -13,7 +13,8 @@ EventExecutorMac::EventExecutorMac(Capturer* capturer)
 EventExecutorMac::~EventExecutorMac() {
 }
 
-void EventExecutorMac::HandleInputEvents(ClientMessageList* messages) {
+void EventExecutorMac::HandleInputEvent(ChromotingClientMessage* message) {
+  delete message;
 }
 
 }  // namespace remoting
diff --git a/remoting/host/event_executor_mac.h b/remoting/host/event_executor_mac.h
index 854d274..f8934b1 100644
--- a/remoting/host/event_executor_mac.h
+++ b/remoting/host/event_executor_mac.h
@@ -17,7 +17,7 @@ class EventExecutorMac : public EventExecutor {
   EventExecutorMac(Capturer* capturer);
   virtual ~EventExecutorMac();
 
-  virtual void HandleInputEvents(ClientMessageList* messages);
+  virtual void HandleInputEvent(ChromotingClientMessage* message);
 
  private:
   DISALLOW_COPY_AND_ASSIGN(EventExecutorMac);
diff --git a/remoting/host/event_executor_win.cc b/remoting/host/event_executor_win.cc
index 24aa841..762cd0c 100644
--- a/remoting/host/event_executor_win.cc
+++ b/remoting/host/event_executor_win.cc
@@ -355,28 +355,21 @@ EventExecutorWin::EventExecutorWin(Capturer* capturer)
 EventExecutorWin::~EventExecutorWin() {
 }
 
-void EventExecutorWin::HandleInputEvents(ClientMessageList* messages) {
-  for (ClientMessageList::iterator it = messages->begin();
-       it != messages->end();
-       ++it) {
-    ChromotingClientMessage* msg = *it;
-    if (msg->has_mouse_set_position_event()) {
-      HandleMouseSetPosition(msg);
-    } else if (msg->has_mouse_move_event()) {
-      HandleMouseMove(msg);
-    } else if (msg->has_mouse_wheel_event()) {
-      HandleMouseWheel(msg);
-    } else if (msg->has_mouse_down_event()) {
-      HandleMouseButtonDown(msg);
-    } else if (msg->has_mouse_up_event()) {
-      HandleMouseButtonUp(msg);
-    } else if (msg->has_key_event()) {
-      HandleKey(msg);
-    }
+void EventExecutorWin::HandleInputEvent(ChromotingClientMessage* msg) {
+  if (msg->has_mouse_set_position_event()) {
+    HandleMouseSetPosition(msg);
+  } else if (msg->has_mouse_move_event()) {
+    HandleMouseMove(msg);
+  } else if (msg->has_mouse_wheel_event()) {
+    HandleMouseWheel(msg);
+  } else if (msg->has_mouse_down_event()) {
+    HandleMouseButtonDown(msg);
+  } else if (msg->has_mouse_up_event()) {
+    HandleMouseButtonUp(msg);
+  } else if (msg->has_key_event()) {
+    HandleKey(msg);
   }
-  // We simply delete all messages.
-  // TODO(hclam): Delete messages processed.
-  STLDeleteElements<ClientMessageList>(messages);
+  delete msg;
 }
 
 void EventExecutorWin::HandleMouseSetPosition(ChromotingClientMessage* msg) {
diff --git a/remoting/host/event_executor_win.h b/remoting/host/event_executor_win.h
index 4a354e6..e9a0269 100644
--- a/remoting/host/event_executor_win.h
+++ b/remoting/host/event_executor_win.h
@@ -17,7 +17,7 @@ class EventExecutorWin : public EventExecutor {
   EventExecutorWin(Capturer* capturer);
   virtual ~EventExecutorWin();
 
-  virtual void HandleInputEvents(ClientMessageList* messages);
+  virtual void HandleInputEvent(ChromotingClientMessage* message);
 
  private:
   void HandleMouseSetPosition(ChromotingClientMessage* msg);
diff --git a/remoting/host/mock_objects.h b/remoting/host/mock_objects.h
index 9e6b631..64dcc55 100644
--- a/remoting/host/mock_objects.h
+++ b/remoting/host/mock_objects.h
@@ -6,10 +6,10 @@
 #define REMOTING_HOST_MOCK_OBJECTS_H_
 
 #include "media/base/data_buffer.h"
-#include "remoting/base/protocol_decoder.h"
 #include "remoting/host/capturer.h"
 #include "remoting/host/client_connection.h"
 #include "remoting/host/event_executor.h"
+#include "remoting/protocol/messages_decoder.h"
 #include "testing/gmock/include/gmock/gmock.h"
 
 namespace remoting {
@@ -36,7 +36,7 @@ class MockEventExecutor : public EventExecutor {
  public:
    MockEventExecutor(Capturer* capturer) : EventExecutor(capturer) {}
 
-  MOCK_METHOD1(HandleInputEvents, void(ClientMessageList* messages));
+  MOCK_METHOD1(HandleInputEvent, void(ChromotingClientMessage* messages));
 
  private:
   DISALLOW_COPY_AND_ASSIGN(MockEventExecutor);
@@ -46,17 +46,14 @@ class MockClientConnection : public ClientConnection {
  public:
   MockClientConnection(){}
 
+  MOCK_METHOD1(Init, void(ChromotingConnection* connection));
   MOCK_METHOD2(SendInitClientMessage, void(int width, int height));
   MOCK_METHOD0(SendBeginUpdateStreamMessage, void());
   MOCK_METHOD1(SendUpdateStreamPacketMessage,
-               void(scoped_refptr<media::DataBuffer> data));
+               void(const ChromotingHostMessage& message));
   MOCK_METHOD0(SendEndUpdateStreamMessage, void());
   MOCK_METHOD0(GetPendingUpdateStreamMessages, int());
-
-  MOCK_METHOD2(OnStateChange, void(JingleChannel* channel,
-                                   JingleChannel::State state));
-  MOCK_METHOD2(OnPacketReceived, void(JingleChannel* channel,
-                                      scoped_refptr<media::DataBuffer> data));
+  MOCK_METHOD0(Disconnect, void());
 
  private:
   DISALLOW_COPY_AND_ASSIGN(MockClientConnection);
@@ -66,8 +63,9 @@ class MockClientConnectionEventHandler : public ClientConnection::EventHandler {
  public:
   MockClientConnectionEventHandler() {}
 
-  MOCK_METHOD2(HandleMessages,
-               void(ClientConnection* viewer, ClientMessageList* messages));
+  MOCK_METHOD2(HandleMessage,
+               void(ClientConnection* viewer,
+                    ChromotingClientMessage* message));
   MOCK_METHOD1(OnConnectionOpened, void(ClientConnection* viewer));
   MOCK_METHOD1(OnConnectionClosed, void(ClientConnection* viewer));
   MOCK_METHOD1(OnConnectionFailed, void(ClientConnection* viewer));
diff --git a/remoting/host/session_manager.cc b/remoting/host/session_manager.cc
index 7ab3bc5..4aacfc8 100644
--- a/remoting/host/session_manager.cc
+++ b/remoting/host/session_manager.cc
@@ -11,9 +11,9 @@
 #include "base/stl_util-inl.h"
 #include "media/base/data_buffer.h"
 #include "remoting/base/capture_data.h"
-#include "remoting/base/protocol_decoder.h"
 #include "remoting/base/tracer.h"
 #include "remoting/host/client_connection.h"
+#include "remoting/protocol/messages_decoder.h"
 
 namespace remoting {
 
@@ -333,24 +333,15 @@ void SessionManager::DoSendUpdate(ChromotingHostMessage* message,
                                   Encoder::EncodingState state) {
   DCHECK_EQ(network_loop_, MessageLoop::current());
 
-  // TODO(ajwong): We shouldn't need EncodingState. Just inspect message.
-  bool is_end_of_update = (message->rectangle_update().flags() |
-      RectangleUpdatePacket::LAST_PACKET) != 0;
-
   TraceContext::tracer()->PrintString("DoSendUpdate");
 
-  // Create a data buffer in wire format from |message|.
-  // Note that this takes ownership of |message|.
-  scoped_refptr<media::DataBuffer> data =
-      ClientConnection::CreateWireFormatDataBuffer(message);
-
   for (ClientConnectionList::const_iterator i = clients_.begin();
        i < clients_.end(); ++i) {
-    (*i)->SendUpdateStreamPacketMessage(data);
-
-    if (is_end_of_update)
-      (*i)->MarkEndOfUpdate();
+    (*i)->SendUpdateStreamPacketMessage(*message);
   }
+
+  delete message;
+
   TraceContext::tracer()->PrintString("DoSendUpdate done");
 }