Get PulseAudio implementation working.

media/audio/pulse/pulse_output.cc

Index: media/audio/pulse/pulse_output.cc
diff --git a/media/audio/pulse/pulse_output.cc b/media/audio/pulse/pulse_output.cc
index 82f9fb926a8ade41747b76db34562c85de3fde20..0687e6ef3b5c4984068208acf46a7fd45e47039d 100644
--- a/media/audio/pulse/pulse_output.cc
+++ b/media/audio/pulse/pulse_output.cc
@@ -4,41 +4,45 @@
 
 #include "media/audio/pulse/pulse_output.h"
 
-#include "base/bind.h"
+#include <pulse/pulseaudio.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 {
 
+// A helper class that acquires pa_threaded_mainloop_lock() while in scope.
+class AutoPulseLock {
+ public:
+  explicit AutoPulseLock(pa_threaded_mainloop* pa_mainloop)
+      : pa_mainloop_(pa_mainloop) {
+    pa_threaded_mainloop_lock(pa_mainloop_);
+  }
+
+  ~AutoPulseLock() {
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+  }
+
+ private:
+  pa_threaded_mainloop* pa_mainloop_;
+
+  DISALLOW_COPY_AND_ASSIGN(AutoPulseLock);
+};
+
 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;
   }
 }
@@ -71,202 +75,245 @@ static pa_channel_position ChromiumToPAChannelPosition(Channels channel) {
       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);
+  for (Channels ch = LEFT; ch < CHANNELS_MAX;
+       ch = static_cast<Channels>(ch + 1)) {
+    int channel_index = ChannelOrder(channel_layout, ch);
+    if (channel_index < 0)
+      continue;
 
-  // All channel maps have the same size array of channel positions.
-  for (unsigned int channel = 0; channel != CHANNELS_MAX; ++channel) {
-    int channel_position = kChannelOrderings[channel_layout][channel];
-    if (channel_position > -1) {
-      channel_map.map[channel_position] = ChromiumToPAChannelPosition(
-          static_cast<Channels>(channel));
-    } else {
-      // PulseAudio expects unused channels in channel maps to be filled with
-      // PA_CHANNEL_POSITION_MONO.
-      channel_map.map[channel_position] = PA_CHANNEL_POSITION_MONO;
-    }
-  }
-
-  // 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;
+    channel_map.map[channel_index] = ChromiumToPAChannelPosition(ch);
   }
 
   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, pa_context_notify_cb
+void PulseAudioOutputStream::ContextNotifyCallback(pa_context* c,
+                                                   void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+
+  // Forward unexpected failures to the AudioSourceCallback if available.  All
+  // these variables are only modified under pa_threaded_mainloop_lock() so this
+  // should be thread safe.
+  if (c && stream->source_callback_ &&
+      pa_context_get_state(c) == PA_CONTEXT_FAILED) {
+    stream->source_callback_->OnError(stream, pa_context_errno(c));
+  }
 
-// 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);
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
 }
 
-// static
-void PulseAudioOutputStream::WriteRequestCallback(pa_stream* playback_handle,
-                                                  size_t length,
-                                                  void* stream_addr) {
-  PulseAudioOutputStream* stream =
-      reinterpret_cast<PulseAudioOutputStream*>(stream_addr);
+// static, pa_stream_notify_cb
+void PulseAudioOutputStream::StreamNotifyCallback(pa_stream* s, void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+
+  // Forward unexpected failures to the AudioSourceCallback if available.  All
+  // these variables are only modified under pa_threaded_mainloop_lock() so this
+  // should be thread safe.
+  if (s && stream->source_callback_ &&
+      pa_stream_get_state(s) == PA_STREAM_FAILED) {
+    stream->source_callback_->OnError(
+        stream, pa_context_errno(stream->pa_context_));
+  }
 
-  DCHECK(stream->manager_->GetMessageLoop()->BelongsToCurrentThread());
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
 
-  stream->write_callback_handled_ = true;
+// static, pa_stream_success_cb_t
+void PulseAudioOutputStream::StreamSuccessCallback(pa_stream* s, int success,
+                                                   void* p_this) {
+  PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
+  pa_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
 
-  // Fulfill write request.
-  stream->FulfillWriteRequest(length);
+// static, pa_stream_request_cb_t
+void PulseAudioOutputStream::StreamRequestCallback(pa_stream* s, size_t len,
+                                                   void* p_this) {
+  // Fulfill write request; must always result in a pa_stream_write() call.
+  static_cast<PulseAudioOutputStream*>(p_this)->FulfillWriteRequest(len);
 }
 
 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),
+      pa_stream_(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_);
+  // All internal structures should already have been freed in Close(), which
+  // calls AudioManagerBase::ReleaseOutputStream() which deletes this object.
+  DCHECK(!pa_stream_);
   DCHECK(!pa_context_);
   DCHECK(!pa_mainloop_);
 }
 
+// Helper macro for Open() to avoid code spam and string bloat.
+#define RETURN_ON_FAILURE(expression, message) do { \
+  if (!(expression)) { \
+    if (pa_context_) { \
+      DLOG(ERROR) << message << "  Error: " \
+                  << pa_strerror(pa_context_errno(pa_context_)); \
+    } else { \
+      DLOG(ERROR) << message; \
+    } \
+    return false; \
+  } \
+} while(0)
+
 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.
+  pa_mainloop_ = pa_threaded_mainloop_new();
+  RETURN_ON_FAILURE(pa_mainloop_, "Failed to create PulseAudio main loop.");
 
-  // 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);
-
-  // 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;
-    }
+  RETURN_ON_FAILURE(pa_context_, "Failed to create PulseAudio context.");
+
+  // A state callback must be set before calling pa_threaded_mainloop_lock() or
+  // pa_threaded_mainloop_wait() calls may lead to dead lock.
+  pa_context_set_state_callback(pa_context_, &ContextNotifyCallback, this);
+
+  // Lock the main loop while setting up the context.  Failure to do so may lead
+  // to crashes as the PulseAudio thread tries to run before things are ready.
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  RETURN_ON_FAILURE(
+      pa_threaded_mainloop_start(pa_mainloop_) == 0,
+      "Failed to start PulseAudio main loop.");
+  RETURN_ON_FAILURE(
+      pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL) == 0,
+      "Failed to connect PulseAudio context.");
+
+  // Wait until |pa_context_| is ready.  pa_threaded_mainloop_wait() must be
+  // called after pa_context_get_state() in case the context is already ready,
+  // otherwise pa_threaded_mainloop_wait() will hang indefinitely.
+  while (true) {
+    pa_context_state_t context_state = pa_context_get_state(pa_context_);
+    RETURN_ON_FAILURE(
+        PA_CONTEXT_IS_GOOD(context_state), "Invalid PulseAudio context state.");
+    if (context_state == PA_CONTEXT_READY)
+      break;
+    pa_threaded_mainloop_wait(pa_mainloop_);
   }
 
   // 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.
   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);
+  pa_stream_ = pa_stream_new(
+      pa_context_, "Playback", &pa_sample_specifications, map);
+  RETURN_ON_FAILURE(pa_stream_, "Failed to create PulseAudio stream.");
+  pa_stream_set_state_callback(pa_stream_, &StreamNotifyCallback, this);
 
-  // Initialize client buffer.
-  uint32 output_packet_size = frames_per_packet_ * bytes_per_frame_;
-  client_buffer_.reset(new media::SeekableBuffer(0, output_packet_size));
+  // Even though we start the stream corked below, PulseAudio will issue one
+  // stream request after setup.  FulfillWriteRequest() must fulfill the write.
+  pa_stream_set_write_callback(pa_stream_, &StreamRequestCallback, this);
 
-  // Set write callback.
-  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.minreq = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.prebuf = params_.GetBytesPerBuffer();
+  pa_buffer_attributes.tlength = 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);
-
-  if (!playback_handle_) {
-    Reset();
-    return false;
+  // TODO(dalecurtis): Pulse tends to want really large buffer sizes if we are
+  // not using the native sample rate.  We should always open the stream with
+  // PA_STREAM_FIX_RATE and ensure this is true.
+  RETURN_ON_FAILURE(
+      pa_stream_connect_playback(
+          pa_stream_, NULL, &pa_buffer_attributes,
+          static_cast<pa_stream_flags_t>(
+              PA_STREAM_ADJUST_LATENCY | PA_STREAM_AUTO_TIMING_UPDATE |
+              PA_STREAM_NOT_MONOTONIC | PA_STREAM_START_CORKED),
+          NULL, NULL) == 0,
+      "Failed to connect PulseAudio stream.");
+
+  // Wait for the stream to be ready.
+  while (true) {
+    pa_stream_state_t stream_state = pa_stream_get_state(pa_stream_);
+    RETURN_ON_FAILURE(
+        PA_STREAM_IS_GOOD(stream_state), "Invalid PulseAudio stream state.");
+    if (stream_state == PA_STREAM_READY)
+      break;
+    pa_threaded_mainloop_wait(pa_mainloop_);
   }
 
   return true;
 }
 
+#undef RETURN_ON_FAILURE
+
 void PulseAudioOutputStream::Reset() {
-  stream_stopped_ = true;
+  if (!pa_mainloop_) {
+    DCHECK(!pa_stream_);
+    DCHECK(!pa_context_);
+    return;
+  }
 
-  // Close the stream.
-  if (playback_handle_) {
-    pa_stream_flush(playback_handle_, NULL, NULL);
-    pa_stream_disconnect(playback_handle_);
+  {
+    AutoPulseLock auto_lock(pa_mainloop_);
+
+    // Close the stream.
+    if (pa_stream_) {
+      // Ensure all samples are played out before shutdown.
+      WaitForPulseOperation(pa_stream_flush(
+          pa_stream_, &StreamSuccessCallback, this));
+
+      // Release PulseAudio structures.
+      pa_stream_disconnect(pa_stream_);
+      pa_stream_set_write_callback(pa_stream_, NULL, NULL);
+      pa_stream_set_state_callback(pa_stream_, NULL, NULL);
+      pa_stream_unref(pa_stream_);
+      pa_stream_ = NULL;
+    }
 
-    // Release PulseAudio structures.
-    pa_stream_unref(playback_handle_);
-    playback_handle_ = NULL;
-  }
-  if (pa_context_) {
-    pa_context_unref(pa_context_);
-    pa_context_ = NULL;
-  }
-  if (pa_mainloop_) {
-    pa_mainloop_free(pa_mainloop_);
-    pa_mainloop_ = NULL;
+    if (pa_context_) {
+      pa_context_disconnect(pa_context_);
+      pa_context_set_state_callback(pa_context_, NULL, NULL);
+      pa_context_unref(pa_context_);
+      pa_context_ = NULL;
+    }
   }
 
-  // Release internal buffer.
-  client_buffer_.reset();
+  pa_threaded_mainloop_stop(pa_mainloop_);
+  pa_threaded_mainloop_free(pa_mainloop_);
+  pa_mainloop_ = NULL;
 }
 
 void PulseAudioOutputStream::Close() {
@@ -279,138 +326,107 @@ void PulseAudioOutputStream::Close() {
   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(pa_stream_, &pa_latency_micros, &negative) != 0)
+    return 0;
 
-  if (stream_stopped_)
-    return;
+  if (negative)
+    return 0;
 
-  // 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()));
-  }
-}
-
-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_;
-
-  DCHECK_LE(packet_size, packet_size_);
-  // 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());
-
-  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;
+  return (pa_latency_micros * params_.sample_rate() *
+          params_.GetBytesPerFrame()) / base::Time::kMicrosecondsPerSecond;
 }
 
 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;
+  CHECK_EQ(requested_bytes, static_cast<size_t>(params_.GetBytesPerBuffer()));
 
-  // 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();
+  int frames_filled = 0;
+  if (source_callback_) {
+    frames_filled = source_callback_->OnMoreData(
+        audio_bus_.get(), AudioBuffersState(0, GetHardwareLatencyInBytes()));
   }
 
-  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);
+  // Zero any unfilled data so it plays back as silence.
+  if (frames_filled < audio_bus_->frames()) {
+    audio_bus_->ZeroFramesPartial(
+        frames_filled, audio_bus_->frames() - frames_filled);
   }
 
-  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;
+  // PulseAudio won't always be able to provide a buffer large enough, so we may
+  // need to request multiple buffers and fill them individually.
+  int current_frame = 0;
+  size_t bytes_remaining = requested_bytes;
+  while (bytes_remaining > 0) {
+    void* buffer = NULL;
+    size_t bytes_to_fill = bytes_remaining;
+    CHECK_GE(pa_stream_begin_write(pa_stream_, &buffer, &bytes_to_fill), 0);
+
+    // In case PulseAudio gives us a bigger buffer than we want, cap our size.
+    bytes_to_fill = std::min(
+        std::min(bytes_remaining, bytes_to_fill),
+        static_cast<size_t>(params_.GetBytesPerBuffer()));
+
+    int frames_to_fill = bytes_to_fill / params_.GetBytesPerFrame();;
+
+    // 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_->ToInterleavedPartial(
+        current_frame, frames_to_fill, params_.bits_per_sample() / 8, buffer);
+    media::AdjustVolume(buffer, bytes_to_fill, params_.channels(),
+                        params_.bits_per_sample() / 8, volume_);
+
+    if (pa_stream_write(pa_stream_, buffer, bytes_to_fill, NULL, 0LL,
+                        PA_SEEK_RELATIVE) < 0) {
+      if (source_callback_) {
+        source_callback_->OnError(this, pa_context_errno(pa_context_));
+      }
+    }
 
-    // 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;
+    bytes_remaining -= bytes_to_fill;
+    current_frame = frames_to_fill;
   }
 }
 
 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_)
+  CHECK(pa_stream_);
+
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  // Ensure the context and stream are ready.
+  if (pa_context_get_state(pa_context_) != PA_CONTEXT_READY &&
+      pa_stream_get_state(pa_stream_) != PA_STREAM_READY) {
+    callback->OnError(this, pa_context_errno(pa_context_));
     return;
+  }
 
   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.
+  WaitForPulseOperation(pa_stream_cork(
+      pa_stream_, 0, &StreamSuccessCallback, this));
 }
 
 void PulseAudioOutputStream::Stop() {
   DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
 
-  stream_stopped_ = true;
+  // Cork (pause) the stream.  Waiting for the main loop lock will ensure
+  // outstanding callbacks have completed.
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  // Flush the stream prior to cork, doing so after will cause hangs.  Write
+  // callbacks are suspended while inside pa_threaded_mainloop_lock() so this
+  // is all thread safe.
+  WaitForPulseOperation(pa_stream_flush(
+      pa_stream_, &StreamSuccessCallback, this));
+
+  WaitForPulseOperation(pa_stream_cork(
+      pa_stream_, 1, &StreamSuccessCallback, this));
+
+  source_callback_ = NULL;
 }
 
 void PulseAudioOutputStream::SetVolume(double volume) {
@@ -425,12 +441,12 @@ void PulseAudioOutputStream::GetVolume(double* volume) {
   *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) {
+  CHECK(op);
+  while (pa_operation_get_state(op) == PA_OPERATION_RUNNING) {
+    pa_threaded_mainloop_wait(pa_mainloop_);
+  }
+  pa_operation_unref(op);
 }
 
 }  // namespace media