[Chromecast] Create control connection for mixer

This allows other processes to directly control the mixer, eg volume
control or postprocessor config changes.

Bug: internal b/127963522
Change-Id: Ibd2dc3b320ae633c9b7ddc9b51889bff34d7f62e
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/1842272Reviewed-by: Yuchen Liu <yucliu@chromium.org>
Commit-Queue: Kenneth MacKay <kmackay@chromium.org>
Cr-Commit-Position: refs/heads/master@{#704873}

chromecast/browser/cast_content_browser_client.cc

@@ -13,6 +13,7 @@
 #include "base/command_line.h"
 #include "base/files/scoped_file.h"
 #include "base/i18n/rtl.h"
+#include "base/message_loop/message_pump_type.h"
 #include "base/path_service.h"
 #include "base/stl_util.h"
 #include "base/strings/string_number_conversions.h"
@@ -265,6 +266,8 @@ CastContentBrowserClient::GetMediaTaskRunner() {
   if (!media_thread_) {
     media_thread_.reset(new base::Thread("CastMediaThread"));
     base::Thread::Options options;
+    // We need the media thread to be IO-capable to use the mixer service.
+    options.message_pump_type = base::MessagePumpType::IO;
     options.priority = base::ThreadPriority::REALTIME_AUDIO;
     CHECK(media_thread_->StartWithOptions(options));
     // Start the media_resource_tracker as soon as the media thread is created.

chromecast/media/audio/mixer_service/BUILD.gn

@@ -68,6 +68,22 @@ cast_source_set("output_stream_connection") {
   ]
 }
 
+cast_source_set("control_connection") {
+  sources = [
+    "control_connection.cc",
+    "control_connection.h",
+  ]
+
+  deps = [
+    ":common",
+    ":connection",
+    ":proto",
+    "//base",
+    "//chromecast/public",
+    "//net",
+  ]
+}
+
 cast_source_set("audio_socket_service") {
   sources = [
     "audio_socket_service.cc",

chromecast/media/audio/mixer_service/control_connection.cc

+// Copyright 2019 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 "chromecast/media/audio/mixer_service/control_connection.h"
+
+#include <utility>
+
+#include "base/logging.h"
+#include "chromecast/media/audio/mixer_service/conversions.h"
+#include "chromecast/media/audio/mixer_service/mixer_service.pb.h"
+#include "net/socket/stream_socket.h"
+
+namespace chromecast {
+namespace media {
+namespace mixer_service {
+
+ControlConnection::ControlConnection() = default;
+
+ControlConnection::~ControlConnection() = default;
+
+void ControlConnection::Connect(ConnectedCallback callback) {
+  connect_callback_ = std::move(callback);
+  MixerConnection::Connect();
+}
+
+void ControlConnection::SetVolume(AudioContentType type,
+                                  float volume_multiplier) {
+  if (type == AudioContentType::kOther) {
+    return;
+  }
+
+  volume_[type] = volume_multiplier;
+  if (socket_) {
+    Generic message;
+    auto* volume = message.mutable_set_device_volume();
+    volume->set_content_type(ConvertContentType(type));
+    volume->set_volume_multiplier(volume_multiplier);
+    socket_->SendProto(message);
+  }
+}
+
+void ControlConnection::SetMuted(AudioContentType type, bool muted) {
+  if (type == AudioContentType::kOther) {
+    return;
+  }
+
+  muted_[type] = muted;
+  if (socket_) {
+    Generic message;
+    auto* muted = message.mutable_set_device_muted();
+    muted->set_content_type(ConvertContentType(type));
+    muted->set_muted(muted);
+    socket_->SendProto(message);
+  }
+}
+
+void ControlConnection::SetVolumeLimit(AudioContentType type,
+                                       float max_volume_multiplier) {
+  if (type == AudioContentType::kOther) {
+    return;
+  }
+
+  volume_limit_[type] = max_volume_multiplier;
+  if (socket_) {
+    Generic message;
+    auto* limit = message.mutable_set_volume_limit();
+    limit->set_content_type(ConvertContentType(type));
+    limit->set_max_volume_multiplier(max_volume_multiplier);
+    socket_->SendProto(message);
+  }
+}
+
+void ControlConnection::ConfigurePostprocessor(const std::string& name,
+                                               const void* config,
+                                               int size_bytes) {
+  if (!socket_) {
+    return;
+  }
+  Generic message;
+  auto* content = message.mutable_configure_postprocessor();
+  content->set_name(name);
+  content->set_config(static_cast<const char*>(config), size_bytes);
+  socket_->SendProto(message);
+}
+
+void ControlConnection::ReloadPostprocessors() {
+  if (!socket_) {
+    return;
+  }
+  Generic message;
+  message.mutable_reload_postprocessors();
+  socket_->SendProto(message);
+}
+
+void ControlConnection::SetStreamCountCallback(StreamCountCallback callback) {
+  stream_count_callback_ = std::move(callback);
+  if (socket_) {
+    Generic message;
+    message.mutable_request_stream_count()->set_subscribe(!callback.is_null());
+    socket_->SendProto(message);
+  }
+}
+
+void ControlConnection::OnConnected(std::unique_ptr<net::StreamSocket> socket) {
+  socket_ = std::make_unique<MixerSocket>(std::move(socket), this);
+  socket_->ReceiveMessages();
+
+  for (const auto& item : volume_limit_) {
+    Generic message;
+    auto* limit = message.mutable_set_volume_limit();
+    limit->set_content_type(ConvertContentType(item.first));
+    limit->set_max_volume_multiplier(item.second);
+    socket_->SendProto(message);
+  }
+
+  for (const auto& item : muted_) {
+    Generic message;
+    auto* muted = message.mutable_set_device_muted();
+    muted->set_content_type(ConvertContentType(item.first));
+    muted->set_muted(item.second);
+    socket_->SendProto(message);
+  }
+
+  for (const auto& item : volume_) {
+    Generic message;
+    auto* volume = message.mutable_set_device_volume();
+    volume->set_content_type(ConvertContentType(item.first));
+    volume->set_volume_multiplier(item.second);
+    socket_->SendProto(message);
+  }
+
+  if (stream_count_callback_) {
+    Generic message;
+    message.mutable_request_stream_count()->set_subscribe(true);
+    socket_->SendProto(message);
+  }
+
+  if (connect_callback_) {
+    connect_callback_.Run();
+  }
+}
+
+void ControlConnection::OnConnectionError() {
+  socket_.reset();
+  MixerConnection::Connect();
+}
+
+bool ControlConnection::HandleMetadata(const Generic& message) {
+  if (stream_count_callback_ && message.has_stream_count()) {
+    stream_count_callback_.Run(message.stream_count().primary(),
+                               message.stream_count().sfx());
+  }
+  return true;
+}
+
+}  // namespace mixer_service
+}  // namespace media
+}  // namespace chromecast

chromecast/media/audio/mixer_service/control_connection.h

+// Copyright 2019 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 CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_CONTROL_CONNECTION_H_
+#define CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_CONTROL_CONNECTION_H_
+
+#include <memory>
+#include <string>
+
+#include "base/callback.h"
+#include "base/containers/flat_map.h"
+#include "base/macros.h"
+#include "chromecast/media/audio/mixer_service/mixer_connection.h"
+#include "chromecast/media/audio/mixer_service/mixer_socket.h"
+#include "chromecast/public/volume_control.h"
+
+namespace net {
+class StreamSocket;
+}  // namespace net
+
+namespace chromecast {
+namespace media {
+namespace mixer_service {
+
+// Mixer service connection for controlling general mixer properties, such as
+// device volume and postprocessor configuration. Not thread-safe; all usage of
+// a given instance must be on the same sequence. Must be created on an IO
+// thread.
+class ControlConnection : public MixerConnection, public MixerSocket::Delegate {
+ public:
+  using ConnectedCallback = base::RepeatingClosure;
+
+  // Callback to receive mixer stream count changes.
+  using StreamCountCallback =
+      base::RepeatingCallback<void(int primary_streams, int sfx_streams)>;
+
+  ControlConnection();
+  ~ControlConnection() override;
+
+  // Connects to the mixer. If the mixer connection is lost, this will
+  // automatically reconnect. If |callback| is nonempty, it will be called each
+  // time a connection is (re)established with the mixer. This can be used to
+  // re-send preprocessor configuration, since it is not persisted across
+  // disconnects.
+  void Connect(ConnectedCallback callback = ConnectedCallback());
+
+  // Sets volume multiplier for all streams of a given content type.
+  void SetVolume(AudioContentType type, float volume_multiplier);
+
+  // Sets mute state all streams of a given content type.
+  void SetMuted(AudioContentType type, bool muted);
+
+  // Sets the maximum effective volume multiplier for a given content type.
+  void SetVolumeLimit(AudioContentType type, float max_volume_multiplier);
+
+  // Sends arbitrary config data to a specific postprocessor. Note that the
+  // config is not persisted across disconnects, and is not saved if
+  // ConfigurePostprocessor() is called when not connected to the mixer, so
+  // use the Connect() callback to determine when to (re)send config, if needed.
+  void ConfigurePostprocessor(const std::string& name,
+                              const void* config,
+                              int size_bytes);
+
+  // Instructs the mixer to reload postprocessors based on the config file.
+  void ReloadPostprocessors();
+
+  // Sets a callback to receive mixer stream count changes. |callback| may be an
+  // empty callback to remove it.
+  void SetStreamCountCallback(StreamCountCallback callback);
+
+ private:
+  // MixerConnection implementation:
+  void OnConnected(std::unique_ptr<net::StreamSocket> socket) override;
+  void OnConnectionError() override;
+
+  // MixerSocket::Delegate implementation:
+  bool HandleMetadata(const Generic& message) override;
+
+  std::unique_ptr<MixerSocket> socket_;
+
+  ConnectedCallback connect_callback_;
+
+  base::flat_map<AudioContentType, float> volume_;
+  base::flat_map<AudioContentType, bool> muted_;
+  base::flat_map<AudioContentType, float> volume_limit_;
+
+  StreamCountCallback stream_count_callback_;
+
+  DISALLOW_COPY_AND_ASSIGN(ControlConnection);
+};
+
+}  // namespace mixer_service
+}  // namespace media
+}  // namespace chromecast
+
+#endif  // CHROMECAST_MEDIA_AUDIO_MIXER_SERVICE_CONTROL_CONNECTION_H_

chromecast/media/audio/mixer_service/mixer_connection.cc

@@ -84,6 +84,7 @@ void MixerConnection::ConnectCallback(int result) {
 
   connection_timeout_.Stop();
   if (result == net::OK) {
+    LOG_IF(INFO, !log_timeout_) << "Now connected to mixer service";
     log_connection_failure_ = true;
     log_timeout_ = true;
     OnConnected(std::move(connecting_socket_));

chromecast/media/audio/mixer_service/mixer_service.proto

@@ -155,6 +155,17 @@ message ConfigurePostprocessor {
 // Instructs the mixer to reload postprocessors based on the config file.
 message ReloadPostprocessors {}
 
+// Asks the mixer to send / stop sending stream count updates.
+message RequestStreamCount {
+  optional bool subscribe = 1;
+}
+
+// Indicates how many output streams are currently being handled by the mixer.
+message StreamCount {
+  optional int32 primary = 1;
+  optional int32 sfx = 2;
+}
+
 // Indicates an error on an audio stream.
 message Error {
   enum Type {
@@ -183,5 +194,7 @@ message Generic {
   optional SetVolumeLimit set_volume_limit = 15;
   optional ConfigurePostprocessor configure_postprocessor = 16;
   optional ReloadPostprocessors reload_postprocessors = 17;
-  optional Error error = 18;
+  optional RequestStreamCount request_stream_count = 18;
+  optional StreamCount stream_count = 19;
+  optional Error error = 20;
 }

chromecast/media/audio/mixer_service/mixer_socket.cc

@@ -11,6 +11,7 @@
 #include "base/big_endian.h"
 #include "base/logging.h"
 #include "chromecast/media/audio/mixer_service/constants.h"
+#include "chromecast/media/audio/mixer_service/mixer_service.pb.h"
 #include "net/base/io_buffer.h"
 #include "net/socket/stream_socket.h"
 

chromecast/media/audio/mixer_service/mixer_socket.h

@@ -11,9 +11,14 @@
 
 #include "base/macros.h"
 #include "base/memory/scoped_refptr.h"
-#include "chromecast/media/audio/mixer_service/mixer_service.pb.h"
 #include "chromecast/net/small_message_socket.h"
 
+namespace google {
+namespace protobuf {
+class MessageLite;
+}  // namespace protobuf
+}  // namespace google
+
 namespace net {
 class IOBuffer;
 class StreamSocket;
@@ -22,6 +27,7 @@ class StreamSocket;
 namespace chromecast {
 namespace media {
 namespace mixer_service {
+class Generic;
 
 // Base class for sending and receiving messages to/from the mixer service.
 // Not thread-safe; all usage of a given instance must be on the same sequence.

chromecast/media/audio/mixer_service/output_stream_connection.cc

@@ -154,7 +154,7 @@ void OutputStreamConnection::OnConnected(
 
 void OutputStreamConnection::OnConnectionError() {
   socket_.reset();
-  Connect();
+  MixerConnection::Connect();
 }
 
 bool OutputStreamConnection::HandleMetadata(const Generic& message) {

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

@@ -61,7 +61,8 @@ class Receiver::InitialSocket : public MixerSocket::Delegate {
                message.has_set_device_muted() ||
                message.has_set_volume_limit() ||
                message.has_configure_postprocessor() ||
-               message.has_reload_postprocessors()) {
+               message.has_reload_postprocessors() ||
+               message.has_request_stream_count()) {
       receiver_->CreateControlConnection(std::move(socket_), message);
       receiver_->RemoveInitialSocket(this);
     }

chromecast/media/cma/backend/BUILD.gn

@@ -43,6 +43,13 @@ cast_source_set("backend") {
     "//chromecast/media/cma/base",
     "//chromecast/media/cma/decoder",
   ]
+
+  if (have_full_mixer) {
+    sources += [ "media_pipeline_backend_manager_mixer.cc" ]
+    deps += [ "//chromecast/media/audio/mixer_service:control_connection" ]
+  } else {
+    sources += [ "media_pipeline_backend_manager_no_mixer.cc" ]
+  }
 }
 
 # Target for OEM partners to override media shared library, i.e.
@@ -209,6 +216,7 @@ cast_source_set("for_mixer_audio") {
     ":volume_map",
     "//base",
     "//chromecast/base",
+    "//chromecast/media/audio/mixer_service:control_connection",
     "//chromecast/media/base:monotonic_clock",
     "//chromecast/media/cma/backend/mixer",
     "//chromecast/media/cma/base",

chromecast/media/cma/backend/media_pipeline_backend_manager.cc

@@ -58,13 +58,15 @@ MediaPipelineBackendManager::MediaPipelineBackendManager(
       buffer_delegate_(nullptr),
       weak_factory_(this) {
   DCHECK(media_task_runner_);
-  DCHECK(playing_audio_streams_count_.size() ==
-         static_cast<unsigned long>(AudioContentType::kNumTypes));
-  DCHECK(playing_noneffects_audio_streams_count_.size() ==
-         static_cast<unsigned long>(AudioContentType::kNumTypes));
+  DCHECK_EQ(playing_audio_streams_count_.size(),
+            static_cast<size_t>(AudioContentType::kNumTypes));
+  DCHECK_EQ(playing_noneffects_audio_streams_count_.size(),
+            static_cast<size_t>(AudioContentType::kNumTypes));
   for (int i = 0; i < NUM_DECODER_TYPES; ++i) {
     decoder_count_[i] = 0;
   }
+
+  RUN_ON_MEDIA_THREAD(CreateMixerConnection);
 }
 
 MediaPipelineBackendManager::~MediaPipelineBackendManager() {
@@ -124,12 +126,43 @@ void MediaPipelineBackendManager::UpdatePlayingAudioCount(
     bool sfx,
     const AudioContentType type,
     int change) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   DCHECK(change == -1 || change == 1) << "bad count change: " << change;
 
   bool had_playing_audio_streams = (TotalPlayingAudioStreamsCount() > 0);
+  // Volume feedback sounds are only allowed when there are no non-effects
+  // audio streams playing.
+  bool prev_allow_feedback = (TotalPlayingNoneffectsAudioStreamsCount() == 0);
+
   playing_audio_streams_count_[type] += change;
   DCHECK_GE(playing_audio_streams_count_[type], 0);
 
+  if (!sfx) {
+    playing_noneffects_audio_streams_count_[type] += change;
+    DCHECK_GE(playing_noneffects_audio_streams_count_[type], 0);
+  }
+
+  HandlePlayingAudioStreamsChange(had_playing_audio_streams,
+                                  prev_allow_feedback);
+}
+
+void MediaPipelineBackendManager::OnMixerStreamCountChange(int primary_streams,
+                                                           int sfx_streams) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+  bool had_playing_audio_streams = (TotalPlayingAudioStreamsCount() > 0);
+  bool prev_allow_feedback = (TotalPlayingNoneffectsAudioStreamsCount() == 0);
+
+  mixer_primary_stream_count_ = primary_streams;
+  mixer_sfx_stream_count_ = sfx_streams;
+
+  HandlePlayingAudioStreamsChange(had_playing_audio_streams,
+                                  prev_allow_feedback);
+}
+
+void MediaPipelineBackendManager::HandlePlayingAudioStreamsChange(
+    bool had_playing_audio_streams,
+    bool prev_allow_feedback) {
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
   int new_playing_audio_streams = TotalPlayingAudioStreamsCount();
   if (new_playing_audio_streams == 0) {
     power_save_timer_.Start(FROM_HERE, kPowerSaveWaitTime, this,
@@ -143,17 +176,7 @@ void MediaPipelineBackendManager::UpdatePlayingAudioCount(
     }
   }
 
-  if (sfx) {
-    return;
-  }
-
-  // Volume feedback sounds are only allowed when there are no non-effects
-  // audio streams playing.
-  bool prev_allow_feedback = (TotalPlayingNoneffectsAudioStreamsCount() == 0);
-  playing_noneffects_audio_streams_count_[type] += change;
-  DCHECK_GE(playing_noneffects_audio_streams_count_[type], 0);
   bool new_allow_feedback = (TotalPlayingNoneffectsAudioStreamsCount() == 0);
-
   if (new_allow_feedback != prev_allow_feedback) {
     allow_volume_feedback_observers_->Notify(
         FROM_HERE, &AllowVolumeFeedbackObserver::AllowVolumeFeedbackSounds,
@@ -166,7 +189,7 @@ int MediaPipelineBackendManager::TotalPlayingAudioStreamsCount() {
   for (auto entry : playing_audio_streams_count_) {
     total += entry.second;
   }
-  return total;
+  return std::max(total, mixer_primary_stream_count_ + mixer_sfx_stream_count_);
 }
 
 int MediaPipelineBackendManager::TotalPlayingNoneffectsAudioStreamsCount() {
@@ -174,7 +197,7 @@ int MediaPipelineBackendManager::TotalPlayingNoneffectsAudioStreamsCount() {
   for (auto entry : playing_noneffects_audio_streams_count_) {
     total += entry.second;
   }
-  return total;
+  return std::max(total, mixer_primary_stream_count_);
 }
 
 void MediaPipelineBackendManager::EnterPowerSaveMode() {
@@ -219,15 +242,6 @@ void MediaPipelineBackendManager::SetBufferDelegate(
   buffer_delegate_ = buffer_delegate;
 }
 
-bool MediaPipelineBackendManager::IsPlaying(bool include_sfx,
-                                            AudioContentType type) {
-  if (include_sfx) {
-    return playing_audio_streams_count_[type];
-  } else {
-    return playing_noneffects_audio_streams_count_[type];
-  }
-}
-
 void MediaPipelineBackendManager::SetPowerSaveEnabled(bool power_save_enabled) {
   MAKE_SURE_MEDIA_THREAD(SetPowerSaveEnabled, power_save_enabled);
   power_save_enabled_ = power_save_enabled;

chromecast/media/cma/backend/media_pipeline_backend_manager.h

@@ -124,8 +124,6 @@ class MediaPipelineBackendManager {
 
   BufferDelegate* buffer_delegate() const { return buffer_delegate_; }
 
-  bool IsPlaying(bool include_sfx, AudioContentType type);
-
   // If |power_save_enabled| is |false|, power save will be turned off and
   // automatic power save will be disabled until this is called with |true|.
   void SetPowerSaveEnabled(bool power_save_enabled);
@@ -133,6 +131,13 @@ class MediaPipelineBackendManager {
  private:
   friend class ActiveMediaPipelineBackendWrapper;
 
+  class MixerConnection {
+   public:
+    virtual ~MixerConnection() = default;
+  };
+
+  void CreateMixerConnection();
+
   // Backend wrapper instances must use these APIs when allocating and releasing
   // decoder objects, so we can enforce global limit on #concurrent decoders.
   bool IncrementDecoderCount(DecoderType type);
@@ -142,6 +147,9 @@ class MediaPipelineBackendManager {
   void UpdatePlayingAudioCount(bool sfx,
                                const AudioContentType type,
                                int change);
+  void OnMixerStreamCountChange(int primary_streams, int sfx_streams);
+  void HandlePlayingAudioStreamsChange(bool had_playing_audio_streams,
+                                       bool prev_allow_feedback);
   int TotalPlayingAudioStreamsCount();
   int TotalPlayingNoneffectsAudioStreamsCount();
 
@@ -170,6 +178,10 @@ class MediaPipelineBackendManager {
 
   base::OneShotTimer power_save_timer_;
 
+  std::unique_ptr<MixerConnection> mixer_connection_;
+  int mixer_primary_stream_count_ = 0;
+  int mixer_sfx_stream_count_ = 0;
+
   base::WeakPtrFactory<MediaPipelineBackendManager> weak_factory_;
 
   DISALLOW_COPY_AND_ASSIGN(MediaPipelineBackendManager);

chromecast/media/cma/backend/media_pipeline_backend_manager_mixer.cc

+// Copyright 2019 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 "chromecast/media/cma/backend/media_pipeline_backend_manager.h"
+
+#include <memory>
+#include <utility>
+
+#include "base/bind.h"
+#include "base/logging.h"
+#include "chromecast/media/audio/mixer_service/control_connection.h"
+
+namespace chromecast {
+namespace media {
+
+void MediaPipelineBackendManager::CreateMixerConnection() {
+  struct RealMixerConnection : public MixerConnection {
+    RealMixerConnection() { connection.Connect(); }
+
+    ~RealMixerConnection() override = default;
+
+    mixer_service::ControlConnection connection;
+  };
+
+  DCHECK(media_task_runner_->BelongsToCurrentThread());
+  auto mixer = std::make_unique<RealMixerConnection>();
+  mixer->connection.SetStreamCountCallback(base::BindRepeating(
+      &MediaPipelineBackendManager::OnMixerStreamCountChange,
+      base::Unretained(this)));
+  mixer_connection_ = std::move(mixer);
+}
+
+}  // namespace media
+}  // namespace chromecast

chromecast/media/cma/backend/media_pipeline_backend_manager_no_mixer.cc

+// Copyright 2019 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 "chromecast/media/cma/backend/media_pipeline_backend_manager.h"
+
+#include "base/logging.h"
+
+namespace chromecast {
+namespace media {
+
+void MediaPipelineBackendManager::CreateMixerConnection() {}
+
+}  // namespace media
+}  // namespace chromecast

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

@@ -240,6 +240,11 @@ int BufferingMixerSource::playout_channel() {
   return playout_channel_;
 }
 
+bool BufferingMixerSource::active() {
+  base::AutoLock lock(lock_);
+  return !paused_;
+}
+
 void BufferingMixerSource::WritePcm(scoped_refptr<DecoderBufferBase> data) {
   DCHECK(caller_task_runner_->BelongsToCurrentThread());
 
@@ -301,13 +306,16 @@ BufferingMixerSource::RenderingDelay BufferingMixerSource::QueueData(
 void BufferingMixerSource::SetPaused(bool paused) {
   LOG(INFO) << (paused ? "Pausing " : "Unpausing ") << device_id_ << " ("
             << this << ")";
-  base::AutoLock lock(lock_);
-  // Clear start timestamp, since a pause should invalidate the start timestamp
-  // anyway. The AV sync code can restart (hard correction) on resume if
-  // needed.
-  playback_start_timestamp_ = INT64_MIN;
-  mixer_rendering_delay_ = RenderingDelay();
-  paused_ = paused;
+  {
+    base::AutoLock lock(lock_);
+    // Clear start timestamp, since a pause should invalidate the start
+    // timestamp anyway. The AV sync code can restart (hard correction) on
+    // resume if needed.
+    playback_start_timestamp_ = INT64_MIN;
+    mixer_rendering_delay_ = RenderingDelay();
+    paused_ = paused;
+  }
+  mixer_->UpdateStreamCounts();
 }
 
 void BufferingMixerSource::SetVolumeMultiplier(float multiplier) {

chromecast/media/cma/backend/mixer/buffering_mixer_source.h

@@ -141,6 +141,7 @@ class BufferingMixerSource : public MixerInput::Source,
   AudioContentType content_type() override;
   int desired_read_size() override;
   int playout_channel() override;
+  bool active() override;
 
   void InitializeAudioPlayback(int read_size,
                                RenderingDelay initial_rendering_delay) override;

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

@@ -102,6 +102,9 @@ int DirectMixerSource::desired_read_size() {
 int DirectMixerSource::playout_channel() {
   return playout_channel_;
 }
+bool DirectMixerSource::active() {
+  return true;
+}
 
 void DirectMixerSource::SetVolumeMultiplier(float multiplier) {
   mixer_->SetVolumeMultiplier(this, multiplier);

chromecast/media/cma/backend/mixer/direct_mixer_source.h

@@ -57,6 +57,7 @@ class DirectMixerSource : public MixerInput::Source,
   AudioContentType content_type() override;
   int desired_read_size() override;
   int playout_channel() override;
+  bool active() override;
 
   void InitializeAudioPlayback(int read_size,
                                RenderingDelay initial_rendering_delay) override;

chromecast/media/cma/backend/mixer/mixer_input.h

@@ -54,6 +54,8 @@ class MixerInput {
     virtual AudioContentType content_type() = 0;
     virtual int desired_read_size() = 0;
     virtual int playout_channel() = 0;
+    // Returns true if the source is currently providing audio to be mixed.
+    virtual bool active() = 0;
 
     // Called when the input has been added to the mixer, before any other
     // calls are made. The |read_size| is the number of frames that will be

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

@@ -391,18 +391,21 @@ void MixerInputConnection::SetPaused(bool paused) {
                             &MixerInputConnection::OnInactivityTimeout);
   }
 
-  base::AutoLock lock(lock_);
-  if (paused == paused_) {
-    return;
-  }
+  {
+    base::AutoLock lock(lock_);
+    if (paused == paused_) {
+      return;
+    }
 
-  paused_ = paused;
-  mixer_rendering_delay_ = RenderingDelay();
-  // Clear start timestamp, since a pause should invalidate the start
-  // timestamp anyway. The AV sync code can restart (hard correction) on
-  // resume if needed.
-  use_start_timestamp_ = false;
-  playback_start_timestamp_ = INT64_MIN;
+    paused_ = paused;
+    mixer_rendering_delay_ = RenderingDelay();
+    // Clear start timestamp, since a pause should invalidate the start
+    // timestamp anyway. The AV sync code can restart (hard correction) on
+    // resume if needed.
+    use_start_timestamp_ = false;
+    playback_start_timestamp_ = INT64_MIN;
+  }
+  mixer_->UpdateStreamCounts();
 }
 
 int MixerInputConnection::num_channels() {
@@ -427,6 +430,11 @@ int MixerInputConnection::playout_channel() {
   return playout_channel_;
 }
 
+bool MixerInputConnection::active() {
+  base::AutoLock lock(lock_);
+  return !paused_;
+}
+
 void MixerInputConnection::WritePcm(scoped_refptr<net::IOBuffer> data) {
   DCHECK(io_task_runner_->RunsTasksInCurrentSequence());
 

chromecast/media/cma/backend/mixer/mixer_input_connection.h

@@ -16,6 +16,7 @@
 #include "base/synchronization/lock.h"
 #include "base/thread_annotations.h"
 #include "base/timer/timer.h"
+#include "chromecast/media/audio/mixer_service/mixer_service.pb.h"
 #include "chromecast/media/audio/mixer_service/mixer_socket.h"
 #include "chromecast/media/cma/backend/audio_fader.h"
 #include "chromecast/media/cma/backend/mixer/mixer_input.h"
@@ -95,6 +96,7 @@ class MixerInputConnection : public mixer_service::MixerSocket::Delegate,
   AudioContentType content_type() override;
   int desired_read_size() override;
   int playout_channel() override;
+  bool active() override;
 
   void InitializeAudioPlayback(int read_size,
                                RenderingDelay initial_rendering_delay) override;

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

@@ -4,22 +4,122 @@
 
 #include "chromecast/media/cma/backend/mixer/mixer_service_receiver.h"
 
+#include <string>
 #include <utility>
 
 #include "base/logging.h"
+#include "chromecast/media/audio/mixer_service/conversions.h"
 #include "chromecast/media/audio/mixer_service/mixer_service.pb.h"
 #include "chromecast/media/audio/mixer_service/mixer_socket.h"
 #include "chromecast/media/cma/backend/mixer/mixer_input_connection.h"
+#include "chromecast/media/cma/backend/mixer/stream_mixer.h"
 
 namespace chromecast {
 namespace media {
 
+class MixerServiceReceiver::ControlConnection
+    : public mixer_service::MixerSocket::Delegate {
+ public:
+  ControlConnection(StreamMixer* mixer,
+                    MixerServiceReceiver* receiver,
+                    std::unique_ptr<mixer_service::MixerSocket> socket)
+      : mixer_(mixer), receiver_(receiver), socket_(std::move(socket)) {
+    DCHECK(mixer_);
+    DCHECK(receiver_);
+    DCHECK(socket_);
+
+    socket_->SetDelegate(this);
+  }
+
+  ~ControlConnection() override = default;
+
+  void OnStreamCountChanged() {
+    if (!send_stream_count_) {
+      return;
+    }
+    mixer_service::Generic message;
+    auto* counts = message.mutable_stream_count();
+    counts->set_primary(receiver_->primary_stream_count_);
+    counts->set_sfx(receiver_->sfx_stream_count_);
+    socket_->SendProto(message);
+  }
+
+ private:
+  friend class MixerServiceReceiver;
+
+  // mixer_service::MixerSocket::Delegate implementation:
+  bool HandleMetadata(const mixer_service::Generic& message) override {
+    if (message.has_set_volume_limit()) {
+      mixer_->SetOutputLimit(
+          mixer_service::ConvertContentType(
+              message.set_volume_limit().content_type()),
+          message.set_volume_limit().max_volume_multiplier());
+    }
+    if (message.has_set_device_muted()) {
+      mixer_->SetMuted(mixer_service::ConvertContentType(
+                           message.set_device_muted().content_type()),
+                       message.set_device_muted().muted());
+    }
+    if (message.has_set_device_volume()) {
+      mixer_->SetVolume(mixer_service::ConvertContentType(
+                            message.set_device_volume().content_type()),
+                        message.set_device_volume().volume_multiplier());
+    }
+    if (message.has_configure_postprocessor()) {
+      mixer_->SetPostProcessorConfig(
+          message.configure_postprocessor().name(),
+          message.configure_postprocessor().config());
+    }
+    if (message.has_reload_postprocessors()) {
+      mixer_->ResetPostProcessors([](bool, const std::string&) {});
+    }
+    if (message.has_request_stream_count()) {
+      send_stream_count_ = message.request_stream_count().subscribe();
+      OnStreamCountChanged();
+    }
+
+    return true;
+  }
+
+  bool HandleAudioData(char* data, int size, int64_t timestamp) override {
+    return true;
+  }
+
+  bool HandleAudioBuffer(scoped_refptr<net::IOBuffer> buffer,
+                         char* data,
+                         int size,
+                         int64_t timestamp) override {
+    return true;
+  }
+
+  void OnConnectionError() override {
+    receiver_->RemoveControlConnection(this);
+  }
+
+  StreamMixer* const mixer_;
+  MixerServiceReceiver* const receiver_;
+  const std::unique_ptr<mixer_service::MixerSocket> socket_;
+
+  bool send_stream_count_ = false;
+
+  DISALLOW_COPY_AND_ASSIGN(ControlConnection);
+};
+
 MixerServiceReceiver::MixerServiceReceiver(StreamMixer* mixer) : mixer_(mixer) {
   DCHECK(mixer_);
 }
 
 MixerServiceReceiver::~MixerServiceReceiver() = default;
 
+void MixerServiceReceiver::OnStreamCountChanged(int primary, int sfx) {
+  primary_stream_count_ = primary;
+  sfx_stream_count_ = sfx;
+
+  for (const auto& control : control_connections_) {
+    control.second->OnStreamCountChanged();
+  }
+}
+
 void MixerServiceReceiver::CreateOutputStream(
     std::unique_ptr<mixer_service::MixerSocket> socket,
     const mixer_service::Generic& message) {
@@ -45,7 +145,15 @@ void MixerServiceReceiver::CreateAudioRedirection(
 void MixerServiceReceiver::CreateControlConnection(
     std::unique_ptr<mixer_service::MixerSocket> socket,
     const mixer_service::Generic& message) {
-  LOG(INFO) << "Unhandled control connection";
+  auto connection =
+      std::make_unique<ControlConnection>(mixer_, this, std::move(socket));
+  ControlConnection* ptr = connection.get();
+  control_connections_[ptr] = std::move(connection);
+  ptr->HandleMetadata(message);
+}
+
+void MixerServiceReceiver::RemoveControlConnection(ControlConnection* ptr) {
+  control_connections_.erase(ptr);
 }
 
 }  // namespace media

chromecast/media/cma/backend/mixer/mixer_service_receiver.h

@@ -7,6 +7,7 @@
 
 #include <memory>
 
+#include "base/containers/flat_map.h"
 #include "base/macros.h"
 #include "chromecast/media/audio/mixer_service/receiver/receiver.h"
 
@@ -24,7 +25,12 @@ class MixerServiceReceiver : public mixer_service::Receiver {
   explicit MixerServiceReceiver(StreamMixer* mixer);
   ~MixerServiceReceiver() override;
 
+  // Called by the mixer when the active stream count changes.
+  void OnStreamCountChanged(int primary, int sfx);
+
  private:
+  class ControlConnection;
+
   // mixer_service::Receiver implementation:
   void CreateOutputStream(std::unique_ptr<mixer_service::MixerSocket> socket,
                           const mixer_service::Generic& message) override;
@@ -38,8 +44,15 @@ class MixerServiceReceiver : public mixer_service::Receiver {
       std::unique_ptr<mixer_service::MixerSocket> socket,
       const mixer_service::Generic& message) override;
 
+  void RemoveControlConnection(ControlConnection* ptr);
+
   StreamMixer* const mixer_;
 
+  base::flat_map<ControlConnection*, std::unique_ptr<ControlConnection>>
+      control_connections_;
+  int primary_stream_count_ = 0;
+  int sfx_stream_count_ = 0;
+
   DISALLOW_COPY_AND_ASSIGN(MixerServiceReceiver);
 };
 

chromecast/media/cma/backend/mixer/mock_mixer_source.h

@@ -39,6 +39,7 @@ class MockMixerSource : public MixerInput::Source {
   AudioContentType content_type() override { return content_type_; }
   int desired_read_size() override { return 1; }
   int playout_channel() override { return playout_channel_; }
+  bool active() override { return true; }
 
   MOCK_METHOD2(InitializeAudioPlayback, void(int, RenderingDelay));
   MOCK_METHOD3(FillAudioPlaybackFrames,

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

@@ -594,6 +594,7 @@ void StreamMixer::SignalError(MixerInput::Source::MixerError error) {
   }
   inputs_.clear();
   SetCloseTimeout();
+  UpdateStreamCounts();
 }
 
 int StreamMixer::GetEffectiveChannelCount(MixerInput::Source* input_source) {
@@ -664,6 +665,7 @@ void StreamMixer::AddInput(MixerInput::Source* input_source) {
 
   inputs_[input_source] = std::move(input);
   UpdatePlayoutChannel();
+  UpdateStreamCounts();
 }
 
 void StreamMixer::RemoveInput(MixerInput::Source* input_source) {
@@ -687,6 +689,7 @@ void StreamMixer::RemoveInputOnThread(MixerInput::Source* input_source) {
 
   ignored_inputs_.erase(input_source);
   UpdatePlayoutChannel();
+  UpdateStreamCounts();
 
   if (inputs_.empty()) {
     SetCloseTimeout();
@@ -725,6 +728,20 @@ void StreamMixer::UpdatePlayoutChannel() {
   mixer_pipeline_->SetPlayoutChannel(playout_channel_);
 }
 
+void StreamMixer::UpdateStreamCounts() {
+  MAKE_SURE_MIXER_THREAD(UpdateStreamCounts);
+
+  int primary = 0;
+  int sfx = 0;
+  for (const auto& it : inputs_) {
+    if (it.second->source()->active()) {
+      (it.second->primary() ? primary : sfx) += 1;
+    }
+  }
+  receiver_.Post(FROM_HERE, &MixerServiceReceiver::OnStreamCountChanged,
+                 primary, sfx);
+}
+
 MediaPipelineBackend::AudioDecoder::RenderingDelay
 StreamMixer::GetTotalRenderingDelay(FilterGroup* filter_group) {
   DCHECK(mixer_task_runner_->BelongsToCurrentThread());
@@ -941,9 +958,9 @@ void StreamMixer::SetVolumeMultiplier(MixerInput::Source* source,
   }
 }
 
-void StreamMixer::SetPostProcessorConfig(const std::string& name,
-                                         const std::string& config) {
-  MAKE_SURE_MIXER_THREAD(SetPostProcessorConfig, name, config);
+void StreamMixer::SetPostProcessorConfig(std::string name, std::string config) {
+  MAKE_SURE_MIXER_THREAD(SetPostProcessorConfig, std::move(name),
+                         std::move(config));
 
   mixer_pipeline_->SetPostProcessorConfig(name, config);
 }

chromecast/media/cma/backend/mixer/stream_mixer.h

@@ -115,11 +115,14 @@ class StreamMixer : public MixerControl {
   void SetVolumeMultiplier(MixerInput::Source* source, float multiplier);
 
   // Sends configuration string |config| to processor |name|.
-  void SetPostProcessorConfig(const std::string& name,
-                              const std::string& config);
+  void SetPostProcessorConfig(std::string name, std::string config);
 
   void ResetPostProcessors(CastMediaShlib::ResultCallback callback);
 
+  // Updates the counts of active streams and signals any observing control
+  // connections.
+  void UpdateStreamCounts();
+
   // Test-only methods.
   StreamMixer(std::unique_ptr<MixerOutputStream> output,
               std::unique_ptr<base::Thread> mixer_thread,

chromecast/media/cma/backend/volume_control.cc

@@ -28,6 +28,7 @@
 #include "base/threading/thread.h"
 #include "base/values.h"
 #include "chromecast/base/serializers.h"
+#include "chromecast/media/audio/mixer_service/control_connection.h"
 #include "chromecast/media/cma/backend/audio_buildflags.h"
 #include "chromecast/media/cma/backend/cast_audio_json.h"
 #include "chromecast/media/cma/backend/mixer/stream_mixer.h"
@@ -219,6 +220,8 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
   void InitializeOnThread() {
     DCHECK(thread_.task_runner()->BelongsToCurrentThread());
     system_volume_control_ = SystemVolumeControl::Create(this);
+    mixer_ = std::make_unique<mixer_service::ControlConnection>();
+    mixer_->Connect();
 
     double dbfs;
     for (auto type : {AudioContentType::kMedia, AudioContentType::kAlarm,
@@ -229,9 +232,9 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
       if (BUILDFLAG(SYSTEM_OWNS_VOLUME)) {
         // If ALSA owns volume, our internal mixer should not apply any scaling
         // multiplier.
-        StreamMixer::Get()->SetVolume(type, 1.0f);
+        mixer_->SetVolume(type, 1.0f);
       } else {
-        StreamMixer::Get()->SetVolume(type, DbFsToScale(dbfs));
+        mixer_->SetVolume(type, DbFsToScale(dbfs));
       }
 
       // Note that mute state is not persisted across reboots.
@@ -279,8 +282,7 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
 
     float dbfs = VolumeControl::VolumeToDbFS(level);
     if (!BUILDFLAG(SYSTEM_OWNS_VOLUME)) {
-      StreamMixer::Get()->SetVolume(
-          type, DbFsToScale(dbfs) * volume_multipliers_[type]);
+      mixer_->SetVolume(type, DbFsToScale(dbfs) * volume_multipliers_[type]);
     }
 
     if (!from_system && type == AudioContentType::kMedia) {
@@ -306,7 +308,7 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
     volume_multipliers_[type] = multiplier;
     float scale =
         DbFsToScale(VolumeControl::VolumeToDbFS(volumes_[type])) * multiplier;
-    StreamMixer::Get()->SetVolume(type, scale);
+    mixer_->SetVolume(type, scale);
   }
 
   void SetMutedOnThread(VolumeChangeSource source,
@@ -325,7 +327,7 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
     }
 
     if (!BUILDFLAG(SYSTEM_OWNS_VOLUME)) {
-      StreamMixer::Get()->SetMuted(type, muted);
+      mixer_->SetMuted(type, muted);
     }
 
     if (!from_system && type == AudioContentType::kMedia) {
@@ -350,8 +352,8 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
       return;
     }
     limit = base::ClampToRange(limit, 0.0f, 1.0f);
-    StreamMixer::Get()->SetOutputLimit(
-        type, DbFsToScale(VolumeControl::VolumeToDbFS(limit)));
+    mixer_->SetVolumeLimit(type,
+                           DbFsToScale(VolumeControl::VolumeToDbFS(limit)));
 
     if (type == AudioContentType::kMedia) {
       system_volume_control_->SetLimit(limit);
@@ -389,6 +391,7 @@ class VolumeControlInternal : public SystemVolumeControl::Delegate {
   base::WaitableEvent initialize_complete_event_;
 
   std::unique_ptr<SystemVolumeControl> system_volume_control_;
+  std::unique_ptr<mixer_service::ControlConnection> mixer_;
 
   DISALLOW_COPY_AND_ASSIGN(VolumeControlInternal);
 };
@@ -402,6 +405,7 @@ VolumeControlInternal& GetVolumeControl() {
 
 // static
 void VolumeControl::Initialize(const std::vector<std::string>& argv) {
+  StreamMixer::Get();
   GetVolumeControl();
 }