Get PulseAudio implementation working.

media/audio/pulse/pulse_output.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 "media/audio/pulse/pulse_output.h"
 
-#include "base/bind.h"
-#include "base/message_loop.h"
+#include "media/audio/audio_manager_base.h"
 #include "media/audio/audio_parameters.h"
 #include "media/audio/audio_util.h"
-#if defined(OS_LINUX)
-#include "media/audio/linux/audio_manager_linux.h"
-#elif defined(OS_OPENBSD)
-#include "media/audio/openbsd/audio_manager_openbsd.h"
-#endif
-#include "media/base/data_buffer.h"
-#include "media/base/seekable_buffer.h"
 
 namespace media {
 
 static pa_sample_format_t BitsToPASampleFormat(int bits_per_sample) {
   switch (bits_per_sample) {
-    // Unsupported sample formats shown for reference.  I am assuming we want
-    // signed and little endian because that is what we gave to ALSA.
     case 8:
       return PA_SAMPLE_U8;
-      // Also 8-bits: PA_SAMPLE_ALAW and PA_SAMPLE_ULAW
     case 16:
       return PA_SAMPLE_S16LE;
-      // Also 16-bits: PA_SAMPLE_S16BE (big endian).
     case 24:
       return PA_SAMPLE_S24LE;
-      // Also 24-bits: PA_SAMPLE_S24BE (big endian).
-      // Other cases: PA_SAMPLE_24_32LE (in LSB of 32-bit field, little endian),
-      //          and PA_SAMPLE_24_32BE (in LSB of 32-bit field, big endian),
     case 32:
       return PA_SAMPLE_S32LE;
-      // Also 32-bits: PA_SAMPLE_S32BE (big endian),
-      //               PA_SAMPLE_FLOAT32LE (floating point little endian),
-      //           and PA_SAMPLE_FLOAT32BE (floating point big endian).
     default:
+      NOTREACHED() << "Invalid bits per sample: " << bits_per_sample;
       return PA_SAMPLE_INVALID;
   }
 }
 
 static pa_channel_position ChromiumToPAChannelPosition(Channels channel) {
   switch (channel) {
     // PulseAudio does not differentiate between left/right and
     // stereo-left/stereo-right, both translate to front-left/front-right.
     case LEFT:
     case STEREO_LEFT:
@@ skipping 14 matching lines @@
     case RIGHT_OF_CENTER:
       return PA_CHANNEL_POSITION_FRONT_RIGHT_OF_CENTER;
     case BACK_CENTER:
       return PA_CHANNEL_POSITION_REAR_CENTER;
     case SIDE_LEFT:
       return PA_CHANNEL_POSITION_SIDE_LEFT;
     case SIDE_RIGHT:
       return PA_CHANNEL_POSITION_SIDE_RIGHT;
     case CHANNELS_MAX:
       return PA_CHANNEL_POSITION_INVALID;
+    default:
+      NOTREACHED() << "Invalid channel: " << channel;
+      return PA_CHANNEL_POSITION_INVALID;
   }
-  NOTREACHED() << "Invalid channel " << channel;
-  return PA_CHANNEL_POSITION_INVALID;
 }
 
 static pa_channel_map ChannelLayoutToPAChannelMap(
     ChannelLayout channel_layout) {
   // Initialize channel map.
   pa_channel_map channel_map;
   pa_channel_map_init(&channel_map);
 
   channel_map.channels = ChannelLayoutToChannelCount(channel_layout);
 
@@ skipping 12 matching lines @@
 
   // Fill in the rest of the unused channels.
   for (unsigned int channel = CHANNELS_MAX; channel != PA_CHANNELS_MAX;
        ++channel) {
     channel_map.map[channel] = PA_CHANNEL_POSITION_MONO;
   }
 
   return channel_map;
 }
 
-static size_t MicrosecondsToBytes(
-    uint32 microseconds, uint32 sample_rate, size_t bytes_per_frame) {
-  return microseconds * sample_rate * bytes_per_frame /
-      base::Time::kMicrosecondsPerSecond;
-}
-
-// static
-void PulseAudioOutputStream::ContextStateCallback(pa_context* context,
-                                                  void* state_addr) {
-  pa_context_state_t* state = static_cast<pa_context_state_t*>(state_addr);
-  *state = pa_context_get_state(context);
-}
-
 // static
 void PulseAudioOutputStream::WriteRequestCallback(pa_stream* playback_handle,
-                                                  size_t length,
-                                                  void* stream_addr) {
-  PulseAudioOutputStream* stream =
-      reinterpret_cast<PulseAudioOutputStream*>(stream_addr);
-
-  DCHECK(stream->manager_->GetMessageLoop()->BelongsToCurrentThread());
-
-  stream->write_callback_handled_ = true;
-
-  // Fulfill write request.
+                                                  size_t length, void* p_this) {
+  // Fulfill write request; must always result in a pa_stream_write() call.
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
   stream->FulfillWriteRequest(length);
 }
 
 PulseAudioOutputStream::PulseAudioOutputStream(const AudioParameters& params,
-                                               AudioManagerPulse* manager)
-    : channel_layout_(params.channel_layout()),
-      channel_count_(ChannelLayoutToChannelCount(channel_layout_)),
-      sample_format_(BitsToPASampleFormat(params.bits_per_sample())),
-      sample_rate_(params.sample_rate()),
-      bytes_per_frame_(params.GetBytesPerFrame()),
+                                               AudioManagerBase* manager)
+    : params_(params),
       manager_(manager),
       pa_context_(NULL),
       pa_mainloop_(NULL),
       playback_handle_(NULL),
-      packet_size_(params.GetBytesPerBuffer()),
-      frames_per_packet_(packet_size_ / bytes_per_frame_),
-      client_buffer_(NULL),
       volume_(1.0f),
-      stream_stopped_(true),
-      write_callback_handled_(false),
-      ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       source_callback_(NULL) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  // TODO(slock): Sanity check input values.
+  CHECK(params_.IsValid());
+  audio_bus_ = AudioBus::Create(params_);
 }
 
 PulseAudioOutputStream::~PulseAudioOutputStream() {
   // All internal structures should already have been freed in Close(),
   // which calls AudioManagerPulse::Release which deletes this object.
   DCHECK(!playback_handle_);
   DCHECK(!pa_context_);
   DCHECK(!pa_mainloop_);
 }
 
+// Helper macro for Open() to avoid code spam and string bloat.
+#define PULSE_CHECK(expression, message) \

[scherkus (not reviewing), 2012/10/11 20:19:31]

this isn't really a CHECK... a more appropriate name would be
RETURN_ON_FAILURE() similar to our code under content/common/gpu/media/

[DaleCurtis, 2012/10/12 00:09:12]

PULSE_CHECK sounds cooler though! :) Sadly, done.

+  if (expression) { \

[scherkus (not reviewing), 2012/10/11 20:19:31]

you should check for the negated expression

reading the CHECKs below seem strange as we're checking for the error value as
opposed to the success value (i.e., CHECK that it's null?!)

[scherkus (not reviewing), 2012/10/11 20:19:31]

this should be wrapped in a do { ... } while (0)

see
http://stackoverflow.com/questions/1067226/c-multi-line-macro-do-while0-vs-sc...

[DaleCurtis, 2012/10/12 00:09:12]

That's not necessary here since the code's already in an if() {} block.

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    if (pa_context_) { \
+      pa_threaded_mainloop_unlock(pa_mainloop_); \

[scherkus (not reviewing), 2012/10/11 20:19:31]

how about declaring AutoPulseAudioLock helper at the top of this file?

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+      DLOG(ERROR) << message << pa_context_errno(pa_context_); \

[scherkus (not reviewing), 2012/10/11 20:19:31]

this forces log messages to know that the context error number gets tacked on
the end

how about a separate DLOG() for pa_context_errno() when pa_context_ != NULL or
applying some spacing/formatting?

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    } else { \
+      DLOG(ERROR) << message; \
+    } \
+    return false; \
+  }
+
 bool PulseAudioOutputStream::Open() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  // TODO(slock): Possibly move most of this to an OpenPlaybackDevice function
-  // in a new class 'pulse_util', like alsa_util.
+  // Create a mainloop API and connect to the default server.
+  pa_mainloop_ = pa_threaded_mainloop_new();
+  PULSE_CHECK(!pa_mainloop_, "Failed to create PulseAudio mainloop.");
 
-  // Create a mainloop API and connect to the default server.
-  pa_mainloop_ = pa_mainloop_new();
-  pa_mainloop_api* pa_mainloop_api = pa_mainloop_get_api(pa_mainloop_);
+  pa_mainloop_api* pa_mainloop_api = pa_threaded_mainloop_get_api(pa_mainloop_);
   pa_context_ = pa_context_new(pa_mainloop_api, "Chromium");
-  pa_context_state_t pa_context_state = PA_CONTEXT_UNCONNECTED;
-  pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOFLAGS, NULL);
+  PULSE_CHECK(!pa_context_, "Failed to create PulseAudio context.");
+
+  // Lock the mainloop while we setup our context.  Failing to do so will lead
+  // to crashes as the PulseAudio thread tries to run before things are ready.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+
+  PULSE_CHECK(pa_threaded_mainloop_start(pa_mainloop_) < 0,
+              "Failed to start PulseAudio mainloop: ");
+  PULSE_CHECK(pa_context_connect(
+                  pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL) < 0,
+              "Failed to connect PulseAudio context: ");
 
   // Wait until PulseAudio is ready.
-  pa_context_set_state_callback(pa_context_, &ContextStateCallback,
-                                &pa_context_state);
-  while (pa_context_state != PA_CONTEXT_READY) {
-    pa_mainloop_iterate(pa_mainloop_, 1, NULL);
-    if (pa_context_state == PA_CONTEXT_FAILED ||
-        pa_context_state == PA_CONTEXT_TERMINATED) {
-      Reset();
-      return false;
-    }
-  }
+  pa_context_state_t context_state;
+  do {
+    pa_threaded_mainloop_wait(pa_mainloop_);

[scherkus (not reviewing), 2012/10/11 20:19:31]

should we wait before checking the state?

i.e., will we hang if state reaches READY before we get to this line as no more
events happen?

[DaleCurtis, 2012/10/12 00:09:12]

No we shouldn't, this will hang. Fixed.

+    context_state = pa_context_get_state(pa_context_);
+    PULSE_CHECK(!PA_CONTEXT_IS_GOOD(context_state),
+                "Invalid PulseAudio context state: ");
+  } while (context_state != PA_CONTEXT_READY);
 
   // Set sample specifications.
   pa_sample_spec pa_sample_specifications;
-  pa_sample_specifications.format = sample_format_;
-  pa_sample_specifications.rate = sample_rate_;
-  pa_sample_specifications.channels = channel_count_;
+  pa_sample_specifications.format = BitsToPASampleFormat(
+      params_.bits_per_sample());
+  pa_sample_specifications.rate = params_.sample_rate();
+  pa_sample_specifications.channels = params_.channels();
 
   // Get channel mapping and open playback stream.
+  // TODO(dalecurtis): Is this section correct?

[scherkus (not reviewing), 2012/10/11 20:19:31]

can you elaborate as to what might be incorrect?

[DaleCurtis, 2012/10/12 00:09:12]

No, I just haven't reviewed it thoroughly. Removed the TODO() for now.

   pa_channel_map* map = NULL;
   pa_channel_map source_channel_map = ChannelLayoutToPAChannelMap(
-      channel_layout_);
+      params_.channel_layout());
   if (source_channel_map.channels != 0) {
     // The source data uses a supported channel map so we will use it rather
     // than the default channel map (NULL).
     map = &source_channel_map;
   }
   playback_handle_ = pa_stream_new(pa_context_, "Playback",
                                    &pa_sample_specifications, map);
+  PULSE_CHECK(!playback_handle_, "Failed to create PulseAudio stream: ");
 
-  // Initialize client buffer.
-  uint32 output_packet_size = frames_per_packet_ * bytes_per_frame_;
-  client_buffer_.reset(new media::SeekableBuffer(0, output_packet_size));
-
-  // Set write callback.
+  // Setup callbacks.
   pa_stream_set_write_callback(playback_handle_, &WriteRequestCallback, this);
 
-  // Set server-side buffer attributes.
-  // (uint32_t)-1 is the default and recommended value from PulseAudio's
-  // documentation, found at:
-  // http://freedesktop.org/software/pulseaudio/doxygen/structpa__buffer__attr.html.
+  // Tell pulse audio we only want callbacks of a certain size.
   pa_buffer_attr pa_buffer_attributes;
-  pa_buffer_attributes.maxlength = static_cast<uint32_t>(-1);
-  pa_buffer_attributes.tlength = output_packet_size;
-  pa_buffer_attributes.prebuf = static_cast<uint32_t>(-1);
-  pa_buffer_attributes.minreq = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.maxlength = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.tlength = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.prebuf = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.minreq = params_.GetBytesPerBuffer();
   pa_buffer_attributes.fragsize = static_cast<uint32_t>(-1);
 
   // Connect playback stream.
-  pa_stream_connect_playback(playback_handle_, NULL,
-                             &pa_buffer_attributes,
-                             (pa_stream_flags_t)
-                             (PA_STREAM_INTERPOLATE_TIMING |
-                             PA_STREAM_ADJUST_LATENCY |
-                             PA_STREAM_AUTO_TIMING_UPDATE),
-                             NULL, NULL);
+  PULSE_CHECK(pa_stream_connect_playback(
+                  playback_handle_, NULL, &pa_buffer_attributes,
+                  static_cast<pa_stream_flags_t>(

[scherkus (not reviewing), 2012/10/11 20:19:31]

nit: construct flags variable outside of this function call w/ doc/summary/link
as to what they do

[DaleCurtis, 2012/10/12 00:09:12]

Removed them all except PA_STREAM_START_CORKED as the others were the cause of a
poor audio clock. May not need that smoothing function after all!

PA_STREAM_START_CORKED is self explanatory and explained elsewhere so I've left
it inline.

+                      PA_STREAM_INTERPOLATE_TIMING | PA_STREAM_ADJUST_LATENCY |
+                      PA_STREAM_AUTO_TIMING_UPDATE | PA_STREAM_START_CORKED),
+                  NULL, NULL) < 0,
+              "Failed to connect PulseAudio stream: ");
 
-  if (!playback_handle_) {
-    Reset();
-    return false;
-  }
+  // Wait for the stream to be ready.
+  pa_stream_state_t stream_state;
+  do {
+    pa_threaded_mainloop_wait(pa_mainloop_);

[scherkus (not reviewing), 2012/10/11 20:19:31]

ditto

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    stream_state = pa_stream_get_state(playback_handle_);
+    PULSE_CHECK(!PA_STREAM_IS_GOOD(stream_state),
+                "Invalid PulseAudio stream state: ");
+  } while (stream_state != PA_STREAM_READY);
+
+  // Unlock the mainloop now that everything is setup.
+  pa_threaded_mainloop_unlock(pa_mainloop_);
 
   return true;
 }
 
+#undef PULSE_CHECK
+
 void PulseAudioOutputStream::Reset() {
-  stream_stopped_ = true;
+  if (!pa_mainloop_) {
+    DCHECK(!playback_handle_);
+    DCHECK(!pa_context_);
+    return;
+  }
+
+  pa_threaded_mainloop_lock(pa_mainloop_);
 
   // Close the stream.
   if (playback_handle_) {
-    pa_stream_flush(playback_handle_, NULL, NULL);
-    pa_stream_disconnect(playback_handle_);
+    // Ensure all samples are played out before shutdown.
+    WaitForPulseOperation(pa_stream_flush(playback_handle_, NULL, NULL));
 
     // Release PulseAudio structures.
+    pa_stream_disconnect(playback_handle_);
     pa_stream_unref(playback_handle_);
     playback_handle_ = NULL;
   }
+
   if (pa_context_) {
+    pa_context_disconnect(pa_context_);
     pa_context_unref(pa_context_);
     pa_context_ = NULL;
   }
-  if (pa_mainloop_) {
-    pa_mainloop_free(pa_mainloop_);
-    pa_mainloop_ = NULL;
-  }
 
-  // Release internal buffer.
-  client_buffer_.reset();
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+  pa_threaded_mainloop_stop(pa_mainloop_);
+  pa_threaded_mainloop_free(pa_mainloop_);
+  pa_mainloop_ = NULL;
 }
 
 void PulseAudioOutputStream::Close() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   Reset();
 
   // Signal to the manager that we're closed and can be removed.
   // This should be the last call in the function as it deletes "this".
   manager_->ReleaseOutputStream(this);
 }
 
-void PulseAudioOutputStream::WaitForWriteRequest() {
-  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+int PulseAudioOutputStream::GetHardwareLatencyInBytes() {
+  int negative = 0;
+  pa_usec_t pa_latency_micros = 0;
+  if (pa_stream_get_latency(playback_handle_, &pa_latency_micros,
+                            &negative) != 0 || negative)

[scherkus (not reviewing), 2012/10/11 20:19:31]

can you split this into two ifs?

it's a bit subtle with the negative out-param

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+    return 0;
 
-  if (stream_stopped_)
-    return;
-
-  // Iterate the PulseAudio mainloop.  If PulseAudio doesn't request a write,
-  // post a task to iterate the mainloop again.
-  write_callback_handled_ = false;
-  pa_mainloop_iterate(pa_mainloop_, 1, NULL);
-  if (!write_callback_handled_) {
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
+  return (pa_latency_micros * params_.sample_rate() *
+          params_.GetBytesPerFrame()) / base::Time::kMicrosecondsPerSecond;
 }
 
-bool PulseAudioOutputStream::BufferPacketFromSource() {
-  uint32 buffer_delay = client_buffer_->forward_bytes();
-  pa_usec_t pa_latency_micros;
-  int negative;
-  pa_stream_get_latency(playback_handle_, &pa_latency_micros, &negative);
-  uint32 hardware_delay = MicrosecondsToBytes(pa_latency_micros,
-                                              sample_rate_,
-                                              bytes_per_frame_);
-  // TODO(slock): Deal with negative latency (negative == 1).  This has yet
-  // to happen in practice though.
-  scoped_refptr<media::DataBuffer> packet =
-      new media::DataBuffer(packet_size_);
-  int frames_filled = RunDataCallback(
-      audio_bus_.get(), AudioBuffersState(buffer_delay, hardware_delay));
-  size_t packet_size = frames_filled * bytes_per_frame_;
+int PulseAudioOutputStream::FillBuffer(void* buffer, size_t buffer_size) {
+  CHECK(source_callback_);
+  int frames_filled = source_callback_->OnMoreData(
+      audio_bus_.get(), AudioBuffersState(0, GetHardwareLatencyInBytes()));
 
-  DCHECK_LE(packet_size, packet_size_);
+  int packet_size = frames_filled * params_.GetBytesPerFrame();
+  if (packet_size == 0)
+    return 0;
+
   // Note: If this ever changes to output raw float the data must be clipped and
   // sanitized since it may come from an untrusted source such as NaCl.
-  audio_bus_->ToInterleaved(
-      frames_filled, bytes_per_frame_ / channel_count_,
-      packet->GetWritableData());
+  CHECK_LE(static_cast<size_t>(packet_size), buffer_size);
+  audio_bus_->ToInterleaved(frames_filled, params_.bits_per_sample(), buffer);
 
-  if (packet_size == 0)
-    return false;
-
-  media::AdjustVolume(packet->GetWritableData(),
-                      packet_size,
-                      channel_count_,
-                      bytes_per_frame_ / channel_count_,
-                      volume_);
-  packet->SetDataSize(packet_size);
-  // Add the packet to the buffer.
-  client_buffer_->Append(packet);
-  return true;
+  media::AdjustVolume(buffer, packet_size, params_.channels(),
+                      params_.bits_per_sample(), volume_);
+  return packet_size;
 }
 
 void PulseAudioOutputStream::FulfillWriteRequest(size_t requested_bytes) {
-  // If we have enough data to fulfill the request, we can finish the write.
-  if (stream_stopped_)
-    return;
+  // Let pa_stream_begin_write auto detect the buffer size, it should choose the

[scherkus (not reviewing), 2012/10/11 20:19:31]

add ()

[DaleCurtis, 2012/10/12 00:09:12]

Done.

+  // same size as the callback request.  We CHECK() to make sure it does.
+  size_t bytes_available = static_cast<size_t>(-1);
+  void* audio_buffer = NULL;
 
-  // Request more data from the source until we can fulfill the request or
-  // fail to receive anymore data.
-  bool buffering_successful = true;
-  size_t forward_bytes = static_cast<size_t>(client_buffer_->forward_bytes());
-  while (forward_bytes < requested_bytes && buffering_successful) {
-    buffering_successful = BufferPacketFromSource();
-  }
+  // Request a buffer from PulseAudio and ensure it's the correct size.
+  CHECK_GE(pa_stream_begin_write(
+      playback_handle_, &audio_buffer, &bytes_available), 0);
+  CHECK_EQ(bytes_available, requested_bytes);
+  CHECK_EQ(requested_bytes, static_cast<size_t>(params_.GetBytesPerBuffer()));
 
-  size_t bytes_written = 0;
-  if (client_buffer_->forward_bytes() > 0) {
-    // Try to fulfill the request by writing as many of the requested bytes to
-    // the stream as we can.
-    WriteToStream(requested_bytes, &bytes_written);
-  }
-
-  if (bytes_written < requested_bytes) {
-    // We weren't able to buffer enough data to fulfill the request.  Try to
-    // fulfill the rest of the request later.
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::FulfillWriteRequest,
-        weak_factory_.GetWeakPtr(),
-        requested_bytes - bytes_written));
-  } else {
-    // Continue playback.
-    manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
-}
-
-void PulseAudioOutputStream::WriteToStream(size_t bytes_to_write,
-                                           size_t* bytes_written) {
-  *bytes_written = 0;
-  while (*bytes_written < bytes_to_write) {
-    const uint8* chunk;
-    int chunk_size;
-
-    // Stop writing if there is no more data available.
-    if (!client_buffer_->GetCurrentChunk(&chunk, &chunk_size))
-      break;
-
-    // Write data to stream.
-    pa_stream_write(playback_handle_, chunk, chunk_size,
-                    NULL, 0LL, PA_SEEK_RELATIVE);
-    client_buffer_->Seek(chunk_size);
-    *bytes_written += chunk_size;
-  }
+  int bytes_filled = FillBuffer(audio_buffer, bytes_available);
+  pa_stream_write(playback_handle_, audio_buffer, bytes_filled, NULL, 0LL,

[scherkus (not reviewing), 2012/10/11 20:19:31]

check return value?

[DaleCurtis, 2012/10/12 00:09:12]

Done. Add a lot of error handling elsewhere too.

+                  PA_SEEK_RELATIVE);
 }
 
 void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
   CHECK(callback);
-  DLOG_IF(ERROR, !playback_handle_)
-      << "Open() has not been called successfully";
-  if (!playback_handle_)
-    return;
+  CHECK(playback_handle_);
 
   source_callback_ = callback;
 
-  // Clear buffer, it might still have data in it.
-  client_buffer_->Clear();
-  stream_stopped_ = false;
-
-  // Start playback.
-  manager_->GetMessageLoop()->PostTask(FROM_HERE, base::Bind(
-      &PulseAudioOutputStream::WaitForWriteRequest,
-      weak_factory_.GetWeakPtr()));
+  // Uncork (resume) the stream.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+  WaitForPulseOperation(pa_stream_cork(playback_handle_, 0, NULL, NULL));
+  pa_threaded_mainloop_unlock(pa_mainloop_);
 }
 
 void PulseAudioOutputStream::Stop() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  stream_stopped_ = true;
+  // Cork (pause) the stream.  Waiting for the mainloop lock (should) ensure
+  // outstanding callbacks have completed.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+  WaitForPulseOperation(pa_stream_cork(playback_handle_, 1, NULL, NULL));
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+
+  source_callback_ = NULL;
 }
 
 void PulseAudioOutputStream::SetVolume(double volume) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   volume_ = static_cast<float>(volume);
 }
 
 void PulseAudioOutputStream::GetVolume(double* volume) {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
   *volume = volume_;
 }
 
-int PulseAudioOutputStream::RunDataCallback(
-    AudioBus* audio_bus, AudioBuffersState buffers_state) {
-  if (source_callback_)
-    return source_callback_->OnMoreData(audio_bus, buffers_state);
-
-  return 0;
+void PulseAudioOutputStream::WaitForPulseOperation(pa_operation* op) {

[scherkus (not reviewing), 2012/10/11 20:19:31]

this function assumes that the caller has locked the mainloop

do you need to acquire the lock for running the operations (i.e., cork/uncork)
or for checking the state + waiting?

if it's the latter, PA docs state that their locks are recursive so you _could_
add lock/unlock inside this function + remove the restriction on the callee

[DaleCurtis, 2012/10/12 00:09:12]

I'm pretty sure those calls need the lock too since we want to make sure there
aren't any outstanding callbacks when we set |source_callback_| to NULL, etc.

+  CHECK(op);
+  while (pa_operation_get_state(op) == PA_OPERATION_RUNNING)
+    pa_threaded_mainloop_wait(pa_mainloop_);
+  pa_operation_unref(op);
 }
 
 }  // namespace media