A patch making the pulseaudio working with the threaded mainloop.

media/audio/linux/pulse_output.cc

 // Copyright (c) 2011 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/linux/pulse_output.h"
 
 #include "base/bind.h"
 #include "base/message_loop.h"
+#include "base/synchronization/waitable_event.h"
 #include "media/audio/audio_parameters.h"
 #include "media/audio/audio_util.h"
 #include "media/audio/linux/audio_manager_linux.h"
 #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?

[tommi (sloooow) - chröme, 2011/11/09 13:57:58]

rather -> other

 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

[enal1, 2011/11/11 17:08:07]

I used to having comments before the related code...

     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:
       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:
-      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 /
       base::Time::kMicrosecondsPerSecond;
 }
 
 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) {
+  base::WaitableEvent* completion =
+      reinterpret_cast< base::WaitableEvent*>(user_data);
+  pa_context_state_t state = pa_context_get_state(context);
+  switch (state) {
+    case PA_CONTEXT_TERMINATED:
+      completion->Signal();
+      break;
+    case PA_CONTEXT_READY:
+      completion->Signal();
+      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,
                                                AudioManagerLinux* 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),
+      pa_glib_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),
       message_loop_(message_loop),
       ALLOW_THIS_IN_INITIALIZER_LIST(weak_factory_(this)),
       source_callback_(NULL) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
   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() {
   // All internal structures should already have been freed in Close(),
   // which calls AudioManagerLinux::Release which deletes this object.
   DCHECK(!playback_handle_);
   DCHECK(!pa_context_);
-  DCHECK(!pa_mainloop_);
+  DCHECK(!pa_glib_mainloop_);
 }
 
 bool PulseAudioOutputStream::Open() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  DCHECK(!pa_glib_mainloop_);
 
-  // TODO(slock): Possibly move most of this to an OpenPlaybackDevice function
-  // in a new class 'pulse_util', like alsa_util.
+  // Use glib mainloop that we don't need to care about any processing.
+  pa_glib_mainloop_ = pa_glib_mainloop_new(NULL);
+  DCHECK(pa_glib_mainloop_) << "Failed to create PA glib mainloop";
+  if (!pa_glib_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_);
+  // TODO(xians): Figure out if we can share one pa_context_ for streams.
+  pa_mainloop_api* pa_mainloop_api =
+      pa_glib_mainloop_get_api(pa_glib_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();
+    return false;
+  }
 
-  // Wait until PulseAudio is ready.
+  base::WaitableEvent state_changed(false, false);
   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;
-    }
+                                &state_changed);
+  if (pa_context_connect(pa_context_, NULL, PA_CONTEXT_NOFLAGS, NULL)) {
+    DLOG(ERROR) << "Failed to connect to the context";
+    Reset();
+    return false;
+  }
+
+  // Wait for state change.
+  const base::TimeDelta kMaxTimeOut = base::TimeDelta::FromMilliseconds(200);
+  if(!state_changed.TimedWait(kMaxTimeOut)) {
+    DLOG(ERROR) <<  "Timeout when waiting for context state change";
+    return false;
+  }
+
+  if (pa_context_get_state(pa_context_) != PA_CONTEXT_READY) {
+    DLOG(ERROR) <<  "Unknown problem connecting to PulseAudio server";
+    Reset();
+    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();
     return false;
   }
 
+  pa_stream_set_write_callback(playback_handle_, &WriteRequestCallback, this);
+  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());
 
   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_EQ(message_loop_, MessageLoop::current());
-
-  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_) {
-    message_loop_->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_);
-  size_t packet_size = RunDataCallback(packet->GetWritableData(),
-                                       packet->GetBufferSize(),
-                                       AudioBuffersState(buffer_delay,
-                                                         hardware_delay));
-
-  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();
-  }
-
-  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.
-    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()));
-  }
-}
-
-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;
-
-    // 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;
-  }
-}
-
 void PulseAudioOutputStream::Start(AudioSourceCallback* callback) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  CHECK(callback);
 
-  CHECK(callback);
+  if (!stream_stopped_)
+    return;
+  stream_stopped_ = false;
+
+  // 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>(0),
+                                         NULL,  // Default volume.
+                                         NULL  // Standalone stream.
+                                         );

[tommi (sloooow) - chröme, 2011/11/09 13:57:58]

just keep ); on the preceding line

+    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);
+
+    // 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);
+
+    operation = pa_stream_trigger(playback_handle_, NULL, NULL);
+    if (!operation) {
+      DLOG(ERROR) << "PulseAudioOutputStream: failed to trigger the playout "
+                  << "callback";
+      return;
+    }
+    pa_operation_unref(operation);
+  }
+
   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()));
+  // Before starting, the buffer might have audio from previous user of this
+  // device.
+  buffer_->Clear();
 }
 
 void PulseAudioOutputStream::Stop() {
   DCHECK_EQ(message_loop_, MessageLoop::current());
+  // Set the flag to false to stop filling new data to soundcard.
+  stream_stopped_ = true;
 
-  stream_stopped_ = true;
+  if (!playback_handle_)
+    return;
+
+  // 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::SetVolume(double volume) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
   volume_ = static_cast<float>(volume);
 }
 
 void PulseAudioOutputStream::GetVolume(double* volume) {
   DCHECK_EQ(message_loop_, MessageLoop::current());
 
   *volume = volume_;
 }
 
-uint32 PulseAudioOutputStream::RunDataCallback(
-    uint8* dest, uint32 max_size, AudioBuffersState buffers_state) {
-  if (source_callback_)
-    return source_callback_->OnMoreData(this, dest, max_size, buffers_state);
+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_);
+  // TODO(slock): Deal with negative latency (negative == 1).  This has yet
+  // to happen in practice though.
 
-  return 0;
+  // 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;
+
+  while (bytes_to_fill > 0) {
+    // Request more data if we have capacity.
+    if (!buffer_->forward_bytes() && bytes_to_fill) {
+      if (!stream_stopped_ && source_callback_)
+        filled = source_callback_->OnMoreData(
+            this,
+            packet->GetWritableData(),
+            packet->GetBufferSize(),
+            AudioBuffersState(0, 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;
+    }
+  }
 }
+
+void PulseAudioOutputStream::Reset() {
+  stream_stopped_ = true;
+
+  // Close the stream.
+  if (playback_handle_) {
+    // Disable all the callbacks before disconnecting.
+    pa_stream_set_state_callback(playback_handle_, NULL, NULL);
+
+    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_glib_mainloop_) {
+    pa_glib_mainloop_free(pa_glib_mainloop_);
+    pa_glib_mainloop_ = NULL;
+  }
+}