Refactor MixerSocket with an abstract interface to facilitate unit tests

Bug: b/257993375
Test: cast_media_unittests
Change-Id: Ia89f3781733ff55bc38a3f22762b7b9ff2a83f27
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/4072004
Reviewed-by: Kenneth MacKay <kmackay@chromium.org>
Commit-Queue: Jenny Wong <jyw@google.com>
Cr-Commit-Position: refs/heads/main@{#1079412}

chromecast/media/audio/mixer_service/BUILD.gn

@@ -146,3 +146,21 @@ cast_source_set("redirected_audio_connection") {
     "//net",
   ]
 }
+
+cast_source_set("test_support") {
+  testonly = true
+
+  sources = [
+    "mock_mixer_socket.cc",
+    "mock_mixer_socket.h",
+  ]
+
+  deps = [
+    ":common",
+    "//base",
+    "//chromecast/net:io_buffer_pool",
+    "//net",
+    "//testing/gmock",
+    "//testing/gtest",
+  ]
+}

chromecast/media/audio/mixer_service/mixer_connection.cc

@@ -80,7 +80,8 @@ void MixerConnection::ConnectCallback(int result) {
     LOG_IF(INFO, !log_timeout_) << "Now connected to mixer service";
     log_connection_failure_ = true;
     log_timeout_ = true;
-    auto socket = std::make_unique<MixerSocket>(std::move(connecting_socket_));
+    auto socket =
+        std::make_unique<MixerSocketImpl>(std::move(connecting_socket_));
     OnConnected(std::move(socket));
     return;
   }

chromecast/media/audio/mixer_service/mixer_socket.cc

@@ -7,32 +7,81 @@
 #include <utility>
 
 #include "base/logging.h"
+#include "base/task/sequenced_task_runner.h"
 #include "chromecast/media/audio/mixer_service/mixer_service_transport.pb.h"
+#include "chromecast/net/io_buffer_pool.h"
+#include "net/base/io_buffer.h"
 #include "net/socket/stream_socket.h"
 
 namespace chromecast {
 namespace media {
 namespace mixer_service {
 
-bool MixerSocket::Delegate::HandleMetadata(const Generic& message) {
+bool MixerSocketImpl::Delegate::HandleMetadata(const Generic& message) {
   return true;
 }
 
-MixerSocket::MixerSocket(std::unique_ptr<net::StreamSocket> socket)
+MixerSocketImpl::MixerSocketImpl(std::unique_ptr<net::StreamSocket> socket)
+    : audio_socket_(std::make_unique<AudioSocketExtension>(std::move(socket))) {
+}
+
+MixerSocketImpl::MixerSocketImpl()
+    : audio_socket_(std::make_unique<AudioSocketExtension>()) {}
+
+MixerSocketImpl::~MixerSocketImpl() = default;
+
+void MixerSocketImpl::SetLocalCounterpart(
+    base::WeakPtr<AudioSocket> local_counterpart,
+    scoped_refptr<base::SequencedTaskRunner> counterpart_task_runner) {
+  audio_socket_->SetLocalCounterpart(std::move(local_counterpart),
+                                     std::move(counterpart_task_runner));
+}
+
+base::WeakPtr<AudioSocket> MixerSocketImpl::GetAudioSocketWeakPtr() {
+  return audio_socket_->GetWeakPtr();
+}
+
+void MixerSocketImpl::SetDelegate(Delegate* delegate) {
+  audio_socket_->SetDelegate(delegate);
+}
+
+void MixerSocketImpl::UseBufferPool(scoped_refptr<IOBufferPool> buffer_pool) {
+  audio_socket_->UseBufferPool(std::move(buffer_pool));
+}
+
+MixerSocketImpl::AudioSocketExtension::AudioSocketExtension(
+    std::unique_ptr<net::StreamSocket> socket)
     : AudioSocket(std::move(socket)) {}
 
-MixerSocket::MixerSocket() = default;
+MixerSocketImpl::AudioSocketExtension::AudioSocketExtension() = default;
 
-MixerSocket::~MixerSocket() = default;
+bool MixerSocketImpl::SendAudioBuffer(scoped_refptr<net::IOBuffer> audio_buffer,
+                                      int filled_bytes,
+                                      int64_t timestamp) {
+  return audio_socket_->SendAudioBuffer(std::move(audio_buffer), filled_bytes,
+                                        timestamp);
+}
 
-void MixerSocket::SetDelegate(Delegate* delegate) {
+bool MixerSocketImpl::SendProto(int type,
+                                const google::protobuf::MessageLite& message) {
+  return audio_socket_->SendProto(type, message);
+}
+
+void MixerSocketImpl::ReceiveMoreMessages() {
+  audio_socket_->ReceiveMoreMessages();
+}
+
+MixerSocketImpl::AudioSocketExtension::~AudioSocketExtension() = default;
+
+void MixerSocketImpl::AudioSocketExtension::SetDelegate(
+    MixerSocket::Delegate* delegate) {
   DCHECK(delegate);
-
   delegate_ = delegate;
   AudioSocket::SetDelegate(delegate);
 }
 
-bool MixerSocket::ParseMetadata(char* data, size_t size) {
+bool MixerSocketImpl::AudioSocketExtension::ParseMetadata(char* data,
+                                                          size_t size) {
   Generic message;
   if (!message.ParseFromArray(data, size)) {
     LOG(INFO) << "Invalid metadata message from " << this;

chromecast/media/audio/mixer_service/mixer_socket.h

@@ -7,6 +7,8 @@
 
 #include <cstdint>
 
+#include "base/memory/scoped_refptr.h"
+#include "base/memory/weak_ptr.h"
 #include "chromecast/media/audio/net/audio_socket.h"
 
 namespace net {
@@ -14,14 +16,17 @@ class StreamSocket;
 }  // namespace net
 
 namespace chromecast {
+class IOBufferPool;
+
 namespace media {
 namespace mixer_service {
 class Generic;
 
-// AudioSocket implementation for sending and receiving messages to/from the
-// mixer service.
-class MixerSocket : public AudioSocket {
+class MixerSocket {
  public:
+  static constexpr int kAudioMessageHeaderSize =
+      AudioSocket::kAudioMessageHeaderSize;
+
   class Delegate : public AudioSocket::Delegate {
    public:
     // Called when metadata is received from the other side of the connection.
@@ -32,21 +37,92 @@ class MixerSocket : public AudioSocket {
     ~Delegate() override = default;
   };
 
-  explicit MixerSocket(std::unique_ptr<net::StreamSocket> socket);
-  MixerSocket();
-  MixerSocket(const MixerSocket&) = delete;
-  MixerSocket& operator=(const MixerSocket&) = delete;
-  ~MixerSocket() override;
+  virtual ~MixerSocket() = default;
+
+  virtual void SetLocalCounterpart(
+      base::WeakPtr<AudioSocket> local_counterpart,
+      scoped_refptr<base::SequencedTaskRunner> counterpart_task_runner) = 0;
+  virtual base::WeakPtr<AudioSocket> GetAudioSocketWeakPtr() = 0;
 
   // Sets/changes the delegate. Must be called immediately after creation
   // (ie, synchronously on the same sequence).
-  void SetDelegate(Delegate* delegate);
+  virtual void SetDelegate(Delegate* delegate) = 0;
+
+  // Adds a |buffer_pool| used to allocate buffers to receive messages into,
+  // and for sending protos. If the pool-allocated buffers are too small for a
+  // given message, a normal IOBuffer will be dynamically allocated instead.
+  virtual void UseBufferPool(scoped_refptr<IOBufferPool> buffer_pool) = 0;
+
+  // Prepares |audio_buffer| and then sends it across the connection. Returns
+  // |false| if the audio could not be sent.
+  virtual bool SendAudioBuffer(scoped_refptr<net::IOBuffer> audio_buffer,
+                               int filled_bytes,
+                               int64_t timestamp) = 0;
+
+  // Sends an arbitrary protobuf across the connection. |type| indicates the
+  // type of message; if the write cannot complete immediately, one message of
+  // each type will be stored for later sending; if a newer message is sent with
+  // the same type, then the previous message is overwritten. When writes become
+  // available again, the stored messages are written in order of |type| (lowest
+  // type first). Note that |type| is completely determined by the caller, and
+  // you can reuse the same type value for different messages as long as they
+  // are on different socket instances. A type of 0 means to never store the
+  // message. Returns |false| if the message was not sent or stored.
+  virtual bool SendProto(int type,
+                         const google::protobuf::MessageLite& message) = 0;
+
+  // Resumes receiving messages. Delegate calls may be called synchronously
+  // from within this method.
+  virtual void ReceiveMoreMessages() = 0;
+};
+
+// AudioSocket implementation for sending and receiving messages to/from the
+// mixer service.
+class MixerSocketImpl : public MixerSocket {
+ public:
+  explicit MixerSocketImpl(std::unique_ptr<net::StreamSocket> socket);
+  MixerSocketImpl(const MixerSocketImpl&) = delete;
+  MixerSocketImpl& operator=(const MixerSocketImpl&) = delete;
+  ~MixerSocketImpl() override;
+
+  // Used to create local (in-process) connections.
+  MixerSocketImpl();
+  void SetLocalCounterpart(base::WeakPtr<AudioSocket> local_counterpart,
+                           scoped_refptr<base::SequencedTaskRunner>
+                               counterpart_task_runner) override;
+  base::WeakPtr<AudioSocket> GetAudioSocketWeakPtr() override;
+
+  void SetDelegate(Delegate* delegate) override;
+
+  void UseBufferPool(scoped_refptr<IOBufferPool> buffer_pool) override;
+
+  bool SendAudioBuffer(scoped_refptr<net::IOBuffer> audio_buffer,
+                       int filled_bytes,
+                       int64_t timestamp) override;
+  bool SendProto(int type,
+                 const google::protobuf::MessageLite& message) override;
+  void ReceiveMoreMessages() override;
 
  private:
-  // AudioSocket implementation:
-  bool ParseMetadata(char* data, size_t size) override;
+  class AudioSocketExtension : public AudioSocket {
+   public:
+    explicit AudioSocketExtension(std::unique_ptr<net::StreamSocket> socket);
+    AudioSocketExtension(const MixerSocketImpl&) = delete;
+    AudioSocketExtension& operator=(const MixerSocketImpl&) = delete;
+    ~AudioSocketExtension() override;
 
-  Delegate* delegate_ = nullptr;
+    // Used to create local (in-process) connections.
+    AudioSocketExtension();
+
+    void SetDelegate(MixerSocket::Delegate* delegate);
+
+   private:
+    bool ParseMetadata(char* data, size_t size) override;
+
+    MixerSocket::Delegate* delegate_ = nullptr;
+  };
+
+  std::unique_ptr<AudioSocketExtension> audio_socket_ = nullptr;
 };
 
 }  // namespace mixer_service

chromecast/media/audio/mixer_service/mock_mixer_socket.cc

@@ -0,0 +1,16 @@
+// Copyright 2022 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chromecast/media/audio/mixer_service/mock_mixer_socket.h"
+
+namespace chromecast {
+namespace media {
+namespace mixer_service {
+
+MockMixerSocket::MockMixerSocket() = default;
+MockMixerSocket::~MockMixerSocket() = default;
+
+}  // namespace mixer_service
+}  // namespace media
+}  // namespace chromecast

chromecast/media/audio/mixer_service/mock_mixer_socket.h

@@ -0,0 +1,50 @@
+// Copyright 2022 The Chromium Authors
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_MOCK_MIXER_SOCKET_H_
+#define CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_MOCK_MIXER_SOCKET_H_
+#include "chromecast/media/audio/mixer_service/mixer_socket.h"
+
+#include "base/task/sequenced_task_runner.h"
+#include "chromecast/net/io_buffer_pool.h"
+#include "net/base/io_buffer.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace chromecast {
+namespace media {
+namespace mixer_service {
+
+class MockMixerSocket : public MixerSocket {
+ public:
+  MockMixerSocket();
+  ~MockMixerSocket();
+
+  MOCK_METHOD(void,
+              SetLocalCounterpart,
+              (base::WeakPtr<AudioSocket>,
+               scoped_refptr<base::SequencedTaskRunner>),
+              (override));
+  MOCK_METHOD(base::WeakPtr<AudioSocket>,
+              GetAudioSocketWeakPtr,
+              (),
+              (override));
+  MOCK_METHOD(void, SetDelegate, (MixerSocket::Delegate*), (override));
+  MOCK_METHOD(void, UseBufferPool, (scoped_refptr<IOBufferPool>), (override));
+  MOCK_METHOD(bool,
+              SendAudioBuffer,
+              (scoped_refptr<net::IOBuffer>, int, int64_t),
+              (override));
+  MOCK_METHOD(bool,
+              SendProto,
+              (int, const google::protobuf::MessageLite& message),
+              (override));
+  MOCK_METHOD(void, ReceiveMoreMessages, (), (override));
+};
+
+}  // namespace mixer_service
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_MOCK_MIXER_SOCKET_H_

chromecast/media/audio/mixer_service/receiver/receiver.cc

@@ -143,13 +143,13 @@ Receiver::~Receiver() {
 }
 
 std::unique_ptr<MixerSocket> Receiver::LocalConnect() {
-  auto receiver_socket = std::make_unique<MixerSocket>();
-  auto caller_socket = std::make_unique<MixerSocket>();
+  auto receiver_socket = std::make_unique<MixerSocketImpl>();
+  auto caller_socket = std::make_unique<MixerSocketImpl>();
 
   receiver_socket->SetLocalCounterpart(
-      caller_socket->GetWeakPtr(),
+      caller_socket->GetAudioSocketWeakPtr(),
       base::SequencedTaskRunner::GetCurrentDefault());
-  caller_socket->SetLocalCounterpart(receiver_socket->GetWeakPtr(),
+  caller_socket->SetLocalCounterpart(receiver_socket->GetAudioSocketWeakPtr(),
                                      task_runner_);
 
   task_runner_->PostTask(
@@ -162,7 +162,7 @@ std::unique_ptr<MixerSocket> Receiver::LocalConnect() {
 
 void Receiver::HandleAcceptedSocket(std::unique_ptr<net::StreamSocket> socket) {
   AddInitialSocket(std::make_unique<InitialSocket>(
-      this, std::make_unique<MixerSocket>(std::move(socket))));
+      this, std::make_unique<MixerSocketImpl>(std::move(socket))));
 }
 
 void Receiver::HandleLocalConnection(std::unique_ptr<MixerSocket> socket) {

chromecast/media/cma/backend/mixer/loopback_test_support.cc

@@ -25,14 +25,14 @@ class FakeMixerDelegate : public mixer_service::MixerSocket::Delegate {
 
 std::unique_ptr<mixer_service::MixerSocket> CreateLoopbackConnectionForTest(
     LoopbackHandler* loopback_handler) {
-  auto receiver_socket = std::make_unique<mixer_service::MixerSocket>();
-  auto caller_socket = std::make_unique<mixer_service::MixerSocket>();
+  auto receiver_socket = std::make_unique<mixer_service::MixerSocketImpl>();
+  auto caller_socket = std::make_unique<mixer_service::MixerSocketImpl>();
 
   receiver_socket->SetLocalCounterpart(
-      caller_socket->GetWeakPtr(),
+      caller_socket->GetAudioSocketWeakPtr(),
       base::SequencedTaskRunner::GetCurrentDefault());
   caller_socket->SetLocalCounterpart(
-      receiver_socket->GetWeakPtr(),
+      receiver_socket->GetAudioSocketWeakPtr(),
       base::SequencedTaskRunner::GetCurrentDefault());
 
   auto mixer_side =