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

remoting/host/linux/pulseaudio_pipe_sink_reader.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 "remoting/host/linux/pulseaudio_pipe_sink_reader.h"
 
 #include <fcntl.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
 
 #include "base/eintr_wrapper.h"
 #include "base/file_path.h"
-#include "base/lazy_instance.h"
 #include "base/logging.h"
 #include "base/stl_util.h"
-#include "remoting/proto/audio.pb.h"
-#include "remoting/host/chromoting_host_context.h"
 
 namespace remoting {
 
 namespace {
 
 // PulseAudio's module-pipe-sink must be configured to use the following
 // parameters for the sink we read from.
-const AudioPacket_SamplingRate kSamplingRate = AudioPacket::SAMPLING_RATE_44100;
+const int kSamplingRate = 48000;

[Wez, 2012/10/31 00:56:28]

nit: Add mention of the change in sample rate to the CL description; I assume
this is to avoid the need to resample when we switch to Opus?

[Sergey Ulanov, 2012/10/31 18:41:24]

Done. Yes, it's so that we don't need to resample when using opus (resampling is
still needed for 44.1k sources, but now it will be happening in pulseaudio)

 const int kChannels = 2;
 const int kBytesPerSample = 2;
 
 // 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;
-
 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) {
+PulseaudioPipeSinkReader::PulseaudioPipeSinkReader(
+    scoped_refptr<base::SingleThreadTaskRunner> task_runner,
+    const FilePath& pipe_name)
+    : task_runner_(task_runner),
+      observers_(new ObserverListThreadSafe<StreamObserver>()) {
+  task_runner_->PostTask(FROM_HERE, base::Bind(
+      &PulseaudioPipeSinkReader::StartOnAudioThread, this, pipe_name));
+}
+
+void PulseaudioPipeSinkReader::StartOnAudioThread(const FilePath& pipe_name) {
+  DCHECK(task_runner_->BelongsToCurrentThread());
+
   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() {
+PulseaudioPipeSinkReader::~PulseaudioPipeSinkReader() {
 }
 
-bool AudioCapturerLinux::Start(const PacketCapturedCallback& callback) {
-  if (pipe_fd_ < 0)
-    return false;
-
-  callback_ = callback;
-
-  return true;
+void PulseaudioPipeSinkReader::AddCapturer(StreamObserver* observer) {
+  observers_->AddObserver(observer);
+}
+void PulseaudioPipeSinkReader::RemoveCapturer(StreamObserver* observer) {
+  observers_->RemoveObserver(observer);
 }
 
-void AudioCapturerLinux::Stop() {
-  callback_.Reset();
-}
-
-bool AudioCapturerLinux::IsStarted() {
-  return !callback_.is_null();
-}
-
-void AudioCapturerLinux::OnFileCanReadWithoutBlocking(int fd) {
+void PulseaudioPipeSinkReader::OnFileCanReadWithoutBlocking(int fd) {
   DCHECK_EQ(fd, pipe_fd_);
   StartTimer();
 }
 
-void AudioCapturerLinux::OnFileCanWriteWithoutBlocking(int fd) {
+void PulseaudioPipeSinkReader::OnFileCanWriteWithoutBlocking(int fd) {
   NOTREACHED();
 }
 
-void AudioCapturerLinux::StartTimer() {
+void PulseaudioPipeSinkReader::StartTimer() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
   started_time_ = base::TimeTicks::Now();
   last_capture_samples_ = 0;
   timer_.Start(FROM_HERE, base::TimeDelta::FromMilliseconds(kCapturingPeriodMs),
-               this, &AudioCapturerLinux::DoCapture);
+               this, &PulseaudioPipeSinkReader::DoCapture);
 }
 
-void AudioCapturerLinux::DoCapture() {
+void PulseaudioPipeSinkReader::DoCapture() {
+  DCHECK(task_runner_->BelongsToCurrentThread());
   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_ptr<AudioPacket> packet(new AudioPacket());
-  packet->add_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());
+  // Dispatch asynchronous notification to the stream observers.
+  scoped_refptr<base::RefCountedString> data_ref =
+      base::RefCountedString::TakeString(&data);
+  observers_->Notify(&StreamObserver::OnDataRead, data_ref);
 }
 
-void AudioCapturerLinux::WaitForPipeReadable() {
+void PulseaudioPipeSinkReader::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();
-}
-
-scoped_ptr<AudioCapturer> AudioCapturer::Create() {
-  FilePath path = g_audio_pipe_name.Get();
-  if (path.empty())
-    return scoped_ptr<AudioCapturer>();
-  return scoped_ptr<AudioCapturer>(new AudioCapturerLinux(path));
-}
-
 }  // namespace remoting