A patch making the pulseaudio working with the threaded mainloop.

media/audio/pulse/pulse_output.cc

Index: media/audio/pulse/pulse_output.cc
===================================================================
--- media/audio/pulse/pulse_output.cc	(revision 110923)
+++ media/audio/pulse/pulse_output.cc	(working copy)
@@ -16,6 +16,7 @@
 #include "media/base/data_buffer.h"
 #include "media/base/seekable_buffer.h"
 
+// TODO(xians): Do we support any sample format rather than PA_SAMPLE_S16LE?
 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
@@ -41,71 +42,6 @@
   }
 }
 
-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:
-      return PA_CHANNEL_POSITION_FRONT_LEFT;
-    case RIGHT:
-    case STEREO_RIGHT:
-      return PA_CHANNEL_POSITION_FRONT_RIGHT;
-    case CENTER:
-      return PA_CHANNEL_POSITION_FRONT_CENTER;
-    case LFE:
-      return PA_CHANNEL_POSITION_LFE;
-    case BACK_LEFT:
-      return PA_CHANNEL_POSITION_REAR_LEFT;
-    case BACK_RIGHT:
-      return PA_CHANNEL_POSITION_REAR_RIGHT;
-    case LEFT_OF_CENTER:
-      return PA_CHANNEL_POSITION_FRONT_LEFT_OF_CENTER;
-    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;
-  }
-  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);
-
-  // 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;
-  }
-
-  return channel_map;
-}
-
 static size_t MicrosecondsToBytes(
     uint32 microseconds, uint32 sample_rate, size_t bytes_per_frame) {
   return microseconds * sample_rate * bytes_per_frame /
@@ -113,42 +49,55 @@
 }
 
 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);
+                                                  void* user_data) {
+  PulseAudioOutputStream* audio_stream =
+      reinterpret_cast<PulseAudioOutputStream*>(user_data);
+  pa_context_state_t state = pa_context_get_state(context);
+  switch (state) {
+    case PA_CONTEXT_TERMINATED:
+      audio_stream->context_state_changed_ = true;
+      pa_threaded_mainloop_signal(audio_stream->pa_mainloop_, 0);
+      break;
+    case PA_CONTEXT_READY:
+      audio_stream->context_state_changed_ = true;
+      pa_threaded_mainloop_signal(audio_stream->pa_mainloop_, 0);
+      break;
+    case PA_CONTEXT_UNCONNECTED:
+    case PA_CONTEXT_CONNECTING:
+    case PA_CONTEXT_AUTHORIZING:
+    case PA_CONTEXT_SETTING_NAME:
+    case PA_CONTEXT_FAILED:
+    default:
+      break;
+  }
 }
 
 void PulseAudioOutputStream::WriteRequestCallback(
-    pa_stream* playback_handle, size_t length, void* stream_addr) {
-  PulseAudioOutputStream* stream =
-      static_cast<PulseAudioOutputStream*>(stream_addr);
+    pa_stream* playback_handle, size_t length, void* user_data) {
+  PulseAudioOutputStream* audio_stream =
+      reinterpret_cast<PulseAudioOutputStream*>(user_data);
 
-  DCHECK_EQ(stream->message_loop_, MessageLoop::current());
-
-  stream->write_callback_handled_ = true;
-
-  // Fulfill write request.
-  stream->FulfillWriteRequest(length);
+  audio_stream->FulfillWriteRequest(length);
 }
 
 PulseAudioOutputStream::PulseAudioOutputStream(const AudioParameters& params,
                                                AudioManagerPulse* manager,
                                                MessageLoop* message_loop)
-    : channel_layout_(params.channel_layout),
-      channel_count_(ChannelLayoutToChannelCount(channel_layout_)),
+    : channels_(params.channels),
       sample_format_(BitsToPASampleFormat(params.bits_per_sample)),
       sample_rate_(params.sample_rate),
       bytes_per_frame_(params.channels * params.bits_per_sample / 8),
+      packet_size_(params.GetPacketSize()),
+      frames_per_packet_(packet_size_ / bytes_per_frame_),
       manager_(manager),
       pa_context_(NULL),
       pa_mainloop_(NULL),
       playback_handle_(NULL),
-      packet_size_(params.GetPacketSize()),
-      frames_per_packet_(packet_size_ / bytes_per_frame_),
-      client_buffer_(NULL),
+      pa_buffer_size_(0),
+      buffer_(NULL),
       volume_(1.0f),
       stream_stopped_(true),
-      write_callback_handled_(false),
+      context_state_changed_(false),
       message_loop_(message_loop),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       source_callback_(NULL) {
@@ -156,6 +105,9 @@
   DCHECK(manager_);
 
   // TODO(slock): Sanity check input values.
+
+  // TODO(xians): Check if PA is available here in runtime, and fall back
+  // to ALSA if not available.
 }
 
 PulseAudioOutputStream::~PulseAudioOutputStream() {
@@ -168,107 +120,79 @@
 
 bool PulseAudioOutputStream::Open() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(!pa_mainloop_);
+DLOG(WARNING) << "PULSE AUDIO";
+  // Create a mainloop API and connection to the default server.
+  // The mainloop is the internal asynchronous API event loop.
+  pa_mainloop_ = pa_threaded_mainloop_new();
+  DCHECK(pa_mainloop_) << "Failed to create PA threaded mainloop";
+  if (!pa_mainloop_)
+    return false;
 
-  // TODO(slock): Possibly move most of this to an OpenPlaybackDevice function
-  // in a new class 'pulse_util', like alsa_util.
+  // Start the threaded mainloop.
+  if (pa_threaded_mainloop_start(pa_mainloop_)) {
+    DLOG(ERROR) << "Failed to start mainloop.";
+    return false;
+  }
 
-  // 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_);
+  // Lock the event loop object, effectively blocking the event loop thread
+  // from processing events. This is necessary.
+  pa_threaded_mainloop_lock(pa_mainloop_);
+
+  // TODO(xians): Share one pa_context_ for streams.
+  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);
+  DCHECK(pa_context_) << "Failed to create PA context";
+  if (!pa_context_) {
+    Reset();
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+    return false;
+  }
 
-  // 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;
-    }
+  context_state_changed_ = false;
+  pa_context_set_state_callback(pa_context_, &ContextStateCallback, this);
+  if (pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOAUTOSPAWN, NULL)) {
+    DLOG(ERROR) << "Failed to connect to the context";
+    Reset();
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+    return false;
   }
 
+  while (!context_state_changed_) {
+    pa_threaded_mainloop_wait(pa_mainloop_);
+  }
+
+  if (pa_context_get_state(pa_context_) != PA_CONTEXT_READY) {
+    DLOG(ERROR) <<  "Unknown problem connecting to PulseAudio server";
+    Reset();
+    pa_threaded_mainloop_unlock(pa_mainloop_);
+    return false;
+  }
+
   // 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.channels = channels_;
 
-  // Get channel mapping and open playback stream.
-  pa_channel_map* map = NULL;
-  pa_channel_map source_channel_map = ChannelLayoutToPAChannelMap(
-      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);
-
-  // 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.
-  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.
-  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.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);
-
+  // Create a new play stream
+  playback_handle_ = pa_stream_new(pa_context_, "PlayStream",
+                                   &pa_sample_specifications, NULL);
   if (!playback_handle_) {
+    DLOG(ERROR) << "Open: failed to create PA stream";
     Reset();
+    pa_threaded_mainloop_unlock(pa_mainloop_);
     return false;
   }
 
+  pa_stream_set_write_callback(playback_handle_, &WriteRequestCallback, this);
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+
+  buffer_.reset(new media::SeekableBuffer(0, packet_size_));
   return true;
 }
 
-void PulseAudioOutputStream::Reset() {
-  stream_stopped_ = true;
-
-  // Close the stream.
-  if (playback_handle_) {
-    pa_stream_flush(playback_handle_, NULL, NULL);
-    pa_stream_disconnect(playback_handle_);
-
-    // 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;
-  }
-
-  // Release internal buffer.
-  client_buffer_.reset();
-}
-
 void PulseAudioOutputStream::Close() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
@@ -279,146 +203,187 @@
   manager_->ReleaseOutputStream(this);
 }
 
-void PulseAudioOutputStream::WaitForWriteRequest() {
+void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  CHECK(callback);
 
-  if (stream_stopped_)
+  if (!stream_stopped_)
     return;
+  stream_stopped_ = false;
 
-  // 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_) {
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
-}
+  // First time to start the stream.
+  if (!source_callback_) {
+    // Set server-side playback buffer metrics. Detailed documentation on what
+    // values should be chosen can be found at
+    // freedesktop.org/software/pulseaudio/doxygen/structpa__buffer__attr.html.
+    pa_buffer_attr pa_buffer_attributes;
+    pa_buffer_size_ = packet_size_;
+    pa_buffer_attributes.maxlength =  static_cast<uint32_t>(-1);
+    pa_buffer_attributes.tlength = pa_buffer_size_;
+    pa_buffer_attributes.minreq = pa_buffer_size_ / 2;
+    pa_buffer_attributes.prebuf =
+        pa_buffer_attributes.tlength  - pa_buffer_attributes.minreq;
+    pa_buffer_attributes.fragsize = packet_size_;
+    int err = pa_stream_connect_playback(playback_handle_,
+                                         NULL,  // Default device.
+                                         &pa_buffer_attributes,
+                                         static_cast<pa_stream_flags_t>
+                                         (PA_STREAM_AUTO_TIMING_UPDATE |
+                                          PA_STREAM_INTERPOLATE_TIMING |
+                                          PA_STREAM_ADJUST_LATENCY),
+                                         NULL,  // Default volume.
+                                         NULL  // Standalone stream.
+                                         );
+    if (err) {
+      DLOG(ERROR) << "pa_stream_connect_playback FAILED " << err;
+      Reset();
+      return;
+    }
+  } else {  // Resume the playout stream.
+    // Flush the stream.
+    pa_operation* operation = pa_stream_flush(playback_handle_, NULL, NULL);
+    if (!operation) {
+      DLOG(ERROR) << "PulseAudioOutputStream: failed to flush the playout "
+                  << "stream";
+      return;
+    }
+    // Do not need to wait for the operation.
+    pa_operation_unref(operation);
 
-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_);
-  size_t packet_size = RunDataCallback(packet->GetWritableData(),
-                                       packet->GetBufferSize(),
-                                       AudioBuffersState(buffer_delay,
-                                                         hardware_delay));
+    // Start the stream.
+    operation = pa_stream_cork(playback_handle_, 0, NULL, NULL);
+    if (!operation) {
+      DLOG(ERROR) << "PulseAudioOutputStream: failed to start the playout "
+                  << "stream";
+      return;
+    }
+    pa_operation_unref(operation);
 
-  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;
-}
-
-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;
-
-  // Request more data from the source until we can fulfill the request or
-  // fail to receive anymore data.
-  bool buffering_successful = true;
-  while (client_buffer_->forward_bytes() < requested_bytes &&
-         buffering_successful) {
-    buffering_successful = BufferPacketFromSource();
+    operation = pa_stream_trigger(playback_handle_, NULL, NULL);
+    if (!operation) {
+      DLOG(ERROR) << "PulseAudioOutputStream: failed to trigger the playout "
+                  << "callback";
+      return;
+    }
+    pa_operation_unref(operation);
   }
 
-  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);
-  }
+  source_callback_ = callback;
 
-  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.
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::FulfillWriteRequest,
-        weak_factory_.GetWeakPtr(),
-        requested_bytes - bytes_written));
-  } else {
-    // Continue playback.
-    message_loop_->PostTask(FROM_HERE, base::Bind(
-        &PulseAudioOutputStream::WaitForWriteRequest,
-        weak_factory_.GetWeakPtr()));
-  }
+  // Before starting, the buffer might have audio from previous user of this
+  // device.
+  buffer_->Clear();
 }
 
-void PulseAudioOutputStream::WriteToStream(size_t bytes_to_write,
-                                           size_t* bytes_written) {
-  *bytes_written = 0;
-  while (*bytes_written < bytes_to_write) {
-    const uint8* chunk;
-    size_t chunk_size;
+void PulseAudioOutputStream::Stop() {
+  DCHECK_EQ(message_loop_, MessageLoop::current());
+  // Set the flag to false to stop filling new data to soundcard.
+  stream_stopped_ = true;
 
-    // Stop writing if there is no more data available.
-    if (!client_buffer_->GetCurrentChunk(&chunk, &chunk_size))
-      break;
+  if (!playback_handle_)
+    return;
 
-    // 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;
+  // Stop the stream.
+  pa_operation* operation = pa_stream_cork(playback_handle_, 1, NULL, NULL);
+  if (!operation) {
+    DLOG(ERROR) << "PulseAudioOutputStream: failed to stop the playout";
+    return;
   }
+  // Do not need to wait for the operation.
+  pa_operation_unref(operation);
 }
 
-void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
+void PulseAudioOutputStream::SetVolume(double volume) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
-  CHECK(callback);
-  source_callback_ = callback;
-
-  // Clear buffer, it might still have data in it.
-  client_buffer_->Clear();
-  stream_stopped_ = false;
-
-  // Start playback.
-  message_loop_->PostTask(FROM_HERE, base::Bind(
-      &PulseAudioOutputStream::WaitForWriteRequest,
-      weak_factory_.GetWeakPtr()));
+  volume_ = static_cast<float>(volume);
 }
 
-void PulseAudioOutputStream::Stop() {
+void PulseAudioOutputStream::GetVolume(double* volume) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
-  stream_stopped_ = true;
+  *volume = volume_;
 }
 
-void PulseAudioOutputStream::SetVolume(double volume) {
-  DCHECK_EQ(message_loop_, MessageLoop::current());
+void PulseAudioOutputStream::FulfillWriteRequest(size_t requested_bytes) {
+  // Update the delay.
+  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_);
+  uint32 buffer_delay = buffer_->forward_bytes();
+  // TODO(slock): Deal with negative latency (negative == 1).  This has yet
+  // to happen in practice though.
 
-  volume_ = static_cast<float>(volume);
+  // Request more data from the source until we can fulfill the request or
+  // fail to receive anymore data.
+  scoped_refptr<media::DataBuffer> packet(new media::DataBuffer(packet_size_));
+  size_t filled = 0;
+  int bytes_to_fill = requested_bytes;
+
+  // Request more data only if we need more.
+  if (!buffer_->forward_bytes() && bytes_to_fill) {
+    if (!stream_stopped_ && source_callback_)
+      filled = source_callback_->OnMoreData(
+          this,
+          packet->GetWritableData(),
+          packet->GetBufferSize(),
+          AudioBuffersState(buffer_delay, hardware_delay));
+    if (filled) {
+      packet->SetDataSize(filled);
+      buffer_->Append(packet);
+    }
+  }
+
+  const uint8* buffer_data;
+  size_t buffer_size;
+  if (buffer_->GetCurrentChunk(&buffer_data, &buffer_size)) {
+    if (buffer_size < static_cast<unsigned int>(bytes_to_fill))
+      filled = buffer_size;
+    else
+      filled = bytes_to_fill;
+     // Write data to stream.
+    if (pa_stream_write(playback_handle_, buffer_data, filled,
+                        NULL, 0, PA_SEEK_RELATIVE)) {
+      DLOG(WARNING) << "FulfillWriteRequest: failed to write "
+                    << filled << " bytes of data";
+    }
+     // Seek forward in the buffer after we've written some data to ALSA.
+    buffer_->Seek(filled);
+    bytes_to_fill -= filled;
+  }
+
+  size_t avialable_space = pa_stream_writable_size(playback_handle_);
+  if (avialable_space >= static_cast<size_t>(packet_size_))
+    FulfillWriteRequest(avialable_space);
 }
 
-void PulseAudioOutputStream::GetVolume(double* volume) {
+void PulseAudioOutputStream::Reset() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  stream_stopped_ = true;
 
-  *volume = volume_;
-}
+  pa_threaded_mainloop_lock(pa_mainloop_);
+  // Close the stream.
+  if (playback_handle_) {
+    // Disable all the callbacks before disconnecting.
+    pa_stream_set_state_callback(playback_handle_, NULL, NULL);
 
-uint32 PulseAudioOutputStream::RunDataCallback(
-    uint8* dest, uint32 max_size, AudioBuffersState buffers_state) {
-  if (source_callback_)
-    return source_callback_->OnMoreData(this, dest, max_size, buffers_state);
+    pa_stream_flush(playback_handle_, NULL, NULL);
+    pa_stream_disconnect(playback_handle_);
 
-  return 0;
+    // Release PulseAudio structures.
+    pa_stream_unref(playback_handle_);
+    playback_handle_ = NULL;
+  }
+  if (pa_context_) {
+    pa_context_unref(pa_context_);
+    pa_context_ = NULL;
+  }
+  pa_threaded_mainloop_unlock(pa_mainloop_);
+  if (pa_mainloop_) {
+    pa_threaded_mainloop_free(pa_mainloop_);
+    pa_mainloop_ = NULL;
+  }
 }