Fix AudioCapturer implementation to read from audio pipe even when there are no active clients.

remoting/host/audio_capturer_linux.cc

 // Copyright (c) 2012 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 "remoting/host/audio_capturer_linux.h"
 
 #include <fcntl.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ skipping 18 matching lines @@
 
 // Read data from the pipe every 40ms.
 const int kCapturingPeriodMs = 40;
 
 #if !defined(F_SETPIPE_SZ)
 // F_SETPIPE_SZ is supported only starting linux 2.6.35, but we want to be able
 // to compile this code on machines with older kernel.
 #define F_SETPIPE_SZ 1031
 #endif  // defined(F_SETPIPE_SZ)
 
-// Pipename used to capture audio stream from.
-// TODO(sergeyu): Pass this to AudioCapturerLinux constructor once we have
-// Linux-specific DesktopEnvironmentFactory
-base::LazyInstance<FilePath>::Leaky
-    g_audio_pipe_name = LAZY_INSTANCE_INITIALIZER;
+
+base::LazyInstance<scoped_refptr<AudioCapturerLinuxCore> >::Leaky
+    g_audio_capturer_core = LAZY_INSTANCE_INITIALIZER;
 
 const int IsPacketOfSilence(const std::string& data) {
   const int64* int_buf = reinterpret_cast<const int64*>(data.data());
   for (size_t i = 0; i < data.size() / sizeof(int64); i++) {
     if (int_buf[i] != 0)
       return false;
   }
   for (size_t i = data.size() - data.size() % sizeof(int64);
        i < data.size(); i++) {
     if (data.data()[i] != 0)
       return false;
   }
   return true;
 }
 
 }  // namespace
 
-AudioCapturerLinux::AudioCapturerLinux(const FilePath& pipe_name) {
+void AudioCapturerLinuxCore::Initialize(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    const FilePath& pipe_name) {
+  g_audio_capturer_core.Get() =
+      new AudioCapturerLinuxCore(task_runner, pipe_name);
+}
+
+AudioCapturerLinuxCore::AudioCapturerLinuxCore(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    const FilePath& pipe_name)
+    : main_task_runner_(task_runner),
+      capturers_(new ObserverListThreadSafe<AudioCapturerLinux>()) {
+  main_task_runner_->PostTask(FROM_HERE, base::Bind(
+      &AudioCapturerLinuxCore::StartReading, this, pipe_name));
+}
+
+void AudioCapturerLinuxCore::StartReading(const FilePath& pipe_name) {
   pipe_fd_ = HANDLE_EINTR(open(
       pipe_name.value().c_str(), O_RDONLY | O_NONBLOCK));
   if (pipe_fd_ < 0) {
     LOG(ERROR) << "Failed to open " << pipe_name.value();
     return;
   }
 
   // Set buffer size for the pipe to the double of what's required for samples
   // of each capturing period.
   int pipe_buffer_size = 2 * kCapturingPeriodMs * kSamplingRate * kChannels *
      kBytesPerSample / base::Time::kMillisecondsPerSecond;
   int result = HANDLE_EINTR(fcntl(pipe_fd_, F_SETPIPE_SZ, pipe_buffer_size));
   if (result < 0) {
     PLOG(ERROR) << "fcntl";
   }
 
   WaitForPipeReadable();
 }
 
-AudioCapturerLinux::~AudioCapturerLinux() {
+AudioCapturerLinuxCore::~AudioCapturerLinuxCore() {
 }
 
-bool AudioCapturerLinux::Start(const PacketCapturedCallback& callback) {
-  if (pipe_fd_ < 0)
-    return false;
-
-  callback_ = callback;
-
-  return true;
+void AudioCapturerLinuxCore::AddCapturer(AudioCapturerLinux* capturer) {
+  capturers_->AddObserver(capturer);
+}
+void AudioCapturerLinuxCore::RemoveCapturer(AudioCapturerLinux* capturer) {
+  capturers_->RemoveObserver(capturer);
 }
 
-void AudioCapturerLinux::Stop() {
-  callback_.Reset();
-}
-
-bool AudioCapturerLinux::IsStarted() {
-  return !callback_.is_null();
-}
-
-void AudioCapturerLinux::OnFileCanReadWithoutBlocking(int fd) {
+void AudioCapturerLinuxCore::OnFileCanReadWithoutBlocking(int fd) {
   DCHECK_EQ(fd, pipe_fd_);
   StartTimer();
 }
 
-void AudioCapturerLinux::OnFileCanWriteWithoutBlocking(int fd) {
+void AudioCapturerLinuxCore::OnFileCanWriteWithoutBlocking(int fd) {
   NOTREACHED();
 }
 
-void AudioCapturerLinux::StartTimer() {
+void AudioCapturerLinuxCore::StartTimer() {
   started_time_ = base::TimeTicks::Now();
   last_capture_samples_ = 0;
   timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kCapturingPeriodMs),
-               this, &AudioCapturerLinux::DoCapture);
+               this, &AudioCapturerLinuxCore::DoCapture);
 }
 
-void AudioCapturerLinux::DoCapture() {
+void AudioCapturerLinuxCore::DoCapture() {
   DCHECK_GT(pipe_fd_, 0);
 
   // Calculate how much we need read from the pipe. Pulseaudio doesn't control
   // how much data it writes to the pipe, so we need to pace the stream, so
   // that we read the exact number of the samples per second we need.
   base::TimeDelta stream_position = base::TimeTicks::Now() - started_time_;
   int64 stream_position_samples = stream_position.InMilliseconds() *
       kSamplingRate / base::Time::kMillisecondsPerSecond;
   int64 samples_to_capture =
       stream_position_samples - last_capture_samples_;
@@ skipping 23 matching lines @@
     return;
   }
 
   // Save any incomplete samples we've read for later. Each packet should
   // contain integer number of samples.
   int incomplete_samples_bytes = pos % (kChannels * kBytesPerSample);
   left_over_bytes_.assign(data, pos - incomplete_samples_bytes,
                           incomplete_samples_bytes);
   data.resize(pos - incomplete_samples_bytes);
 
-  if (callback_.is_null())
-    return;
-
   if (IsPacketOfSilence(data))
     return;
 
+  scoped_refptr<base::RefCountedString> data_ref =

[Wez, 2012/10/30 04:20:11]

nit: Add a comment e.g. "Dispatch asynchronous notification to capturers"

[Sergey Ulanov, 2012/10/30 23:59:48]

Done.

+      base::RefCountedString::TakeString(&data);
+  capturers_->Notify(&AudioCapturerLinux::OnDataRead, data_ref);
+}
+
+void AudioCapturerLinuxCore::WaitForPipeReadable() {
+  timer_.Stop();
+  MessageLoopForIO::current()->WatchFileDescriptor(
+      pipe_fd_, false, MessageLoopForIO::WATCH_READ,
+      &file_descriptor_watcher_, this);
+}
+
+AudioCapturerLinux::AudioCapturerLinux(
+    scoped_refptr<AudioCapturerLinuxCore> core)
+    : core_(core) {
+}
+
+AudioCapturerLinux::~AudioCapturerLinux() {
+}
+
+bool AudioCapturerLinux::Start(const PacketCapturedCallback& callback) {
+  callback_ = callback;
+  core_->AddCapturer(this);

[Wez, 2012/10/30 04:20:11]

nit: Use |g_audio_capturer_core| directly here?

[Sergey Ulanov, 2012/10/30 23:59:48]

See my other comments. I'd like to avoid usage of globals as much as possible.

+  return true;
+}
+
+void AudioCapturerLinux::Stop() {
+  core_->RemoveCapturer(this);
+  callback_.Reset();
+}
+
+bool AudioCapturerLinux::IsStarted() {
+  return !callback_.is_null();
+}
+
+void AudioCapturerLinux::OnDataRead(
+    scoped_refptr<base::RefCountedString> data) {
+  DCHECK(!callback_.is_null());
+
   scoped_ptr<AudioPacket> packet(new AudioPacket());
-  packet->add_data(data);
+  packet->add_data(data->data());
   packet->set_encoding(AudioPacket::ENCODING_RAW);
   packet->set_sampling_rate(kSamplingRate);
   packet->set_bytes_per_sample(AudioPacket::BYTES_PER_SAMPLE_2);
   packet->set_channels(AudioPacket::CHANNELS_STEREO);
   callback_.Run(packet.Pass());
 }
 
-void AudioCapturerLinux::WaitForPipeReadable() {
-  timer_.Stop();
-  MessageLoopForIO::current()->WatchFileDescriptor(
-      pipe_fd_, false, MessageLoopForIO::WATCH_READ,
-      &file_descriptor_watcher_, this);
-}
-
-void AudioCapturerLinux::SetPipeName(const FilePath& pipe_name) {
-  g_audio_pipe_name.Get() = pipe_name;
-}
-
 bool AudioCapturer::IsSupported() {
-  return !g_audio_pipe_name.Get().empty();
+  return g_audio_capturer_core.Get() != NULL;
 }
 
 scoped_ptr<AudioCapturer> AudioCapturer::Create() {
-  FilePath path = g_audio_pipe_name.Get();
-  if (path.empty())
+  scoped_refptr<AudioCapturerLinuxCore> core = g_audio_capturer_core.Get();
+  if (!core)
     return scoped_ptr<AudioCapturer>();
-  return scoped_ptr<AudioCapturer>(new AudioCapturerLinux(path));
+  return scoped_ptr<AudioCapturer>(new AudioCapturerLinux(core));
 }
 
 }  // namespace remoting