Add implementation and unit tests for BlimpMessageMultiplexer.

This creates MessageProcessors that receive outgoing messages
and multiplex them on to a single output message processor.

R=wez@chromium.org,haibinlu@chromium.org
BUG=554179

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

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

4 files changed, 216 insertions, 0 deletions

diff --git a/blimp/net/BUILD.gn b/blimp/net/BUILD.gn
index 8c7edb4..4635fe7 100644
--- a/blimp/net/BUILD.gn
+++ b/blimp/net/BUILD.gn
@@ -9,6 +9,8 @@ component("blimp_net") {
     "blimp_message_checkpoint_observer.h",
     "blimp_message_demultiplexer.cc",
     "blimp_message_demultiplexer.h",
+    "blimp_message_multiplexer.cc",
+    "blimp_message_multiplexer.h",
     "blimp_message_output_buffer.cc",
     "blimp_message_output_buffer.h",
     "blimp_message_processor.h",
@@ -48,6 +50,7 @@ source_set("unit_tests") {
 
   sources = [
     "blimp_message_demultiplexer_unittest.cc",
+    "blimp_message_multiplexer_unittest.cc",
     "stream_packet_reader_unittest.cc",
     "stream_packet_writer_unittest.cc",
     "tcp_transport_unittest.cc",
diff --git a/blimp/net/blimp_message_multiplexer.cc b/blimp/net/blimp_message_multiplexer.cc
new file mode 100644
index 0000000..644d747
--- /dev/null
+++ b/blimp/net/blimp_message_multiplexer.cc
@@ -0,0 +1,63 @@
+// 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/blimp_message_multiplexer.h"
+
+#include "base/logging.h"
+#include "blimp/common/proto/blimp_message.pb.h"
+#include "blimp/net/blimp_message_processor.h"
+
+namespace blimp {
+namespace {
+
+class MultiplexedSender : public BlimpMessageProcessor {
+ public:
+  MultiplexedSender(base::WeakPtr<BlimpMessageProcessor> output_processor,
+                    BlimpMessage::Type type);
+  ~MultiplexedSender() override;
+
+  // BlimpMessageProcessor implementation.
+  // |message.type|, if set, must match the sender's type.
+  void ProcessMessage(scoped_ptr<BlimpMessage> message,
+                      const net::CompletionCallback& callback) override;
+
+ private:
+  base::WeakPtr<BlimpMessageProcessor> output_processor_;
+  BlimpMessage::Type type_;
+
+  DISALLOW_COPY_AND_ASSIGN(MultiplexedSender);
+};
+
+MultiplexedSender::MultiplexedSender(
+    base::WeakPtr<BlimpMessageProcessor> output_processor,
+    BlimpMessage::Type type)
+    : output_processor_(output_processor), type_(type) {}
+
+MultiplexedSender::~MultiplexedSender() {}
+
+void MultiplexedSender::ProcessMessage(
+    scoped_ptr<BlimpMessage> message,
+    const net::CompletionCallback& callback) {
+  if (message->has_type()) {
+    DCHECK_EQ(type_, message->type());
+  } else {
+    message->set_type(type_);
+  }
+  output_processor_->ProcessMessage(message.Pass(), callback);
+}
+
+}  // namespace
+
+BlimpMessageMultiplexer::BlimpMessageMultiplexer(
+    BlimpMessageProcessor* output_processor)
+    : output_weak_factory_(output_processor) {}
+
+BlimpMessageMultiplexer::~BlimpMessageMultiplexer() {}
+
+scoped_ptr<BlimpMessageProcessor> BlimpMessageMultiplexer::CreateSenderForType(
+    BlimpMessage::Type type) {
+  return make_scoped_ptr(
+      new MultiplexedSender(output_weak_factory_.GetWeakPtr(), type));
+}
+}  // namespace blimp
diff --git a/blimp/net/blimp_message_multiplexer.h b/blimp/net/blimp_message_multiplexer.h
new file mode 100644
index 0000000..27713af
--- /dev/null
+++ b/blimp/net/blimp_message_multiplexer.h
@@ -0,0 +1,41 @@
+// 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_BLIMP_MESSAGE_MULTIPLEXER_H_
+#define BLIMP_NET_BLIMP_MESSAGE_MULTIPLEXER_H_
+
+#include "base/macros.h"
+#include "base/memory/weak_ptr.h"
+#include "blimp/common/proto/blimp_message.pb.h"
+#include "blimp/net/blimp_net_export.h"
+
+namespace blimp {
+
+class BlimpConnection;
+class BlimpMessageProcessor;
+
+// Creates MessageProcessors that receive outgoing messages and put them
+// onto a multiplexed message stream.
+class BLIMP_NET_EXPORT BlimpMessageMultiplexer {
+ public:
+  // |output_processor|: A pointer to the MessageProcessor that will receive the
+  // multiplexed message stream.
+  explicit BlimpMessageMultiplexer(BlimpMessageProcessor* output_processor);
+
+  ~BlimpMessageMultiplexer();
+
+  // Creates a BlimpMessageProcessor object for sending messages of type |type|.
+  // Any number of senders can be created at a time for a given type.
+  scoped_ptr<BlimpMessageProcessor> CreateSenderForType(
+      BlimpMessage::Type type);
+
+ private:
+  base::WeakPtrFactory<BlimpMessageProcessor> output_weak_factory_;
+
+  DISALLOW_COPY_AND_ASSIGN(BlimpMessageMultiplexer);
+};
+
+}  // namespace blimp
+
+#endif  // BLIMP_NET_BLIMP_MESSAGE_MULTIPLEXER_H_
diff --git a/blimp/net/blimp_message_multiplexer_unittest.cc b/blimp/net/blimp_message_multiplexer_unittest.cc
new file mode 100644
index 0000000..4c757b2
--- /dev/null
+++ b/blimp/net/blimp_message_multiplexer_unittest.cc
@@ -0,0 +1,109 @@
+// 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/common/proto/blimp_message.pb.h"
+#include "blimp/common/proto/input.pb.h"
+#include "blimp/common/proto/navigation.pb.h"
+#include "blimp/net/blimp_message_multiplexer.h"
+#include "blimp/net/test_common.h"
+#include "net/base/net_errors.h"
+#include "net/base/test_completion_callback.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+using testing::_;
+using testing::DoAll;
+using testing::Ref;
+using testing::SaveArg;
+
+namespace blimp {
+namespace {
+
+class BlimpMessageMultiplexerTest : public testing::Test {
+ public:
+  BlimpMessageMultiplexerTest()
+      : multiplexer_(&mock_output_processor_),
+        input_message_(new BlimpMessage),
+        navigation_message_(new BlimpMessage),
+        input_processor_(multiplexer_.CreateSenderForType(BlimpMessage::INPUT)),
+        navigation_processor_(
+            multiplexer_.CreateSenderForType(BlimpMessage::NAVIGATION)) {}
+
+  void SetUp() override {
+    EXPECT_CALL(mock_output_processor_, MockableProcessMessage(_, _))
+        .WillRepeatedly(
+            DoAll(SaveArg<0>(&captured_message_), SaveArg<1>(&captured_cb_)));
+
+    input_message_->mutable_input()->set_type(InputMessage::DRAG);
+    navigation_message_->mutable_navigation()->set_type(
+        NavigationMessage::LOAD_URL);
+  }
+
+ protected:
+  MockBlimpMessageProcessor mock_output_processor_;
+  BlimpMessageMultiplexer multiplexer_;
+  scoped_ptr<BlimpMessage> input_message_;
+  scoped_ptr<BlimpMessage> navigation_message_;
+  BlimpMessage captured_message_;
+  net::CompletionCallback captured_cb_;
+  scoped_ptr<BlimpMessageProcessor> input_processor_;
+  scoped_ptr<BlimpMessageProcessor> navigation_processor_;
+};
+
+// Verify that each sender propagates its types and copies the message payload
+// correctly.
+TEST_F(BlimpMessageMultiplexerTest, TypeSetByMux) {
+  net::TestCompletionCallback cb_1;
+  input_processor_->ProcessMessage(input_message_.Pass(), cb_1.callback());
+  EXPECT_EQ(BlimpMessage::INPUT, captured_message_.type());
+  EXPECT_EQ(InputMessage::DRAG, captured_message_.input().type());
+  captured_cb_.Run(net::OK);
+  EXPECT_EQ(net::OK, cb_1.WaitForResult());
+
+  net::TestCompletionCallback cb_2;
+  navigation_processor_->ProcessMessage(navigation_message_.Pass(),
+                                        cb_2.callback());
+  EXPECT_EQ(BlimpMessage::NAVIGATION, captured_message_.type());
+  EXPECT_EQ(NavigationMessage::LOAD_URL, captured_message_.navigation().type());
+  captured_cb_.Run(net::ERR_FAILED);
+  EXPECT_EQ(net::ERR_FAILED, cb_2.WaitForResult());
+}
+
+// Verify that the multiplexer allows the caller to supply a message type.
+TEST_F(BlimpMessageMultiplexerTest, TypeSetByCaller) {
+  input_message_->set_type(BlimpMessage::INPUT);
+
+  net::TestCompletionCallback cb_1;
+  input_processor_->ProcessMessage(input_message_.Pass(), cb_1.callback());
+  EXPECT_EQ(BlimpMessage::INPUT, captured_message_.type());
+  EXPECT_EQ(InputMessage::DRAG, captured_message_.input().type());
+  captured_cb_.Run(net::OK);
+  EXPECT_EQ(net::OK, cb_1.WaitForResult());
+}
+
+// Verify that senders for a given type can be torn down and recreated.
+TEST_F(BlimpMessageMultiplexerTest, SenderTransience) {
+  net::TestCompletionCallback cb_3;
+  input_processor_ = multiplexer_.CreateSenderForType(BlimpMessage::INPUT);
+  input_processor_->ProcessMessage(input_message_.Pass(), cb_3.callback());
+  EXPECT_EQ(BlimpMessage::INPUT, captured_message_.type());
+  EXPECT_EQ(InputMessage::DRAG, captured_message_.input().type());
+  captured_cb_.Run(net::OK);
+  EXPECT_EQ(net::OK, cb_3.WaitForResult());
+}
+
+// Verify that there is no limit on the number of senders for a given type.
+TEST_F(BlimpMessageMultiplexerTest, SenderMultiplicity) {
+  net::TestCompletionCallback cb_4;
+  scoped_ptr<BlimpMessageProcessor> input_processor_2 =
+      multiplexer_.CreateSenderForType(BlimpMessage::INPUT);
+  input_processor_2->ProcessMessage(input_message_.Pass(), cb_4.callback());
+  EXPECT_EQ(BlimpMessage::INPUT, captured_message_.type());
+  EXPECT_EQ(InputMessage::DRAG, captured_message_.input().type());
+  captured_cb_.Run(net::ERR_INVALID_ARGUMENT);
+  EXPECT_EQ(net::ERR_INVALID_ARGUMENT, cb_4.WaitForResult());
+}
+
+}  // namespace
+}  // namespace blimp