[Blimp Net] Adds BrowserConnectionHandler.

It Contains a demultiplexer for routing incoming messages to their respective
features, and a multiplexer/buffer object for writing outgoing messages.

It is created on browser startup, and persists for the lifetime of the application.

BUG=561207

Review URL: https://codereview.chromium.org/1474583002

Cr-Commit-Position: refs/heads/master@{#362192}

5 files changed, 168 insertions, 3 deletions

diff --git a/blimp/net/BUILD.gn b/blimp/net/BUILD.gn
index 63de25b..5ab0ab7 100644
--- a/blimp/net/BUILD.gn
+++ b/blimp/net/BUILD.gn
@@ -19,6 +19,8 @@ component("blimp_net") {
     "blimp_message_receiver.h",
     "blimp_net_export.h",
     "blimp_transport.h",
+    "browser_connection_handler.cc",
+    "browser_connection_handler.h",
     "client_connection_manager.cc",
     "client_connection_manager.h",
     "common.cc",
diff --git a/blimp/net/blimp_message_pump.cc b/blimp/net/blimp_message_pump.cc
index 14d8d2e..70c3217 100644
--- a/blimp/net/blimp_message_pump.cc
+++ b/blimp/net/blimp_message_pump.cc
@@ -26,10 +26,11 @@ BlimpMessagePump::BlimpMessagePump(PacketReader* reader)
 BlimpMessagePump::~BlimpMessagePump() {}
 
 void BlimpMessagePump::SetMessageProcessor(BlimpMessageProcessor* processor) {
-  DCHECK(processor);
-  bool reading_packet = !!processor_;
+  process_msg_callback_.Cancel();
   processor_ = processor;
-  if (!reading_packet) {
+  if (!processor_) {
+    read_packet_callback_.Cancel();
+  } else if (read_packet_callback_.IsCancelled()) {
     buffer_->SetCapacity(kMaxPacketPayloadSizeBytes);
     ReadNextPacket();
   }
@@ -49,6 +50,7 @@ void BlimpMessagePump::ReadNextPacket() {
 }
 
 void BlimpMessagePump::OnReadPacketComplete(int result) {
+  read_packet_callback_.Cancel();
   if (result > 0) {
     // The result is the size of the packet in bytes.
     scoped_ptr<BlimpMessage> message(new BlimpMessage);
@@ -70,6 +72,7 @@ void BlimpMessagePump::OnReadPacketComplete(int result) {
 void BlimpMessagePump::OnProcessMessageComplete(int result) {
   // No error is expected from the message receiver.
   DCHECK_EQ(result, net::OK);
+  process_msg_callback_.Cancel();
   ReadNextPacket();
 }
 
diff --git a/blimp/net/blimp_message_pump_unittest.cc b/blimp/net/blimp_message_pump_unittest.cc
index ec7987e..7f9ab11 100644
--- a/blimp/net/blimp_message_pump_unittest.cc
+++ b/blimp/net/blimp_message_pump_unittest.cc
@@ -5,6 +5,7 @@
 #include <stddef.h>
 #include <string>
 
+#include "base/callback_helpers.h"
 #include "blimp/common/proto/blimp_message.pb.h"
 #include "blimp/net/blimp_message_pump.h"
 #include "blimp/net/common.h"
@@ -153,6 +154,38 @@ TEST_F(BlimpMessagePumpTest, InvalidPacket) {
   message_pump_->SetMessageProcessor(&receiver_);
 }
 
+TEST_F(BlimpMessagePumpTest, ClearMessageProcessorAfterRead) {
+  net::CompletionCallback read_packet_cb;
+  EXPECT_CALL(reader_, ReadPacket(NotNull(), _))
+      .WillOnce(DoAll(FillBufferFromMessage<0>(message1_.get()),
+                      SaveArg<1>(&read_packet_cb),
+                      Return(net::ERR_IO_PENDING)));
+  net::CompletionCallback process_msg_cb;
+  EXPECT_CALL(receiver_, MockableProcessMessage(EqualsProto(*message1_), _))
+      .WillOnce(SaveArg<1>(&process_msg_cb));
+
+  message_pump_->SetMessageProcessor(&receiver_);
+  base::ResetAndReturn(&read_packet_cb).Run(message1_->ByteSize());
+
+  message_pump_->SetMessageProcessor(nullptr);
+
+  // Completing message processing will not trigger next packet read.
+  base::ResetAndReturn(&process_msg_cb).Run(net::OK);
+}
+
+TEST_F(BlimpMessagePumpTest, ClearMessageProcessorDuringRead) {
+  net::CompletionCallback read_packet_cb;
+  EXPECT_CALL(reader_, ReadPacket(NotNull(), _))
+      .WillOnce(DoAll(FillBufferFromMessage<0>(message1_.get()),
+                      SaveArg<1>(&read_packet_cb),
+                      Return(net::ERR_IO_PENDING)));
+
+  // Receiver will not get any message.
+  message_pump_->SetMessageProcessor(&receiver_);
+  message_pump_->SetMessageProcessor(nullptr);
+  base::ResetAndReturn(&read_packet_cb).Run(message1_->ByteSize());
+}
+
 }  // namespace
 
 }  // namespace blimp
diff --git a/blimp/net/browser_connection_handler.cc b/blimp/net/browser_connection_handler.cc
new file mode 100644
index 0000000..d47623e8
--- /dev/null
+++ b/blimp/net/browser_connection_handler.cc
@@ -0,0 +1,57 @@
+// Copyright 2015 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 "blimp/net/browser_connection_handler.h"
+
+#include "base/logging.h"
+#include "base/macros.h"
+#include "blimp/net/blimp_connection.h"
+#include "blimp/net/blimp_message_demultiplexer.h"
+#include "blimp/net/blimp_message_multiplexer.h"
+#include "blimp/net/blimp_message_output_buffer.h"
+#include "blimp/net/blimp_message_processor.h"
+
+namespace blimp {
+
+BrowserConnectionHandler::BrowserConnectionHandler()
+    : demultiplexer_(new BlimpMessageDemultiplexer),
+      output_buffer_(new BlimpMessageOutputBuffer),
+      multiplexer_(new BlimpMessageMultiplexer(output_buffer_.get())) {}
+
+BrowserConnectionHandler::~BrowserConnectionHandler() {}
+
+scoped_ptr<BlimpMessageProcessor> BrowserConnectionHandler::RegisterFeature(
+    BlimpMessage::Type type,
+    BlimpMessageProcessor* incoming_processor) {
+  demultiplexer_->AddProcessor(type, incoming_processor);
+  return multiplexer_->CreateSenderForType(type);
+}
+
+void BrowserConnectionHandler::HandleConnection(
+    scoped_ptr<BlimpConnection> connection) {
+  // Since there is only a single Client, assume a newer connection should
+  // replace an existing one.
+  DropCurrentConnection();
+  connection_ = connection.Pass();
+
+  // Connect the incoming & outgoing message streams.
+  connection_->SetIncomingMessageProcessor(demultiplexer_.get());
+  output_buffer_->set_output_processor(
+      connection_->GetOutgoingMessageProcessor());
+}
+
+void BrowserConnectionHandler::DropCurrentConnection() {
+  if (!connection_)
+    return;
+  connection_->SetIncomingMessageProcessor(nullptr);
+  output_buffer_->set_output_processor(nullptr);
+  connection_.reset();
+}
+
+void BrowserConnectionHandler::OnConnectionError(int error) {
+  LOG(WARNING) << "Connection error " << error;
+  DropCurrentConnection();
+}
+
+}  // namespace blimp
diff --git a/blimp/net/browser_connection_handler.h b/blimp/net/browser_connection_handler.h
new file mode 100644
index 0000000..d739e50
--- /dev/null
+++ b/blimp/net/browser_connection_handler.h
@@ -0,0 +1,70 @@
+// Copyright 2015 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 BLIMP_NET_BROWSER_CONNECTION_HANDLER_H_
+#define BLIMP_NET_BROWSER_CONNECTION_HANDLER_H_
+
+#include "base/macros.h"
+#include "base/memory/scoped_ptr.h"
+#include "blimp/common/proto/blimp_message.pb.h"
+#include "blimp/net/blimp_net_export.h"
+#include "blimp/net/connection_error_observer.h"
+#include "blimp/net/connection_handler.h"
+
+namespace blimp {
+
+class BlimpConnection;
+class BlimpMessageDemultiplexer;
+class BlimpMessageMultiplexer;
+class BlimpMessageOutputBuffer;
+class BlimpMessageProcessor;
+
+// Routes incoming messages to their respective features, and buffers and sends
+// messages out via underlying BlimpConnection.
+// A BrowserConnectionHandler is created on browser startup, and persists for
+// the lifetime of the application.
+class BLIMP_NET_EXPORT BrowserConnectionHandler
+    : public ConnectionHandler,
+      public ConnectionErrorObserver {
+  BrowserConnectionHandler();
+  ~BrowserConnectionHandler() override;
+
+  // Registers a message processor which will receive all messages of the |type|
+  // specified. Only one handler may be added per type.
+  // That caller must ensure |incoming_processor| remains valid while
+  // this object is in-use.
+  //
+  // Returns a BlimpMessageProcessor object for sending messages of type |type|.
+  scoped_ptr<BlimpMessageProcessor> RegisterFeature(
+      BlimpMessage::Type type,
+      BlimpMessageProcessor* incoming_processor);
+
+  // ConnectionHandler implementation.
+  void HandleConnection(scoped_ptr<BlimpConnection> connection) override;
+
+  // ConnectionErrorObserver implementation.
+  void OnConnectionError(int error) override;
+
+ private:
+  void DropCurrentConnection();
+
+  // Holds network resources while there is a Client connected.
+  scoped_ptr<BlimpConnection> connection_;
+
+  // Routes incoming messages to the relevant feature-specific handlers.
+  scoped_ptr<BlimpMessageDemultiplexer> demultiplexer_;
+
+  // Provides buffering of outgoing messages, for use in session-recovery.
+  scoped_ptr<BlimpMessageOutputBuffer> output_buffer_;
+
+  // Routes outgoing messages from feature-specific handlers to a single
+  // message stream.
+  scoped_ptr<BlimpMessageMultiplexer> multiplexer_;
+
+  DISALLOW_COPY_AND_ASSIGN(BrowserConnectionHandler);
+};
+
+}  // namespace blimp
+
+#endif  // BLIMP_NET_BROWSER_CONNECTION_HANDLER_H_