Added the unified IO on Linux.

This patch uses PulseAudio threaded mainloop to add a unified input and output implementation to Linux.
The implementation is used only by webaudio live audio.

BUG=145092
TEST=http://www.corp.google.com/~henrika/no_crawl/mic_effect_send_to_peer.html
http://chromium.googlecode.com/svn/trunk/samples/audio/visualizer-live.html

Review URL: https://chromiumcodereview.appspot.com/13031007

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@191247 0039d316-1c4b-4281-b951-d872f2087c98

8 files changed, 606 insertions, 194 deletions

diff --git a/media/audio/pulse/audio_manager_pulse.cc b/media/audio/pulse/audio_manager_pulse.cc
index 758d227..c127d17 100644
--- a/media/audio/pulse/audio_manager_pulse.cc
+++ b/media/audio/pulse/audio_manager_pulse.cc
@@ -16,6 +16,7 @@
 #include "media/audio/linux/audio_manager_linux.h"
 #include "media/audio/pulse/pulse_input.h"
 #include "media/audio/pulse/pulse_output.h"
+#include "media/audio/pulse/pulse_unified.h"
 #include "media/audio/pulse/pulse_util.h"
 #include "media/base/channel_layout.h"
 
@@ -158,6 +159,10 @@ AudioParameters AudioManagerPulse::GetPreferredOutputStreamParameters(
 
 AudioOutputStream* AudioManagerPulse::MakeOutputStream(
     const AudioParameters& params) {
+  if (params.input_channels()) {
+    return new PulseAudioUnifiedStream(params, this);
+  }
+
   return new PulseAudioOutputStream(params, this);
 }
 
diff --git a/media/audio/pulse/pulse_input.cc b/media/audio/pulse/pulse_input.cc
index 791ce7c..4e48e10 100644
--- a/media/audio/pulse/pulse_input.cc
+++ b/media/audio/pulse/pulse_input.cc
@@ -45,70 +45,12 @@ PulseAudioInputStream::~PulseAudioInputStream() {
 bool PulseAudioInputStream::Open() {
   DCHECK(thread_checker_.CalledOnValidThread());
   AutoPulseLock auto_lock(pa_mainloop_);
-
-  // Set sample specifications.
-  pa_sample_spec pa_sample_specifications;
-  pa_sample_specifications.format = pulse::BitsToPASampleFormat(
-      params_.bits_per_sample());
-  pa_sample_specifications.rate = params_.sample_rate();
-  pa_sample_specifications.channels = params_.channels();
-
-  // Get channel mapping and open recording stream.
-  pa_channel_map source_channel_map = pulse::ChannelLayoutToPAChannelMap(
-      params_.channel_layout());
-  pa_channel_map* map = (source_channel_map.channels != 0) ?
-      &source_channel_map : NULL;
-
-  // Create a new recording stream.
-  handle_ = pa_stream_new(pa_context_, "RecordStream",
-                          &pa_sample_specifications, map);
-  if (!handle_) {
-    DLOG(ERROR) << "Open: failed to create PA stream";
-    return false;
-  }
-
-  pa_stream_set_state_callback(handle_, &StreamNotifyCallback, this);
-
-  // Set server-side capture 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 buffer_attributes;
-  const unsigned int buffer_size = params_.GetBytesPerBuffer();
-  buffer_attributes.maxlength = static_cast<uint32_t>(-1);
-  buffer_attributes.tlength = buffer_size;
-  buffer_attributes.minreq = buffer_size;
-  buffer_attributes.prebuf = static_cast<uint32_t>(-1);
-  buffer_attributes.fragsize = buffer_size;
-  int flags = PA_STREAM_AUTO_TIMING_UPDATE |
-              PA_STREAM_INTERPOLATE_TIMING |
-              PA_STREAM_ADJUST_LATENCY |
-              PA_STREAM_START_CORKED;
-  int err = pa_stream_connect_record(
-      handle_,
-      device_name_ == AudioManagerBase::kDefaultDeviceId ?
-          NULL : device_name_.c_str(),
-      &buffer_attributes,
-      static_cast<pa_stream_flags_t>(flags));
-  if (err) {
-    DLOG(ERROR) << "pa_stream_connect_playback FAILED " << err;
+  if (!pulse::CreateInputStream(pa_mainloop_, pa_context_, &handle_, params_,
+                                device_name_, &StreamNotifyCallback, this)) {
     return false;
   }
 
-  // Wait for the stream to be ready.
-  while (true) {
-    pa_stream_state_t stream_state = pa_stream_get_state(handle_);
-    if (!PA_STREAM_IS_GOOD(stream_state)) {
-      DLOG(ERROR) << "Invalid PulseAudio stream state";
-      return false;
-    }
-
-    if (stream_state == PA_STREAM_READY)
-      break;
-    pa_threaded_mainloop_wait(pa_mainloop_);
-  }
-
-  pa_stream_set_read_callback(handle_, &ReadCallback, this);
-  pa_stream_readable_size(handle_);
+  DCHECK(handle_);
 
   buffer_.reset(new media::SeekableBuffer(0, 2 * params_.GetBytesPerBuffer()));
   audio_data_buffer_.reset(new uint8[params_.GetBytesPerBuffer()]);
@@ -129,8 +71,10 @@ void PulseAudioInputStream::Start(AudioInputCallback* callback) {
   buffer_->Clear();
 
   // Start the streaming.
-  stream_started_ = true;
   callback_ = callback;
+  pa_stream_set_read_callback(handle_, &ReadCallback, this);
+  pa_stream_readable_size(handle_);
+  stream_started_ = true;
 
   pa_operation* operation = pa_stream_cork(handle_, 0, NULL, NULL);
   WaitForOperationCompletion(pa_mainloop_, operation);
diff --git a/media/audio/pulse/pulse_output.cc b/media/audio/pulse/pulse_output.cc
index 1a4b275..eccf463 100644
--- a/media/audio/pulse/pulse_output.cc
+++ b/media/audio/pulse/pulse_output.cc
@@ -17,22 +17,6 @@ namespace media {
 using pulse::AutoPulseLock;
 using pulse::WaitForOperationCompletion;
 
-// 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_threaded_mainloop_signal(stream->pa_mainloop_, 0);
-}
-
 // static, pa_stream_notify_cb
 void PulseAudioOutputStream::StreamNotifyCallback(pa_stream* s, void* p_this) {
   PulseAudioOutputStream* stream = static_cast<PulseAudioOutputStream*>(p_this);
@@ -78,121 +62,13 @@ PulseAudioOutputStream::~PulseAudioOutputStream() {
   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());
-
-  pa_mainloop_ = pa_threaded_mainloop_new();
-  RETURN_ON_FAILURE(pa_mainloop_, "Failed to create PulseAudio main loop.");
-
-  pa_mainloop_api* pa_mainloop_api = pa_threaded_mainloop_get_api(pa_mainloop_);
-  pa_context_ = pa_context_new(pa_mainloop_api, "Chromium");
-  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 = pulse::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 = pulse::ChannelLayoutToPAChannelMap(
-      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;
-  }
-  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);
-
-  // 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);
-
-  // Pulse is very finicky with the small buffer sizes used by Chrome.  The
-  // settings below are mostly found through trial and error.  Essentially we
-  // want Pulse to auto size its internal buffers, but call us back nearly every
-  // |minreq| bytes.  |tlength| should be a multiple of |minreq|; too low and
-  // Pulse will issue callbacks way too fast, too high and we don't get
-  // callbacks frequently enough.
-  pa_buffer_attr pa_buffer_attributes;
-  pa_buffer_attributes.maxlength = static_cast<uint32_t>(-1);
-  pa_buffer_attributes.minreq = params_.GetBytesPerBuffer();
-  pa_buffer_attributes.prebuf = static_cast<uint32_t>(-1);
-  pa_buffer_attributes.tlength = params_.GetBytesPerBuffer() * 3;
-  pa_buffer_attributes.fragsize = static_cast<uint32_t>(-1);
-
-  // Connect playback stream.  Like pa_buffer_attr, the pa_stream_flags have a
-  // huge impact on the performance of the stream and were chosen through trial
-  // and error.
-  RETURN_ON_FAILURE(
-      pa_stream_connect_playback(
-          pa_stream_, NULL, &pa_buffer_attributes,
-          static_cast<pa_stream_flags_t>(
-              PA_STREAM_FIX_RATE | PA_STREAM_INTERPOLATE_TIMING |
-              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;
+  return pulse::CreateOutputStream(&pa_mainloop_, &pa_context_, &pa_stream_,
+                                   params_, &StreamNotifyCallback,
+                                   &StreamRequestCallback, this);
 }
 
-#undef RETURN_ON_FAILURE
-
 void PulseAudioOutputStream::Reset() {
   if (!pa_mainloop_) {
     DCHECK(!pa_stream_);
diff --git a/media/audio/pulse/pulse_output.h b/media/audio/pulse/pulse_output.h
index 114e1b5..583cce7 100644
--- a/media/audio/pulse/pulse_output.h
+++ b/media/audio/pulse/pulse_output.h
@@ -48,10 +48,9 @@ class PulseAudioOutputStream : public AudioOutputStream {
   virtual void GetVolume(double* volume) OVERRIDE;
 
  private:
-  // Called by PulseAudio when |pa_context_| and |pa_stream_| change state.  If
-  // an unexpected failure state change happens and |source_callback_| is set
-  // these methods will forward the error via OnError().
-  static void ContextNotifyCallback(pa_context* c, void* p_this);
+  // Called by PulseAudio when |pa_stream_| change state.  If an unexpected
+  // failure state change happens and |source_callback_| is set
+  // this method will forward the error via OnError().
   static void StreamNotifyCallback(pa_stream* s, void* p_this);
 
   // Called by PulseAudio when it needs more audio data.
diff --git a/media/audio/pulse/pulse_unified.cc b/media/audio/pulse/pulse_unified.cc
new file mode 100644
index 0000000..791a3a8
--- /dev/null
+++ b/media/audio/pulse/pulse_unified.cc
@@ -0,0 +1,293 @@
+// Copyright 2013 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_unified.h"
+
+#include "base/message_loop.h"
+#include "base/time.h"
+#include "media/audio/audio_manager_base.h"
+#include "media/audio/audio_parameters.h"
+#include "media/audio/audio_util.h"
+#include "media/audio/pulse/pulse_util.h"
+#include "media/base/seekable_buffer.h"
+
+namespace media {
+
+using pulse::AutoPulseLock;
+using pulse::WaitForOperationCompletion;
+
+static const int kFifoSizeInPackets = 10;
+
+// static, pa_stream_notify_cb
+void PulseAudioUnifiedStream::StreamNotifyCallback(pa_stream* s,
+                                                   void* user_data) {
+  PulseAudioUnifiedStream* stream =
+      static_cast<PulseAudioUnifiedStream*>(user_data);
+
+  // 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_threaded_mainloop_signal(stream->pa_mainloop_, 0);
+}
+
+// static, used by pa_stream_set_read_callback.
+void PulseAudioUnifiedStream::ReadCallback(pa_stream* handle, size_t length,
+                                           void* user_data) {
+  static_cast<PulseAudioUnifiedStream*>(user_data)->ReadData();
+}
+
+PulseAudioUnifiedStream::PulseAudioUnifiedStream(const AudioParameters& params,
+                                                 AudioManagerBase* manager)
+    : params_(params),
+      manager_(manager),
+      pa_context_(NULL),
+      pa_mainloop_(NULL),
+      input_stream_(NULL),
+      output_stream_(NULL),
+      volume_(1.0f),
+      source_callback_(NULL) {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+  CHECK(params_.IsValid());
+  input_bus_ = AudioBus::Create(params_);
+  output_bus_ = AudioBus::Create(params_);
+}
+
+PulseAudioUnifiedStream::~PulseAudioUnifiedStream() {
+  // All internal structures should already have been freed in Close(), which
+  // calls AudioManagerBase::ReleaseOutputStream() which deletes this object.
+  DCHECK(!input_stream_);
+  DCHECK(!output_stream_);
+  DCHECK(!pa_context_);
+  DCHECK(!pa_mainloop_);
+}
+
+bool PulseAudioUnifiedStream::Open() {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+  // Prepare the recording buffers for the callbacks.
+  fifo_.reset(new media::SeekableBuffer(
+      0, kFifoSizeInPackets * params_.GetBytesPerBuffer()));
+  input_data_buffer_.reset(new uint8[params_.GetBytesPerBuffer()]);
+
+  if (!pulse::CreateOutputStream(&pa_mainloop_, &pa_context_, &output_stream_,
+                                 params_, &StreamNotifyCallback, NULL, this))
+    return false;
+
+  // TODO(xians): Add support for non-default device.
+  if (!pulse::CreateInputStream(pa_mainloop_, pa_context_, &input_stream_,
+                                params_, AudioManagerBase::kDefaultDeviceId,
+                                &StreamNotifyCallback, this))
+    return false;
+
+  DCHECK(pa_mainloop_);
+  DCHECK(pa_context_);
+  DCHECK(input_stream_);
+  DCHECK(output_stream_);
+  return true;
+}
+
+void PulseAudioUnifiedStream::Reset() {
+  if (!pa_mainloop_) {
+    DCHECK(!input_stream_);
+    DCHECK(!output_stream_);
+    DCHECK(!pa_context_);
+    return;
+  }
+
+  {
+    AutoPulseLock auto_lock(pa_mainloop_);
+
+    // Close the input stream.
+    if (input_stream_) {
+      // Disable all the callbacks before disconnecting.
+      pa_stream_set_state_callback(input_stream_, NULL, NULL);
+      pa_stream_flush(input_stream_, NULL, NULL);
+      pa_stream_disconnect(input_stream_);
+
+      // Release PulseAudio structures.
+      pa_stream_unref(input_stream_);
+      input_stream_ = NULL;
+    }
+
+    // Close the ouput stream.
+    if (output_stream_) {
+      // Release PulseAudio output stream structures.
+      pa_stream_set_state_callback(output_stream_, NULL, NULL);
+      pa_stream_disconnect(output_stream_);
+      pa_stream_unref(output_stream_);
+      output_stream_ = 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;
+    }
+  }
+
+  pa_threaded_mainloop_stop(pa_mainloop_);
+  pa_threaded_mainloop_free(pa_mainloop_);
+  pa_mainloop_ = NULL;
+}
+
+void PulseAudioUnifiedStream::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 PulseAudioUnifiedStream::WriteData(size_t requested_bytes) {
+  CHECK_EQ(requested_bytes, static_cast<size_t>(params_.GetBytesPerBuffer()));
+
+  void* buffer = NULL;
+  int frames_filled = 0;
+  if (source_callback_) {
+    CHECK_GE(pa_stream_begin_write(
+        output_stream_, &buffer, &requested_bytes), 0);
+    uint32 hardware_delay = pulse::GetHardwareLatencyInBytes(
+        output_stream_, params_.sample_rate(),
+        params_.GetBytesPerFrame());
+    source_callback_->WaitTillDataReady();
+    fifo_->Read(input_data_buffer_.get(), requested_bytes);
+    input_bus_->FromInterleaved(
+        input_data_buffer_.get(), params_.frames_per_buffer(), 2);
+
+    frames_filled = source_callback_->OnMoreIOData(
+        input_bus_.get(),
+        output_bus_.get(),
+        AudioBuffersState(0, hardware_delay));
+  }
+
+  // Zero the unfilled data so it plays back as silence.
+  if (frames_filled < output_bus_->frames()) {
+    output_bus_->ZeroFramesPartial(
+        frames_filled, output_bus_->frames() - frames_filled);
+  }
+
+  // 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.
+  output_bus_->ToInterleaved(
+      output_bus_->frames(), params_.bits_per_sample() / 8, buffer);
+  media::AdjustVolume(buffer, requested_bytes, params_.channels(),
+                      params_.bits_per_sample() / 8, volume_);
+
+  if (pa_stream_write(output_stream_, buffer, requested_bytes, NULL, 0LL,
+                      PA_SEEK_RELATIVE) < 0) {
+    if (source_callback_) {
+      source_callback_->OnError(this);
+    }
+  }
+}
+
+void PulseAudioUnifiedStream::ReadData() {
+  do {
+    size_t length = 0;
+    const void* data = NULL;
+    pa_stream_peek(input_stream_, &data, &length);
+    if (!data || length == 0)
+      break;
+
+    fifo_->Append(reinterpret_cast<const uint8*>(data), length);
+
+    // Deliver the recording data to the renderer and drive the playout.
+    int packet_size = params_.GetBytesPerBuffer();
+    while (fifo_->forward_bytes() >= packet_size) {
+      WriteData(packet_size);
+    }
+
+    // Checks if we still have data.
+    pa_stream_drop(input_stream_);
+  } while (pa_stream_readable_size(input_stream_) > 0);
+
+  pa_threaded_mainloop_signal(pa_mainloop_, 0);
+}
+
+void PulseAudioUnifiedStream::Start(AudioSourceCallback* callback) {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+  CHECK(callback);
+  CHECK(input_stream_);
+  CHECK(output_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(output_stream_) != PA_STREAM_READY &&
+      pa_stream_get_state(input_stream_) != PA_STREAM_READY) {
+    callback->OnError(this);
+    return;
+  }
+
+  source_callback_ = callback;
+
+  fifo_->Clear();
+
+  // Uncork (resume) the input stream.
+  pa_stream_set_read_callback(input_stream_, &ReadCallback, this);
+  pa_stream_readable_size(input_stream_);
+  pa_operation* operation = pa_stream_cork(input_stream_, 0, NULL, NULL);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+
+  // Uncork (resume) the output stream.
+  // We use the recording stream to drive the playback, so we do not need to
+  // register the write callback using pa_stream_set_write_callback().
+  operation = pa_stream_cork(output_stream_, 0,
+                             &pulse::StreamSuccessCallback, pa_mainloop_);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+}
+
+void PulseAudioUnifiedStream::Stop() {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+
+  // Cork (pause) the stream.  Waiting for the main loop lock will ensure
+  // outstanding callbacks have completed.
+  AutoPulseLock auto_lock(pa_mainloop_);
+
+  // Set |source_callback_| to NULL so all FulfillWriteRequest() calls which may
+  // occur while waiting on the flush and cork exit immediately.
+  source_callback_ = NULL;
+
+  // Set the read callback to NULL before flushing the stream, otherwise it
+  // will cause deadlock on the operation.
+  pa_stream_set_read_callback(input_stream_, NULL, NULL);
+  pa_operation* operation = pa_stream_flush(
+      input_stream_, &pulse::StreamSuccessCallback, pa_mainloop_);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+
+  operation = pa_stream_cork(input_stream_, 1, &pulse::StreamSuccessCallback,
+                             pa_mainloop_);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+
+  // 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.
+  operation = pa_stream_flush(
+      output_stream_, &pulse::StreamSuccessCallback, pa_mainloop_);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+
+  operation = pa_stream_cork(output_stream_, 1, &pulse::StreamSuccessCallback,
+                             pa_mainloop_);
+  WaitForOperationCompletion(pa_mainloop_, operation);
+}
+
+void PulseAudioUnifiedStream::SetVolume(double volume) {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+
+  volume_ = static_cast<float>(volume);
+}
+
+void PulseAudioUnifiedStream::GetVolume(double* volume) {
+  DCHECK(manager_->GetMessageLoop()->BelongsToCurrentThread());
+
+  *volume = volume_;
+}
+
+}  // namespace media
diff --git a/media/audio/pulse/pulse_unified.h b/media/audio/pulse/pulse_unified.h
new file mode 100644
index 0000000..0d52a13
--- /dev/null
+++ b/media/audio/pulse/pulse_unified.h
@@ -0,0 +1,85 @@
+// Copyright 2013 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.
+
+#ifndef MEDIA_AUDIO_PULSE_PULSE_UNIFIED_H_
+#define MEDIA_AUDIO_PULSE_PULSE_UNIFIED_H_
+
+#include <pulse/pulseaudio.h>
+
+#include "base/memory/scoped_ptr.h"
+#include "media/audio/audio_io.h"
+#include "media/audio/audio_parameters.h"
+#include "media/base/audio_fifo.h"
+
+namespace media {
+
+class AudioManagerBase;
+class SeekableBuffer;
+
+class PulseAudioUnifiedStream : public AudioOutputStream {
+ public:
+  PulseAudioUnifiedStream(const AudioParameters& params,
+                          AudioManagerBase* manager);
+
+  virtual ~PulseAudioUnifiedStream();
+
+  // Implementation of PulseAudioUnifiedStream.
+  virtual bool Open() OVERRIDE;
+  virtual void Close() OVERRIDE;
+  virtual void Start(AudioSourceCallback* callback) OVERRIDE;
+  virtual void Stop() OVERRIDE;
+  virtual void SetVolume(double volume) OVERRIDE;
+  virtual void GetVolume(double* volume) OVERRIDE;
+
+ private:
+  // Called by PulseAudio when |pa_stream_| change state.  If an unexpected
+  // failure state change happens and |source_callback_| is set
+  // this method will forward the error via OnError().
+  static void StreamNotifyCallback(pa_stream* s, void* user_data);
+
+  // Called by PulseAudio recording stream when it has data.
+  static void ReadCallback(pa_stream* s, size_t length, void* user_data);
+
+  // Helpers for ReadCallback() to read and write data.
+  void WriteData(size_t requested_bytes);
+  void ReadData();
+
+  // Close() helper function to free internal structs.
+  void Reset();
+
+  // AudioParameters from the constructor.
+  const AudioParameters params_;
+
+  // Audio manager that created us.  Used to report that we've closed.
+  AudioManagerBase* manager_;
+
+  // PulseAudio API structs.
+  pa_context* pa_context_;
+  pa_threaded_mainloop* pa_mainloop_;
+  pa_stream* input_stream_;
+  pa_stream* output_stream_;
+
+  // Float representation of volume from 0.0 to 1.0.
+  float volume_;
+
+  // Callback to audio data source.  Must only be modified while holding a lock
+  // on |pa_mainloop_| via pa_threaded_mainloop_lock().
+  AudioSourceCallback* source_callback_;
+
+  scoped_ptr<AudioBus> input_bus_;
+  scoped_ptr<AudioBus> output_bus_;
+
+  // Used for input to output buffering.
+  scoped_ptr<media::SeekableBuffer> fifo_;
+
+  // Temporary storage for recorded data. It gets a packet of data from
+  // |fifo_| and deliver the data to OnMoreIOData() callback.
+  scoped_array<uint8> input_data_buffer_;
+
+  DISALLOW_COPY_AND_ASSIGN(PulseAudioUnifiedStream);
+};
+
+}  // namespace media
+
+#endif  // MEDIA_AUDIO_PULSE_PULSE_UNIFIED_H_
diff --git a/media/audio/pulse/pulse_util.cc b/media/audio/pulse/pulse_util.cc
index b791569..0a4734b 100644
--- a/media/audio/pulse/pulse_util.cc
+++ b/media/audio/pulse/pulse_util.cc
@@ -7,6 +7,7 @@
 #include "base/logging.h"
 #include "base/time.h"
 #include "media/audio/audio_manager_base.h"
+#include "media/audio/audio_parameters.h"
 
 namespace media {
 
@@ -126,6 +127,189 @@ int GetHardwareLatencyInBytes(pa_stream* stream,
       base::Time::kMicrosecondsPerSecond;
 }
 
+// Helper macro for CreateInput/OutputStream() to avoid code spam and
+// string bloat.
+#define RETURN_ON_FAILURE(expression, message) do { \
+  if (!(expression)) { \
+    DLOG(ERROR) << message; \
+    return false; \
+  } \
+} while(0)
+
+bool CreateInputStream(pa_threaded_mainloop* mainloop,
+                       pa_context* context,
+                       pa_stream** stream,
+                       const AudioParameters& params,
+                       const std::string& device_id,
+                       pa_stream_notify_cb_t stream_callback,
+                       void* user_data) {
+  DCHECK(mainloop);
+  DCHECK(context);
+
+  // Set sample specifications.
+  pa_sample_spec sample_specifications;
+  sample_specifications.format = BitsToPASampleFormat(
+      params.bits_per_sample());
+  sample_specifications.rate = params.sample_rate();
+  sample_specifications.channels = params.channels();
+
+  // Get channel mapping and open recording stream.
+  pa_channel_map source_channel_map = ChannelLayoutToPAChannelMap(
+      params.channel_layout());
+  pa_channel_map* map = (source_channel_map.channels != 0) ?
+      &source_channel_map : NULL;
+
+  // Create a new recording stream.
+  *stream = pa_stream_new(context, "RecordStream", &sample_specifications, map);
+  RETURN_ON_FAILURE(*stream, "failed to create PA recording stream");
+
+  pa_stream_set_state_callback(*stream, stream_callback, user_data);
+
+  // Set server-side capture 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 buffer_attributes;
+  const unsigned int buffer_size = params.GetBytesPerBuffer();
+  buffer_attributes.maxlength = static_cast<uint32_t>(-1);
+  buffer_attributes.tlength = buffer_size;
+  buffer_attributes.minreq = buffer_size;
+  buffer_attributes.prebuf = static_cast<uint32_t>(-1);
+  buffer_attributes.fragsize = buffer_size;
+  int flags = PA_STREAM_AUTO_TIMING_UPDATE |
+              PA_STREAM_INTERPOLATE_TIMING |
+              PA_STREAM_ADJUST_LATENCY |
+              PA_STREAM_START_CORKED;
+  RETURN_ON_FAILURE(
+      pa_stream_connect_record(
+          *stream,
+          device_id == AudioManagerBase::kDefaultDeviceId ?
+              NULL : device_id.c_str(),
+          &buffer_attributes,
+          static_cast<pa_stream_flags_t>(flags)) == 0,
+      "pa_stream_connect_record FAILED ");
+
+  // Wait for the stream to be ready.
+  while (true) {
+    pa_stream_state_t stream_state = pa_stream_get_state(*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(mainloop);
+  }
+
+  return true;
+}
+
+bool CreateOutputStream(pa_threaded_mainloop** mainloop,
+                        pa_context** context,
+                        pa_stream** stream,
+                        const AudioParameters& params,
+                        pa_stream_notify_cb_t stream_callback,
+                        pa_stream_request_cb_t write_callback,
+                        void* user_data) {
+  DCHECK(!*mainloop);
+  DCHECK(!*context);
+
+  *mainloop = pa_threaded_mainloop_new();
+  RETURN_ON_FAILURE(*mainloop, "Failed to create PulseAudio main loop.");
+
+  pa_mainloop_api* pa_mainloop_api = pa_threaded_mainloop_get_api(*mainloop);
+  *context = pa_context_new(pa_mainloop_api, "Chromium");
+  RETURN_ON_FAILURE(*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(*context, &ContextStateCallback, *mainloop);
+
+  // 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(*mainloop);
+
+  RETURN_ON_FAILURE(pa_threaded_mainloop_start(*mainloop) == 0,
+                    "Failed to start PulseAudio main loop.");
+  RETURN_ON_FAILURE(
+      pa_context_connect(*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(*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(*mainloop);
+  }
+
+  // Set sample specifications.
+  pa_sample_spec sample_specifications;
+  sample_specifications.format = BitsToPASampleFormat(
+      params.bits_per_sample());
+  sample_specifications.rate = params.sample_rate();
+  sample_specifications.channels = params.channels();
+
+  // Get channel mapping and open playback stream.
+  pa_channel_map* map = NULL;
+  pa_channel_map source_channel_map = ChannelLayoutToPAChannelMap(
+      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;
+  }
+  *stream = pa_stream_new(*context, "Playback", &sample_specifications, map);
+  RETURN_ON_FAILURE(*stream, "failed to create PA playback stream");
+
+  pa_stream_set_state_callback(*stream, stream_callback, user_data);
+
+  // Even though we start the stream corked above, PulseAudio will issue one
+  // stream request after setup.  write_callback() must fulfill the write.
+  pa_stream_set_write_callback(*stream, write_callback, user_data);
+
+  // Pulse is very finicky with the small buffer sizes used by Chrome.  The
+  // settings below are mostly found through trial and error.  Essentially we
+  // want Pulse to auto size its internal buffers, but call us back nearly every
+  // |minreq| bytes.  |tlength| should be a multiple of |minreq|; too low and
+  // Pulse will issue callbacks way too fast, too high and we don't get
+  // callbacks frequently enough.
+  pa_buffer_attr pa_buffer_attributes;
+  pa_buffer_attributes.maxlength = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.minreq = params.GetBytesPerBuffer();
+  pa_buffer_attributes.prebuf = static_cast<uint32_t>(-1);
+  pa_buffer_attributes.tlength = params.GetBytesPerBuffer() * 3;
+  pa_buffer_attributes.fragsize = static_cast<uint32_t>(-1);
+
+  // Connect playback stream.  Like pa_buffer_attr, the pa_stream_flags have a
+  // huge impact on the performance of the stream and were chosen through trial
+  // and error.
+  RETURN_ON_FAILURE(
+      pa_stream_connect_playback(
+          *stream, NULL, &pa_buffer_attributes,
+          static_cast<pa_stream_flags_t>(
+              PA_STREAM_FIX_RATE | PA_STREAM_INTERPOLATE_TIMING |
+              PA_STREAM_ADJUST_LATENCY | PA_STREAM_AUTO_TIMING_UPDATE |
+              PA_STREAM_NOT_MONOTONIC | PA_STREAM_START_CORKED),
+          NULL, NULL) == 0,
+      "pa_stream_connect_playback FAILED ");
+
+  // Wait for the stream to be ready.
+  while (true) {
+    pa_stream_state_t stream_state = pa_stream_get_state(*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(*mainloop);
+  }
+
+  return true;
+}
+
+#undef RETURN_ON_FAILURE
+
 }  // namespace pulse
 
 }  // namespace media
diff --git a/media/audio/pulse/pulse_util.h b/media/audio/pulse/pulse_util.h
index 6443f3c..da0cb0f 100644
--- a/media/audio/pulse/pulse_util.h
+++ b/media/audio/pulse/pulse_util.h
@@ -13,6 +13,8 @@
 
 namespace media {
 
+class AudioParameters;
+
 namespace pulse {
 
 // A helper class that acquires pa_threaded_mainloop_lock() while in scope.
@@ -40,13 +42,37 @@ pa_sample_format_t BitsToPASampleFormat(int bits_per_sample);
 
 pa_channel_map ChannelLayoutToPAChannelMap(ChannelLayout channel_layout);
 
-void WaitForOperationCompletion(pa_threaded_mainloop* pa_mainloop,
+void WaitForOperationCompletion(pa_threaded_mainloop* mainloop,
                                 pa_operation* operation);
 
 int GetHardwareLatencyInBytes(pa_stream* stream,
                               int sample_rate,
                               int bytes_per_frame);
 
+// Create a recording stream for the threaded mainloop, return true if success,
+// otherwise false. |mainloop| and |context| have to be from a valid Pulse
+// threaded mainloop and the handle of the created stream will be returned by
+// |stream|.
+bool CreateInputStream(pa_threaded_mainloop* mainloop,
+                       pa_context* context,
+                       pa_stream** stream,
+                       const AudioParameters& params,
+                       const std::string& device_id,
+                       pa_stream_notify_cb_t stream_callback,
+                       void* user_data);
+
+// Create a playback stream for the threaded mainloop, return true if success,
+// otherwise false. This function will create a new Pulse threaded mainloop,
+// and the handles of the mainloop, context and stream will be returned by
+// |mainloop|, |context| and |stream|.
+bool CreateOutputStream(pa_threaded_mainloop** mainloop,
+                        pa_context** context,
+                        pa_stream** stream,
+                        const AudioParameters& params,
+                        pa_stream_notify_cb_t stream_callback,
+                        pa_stream_request_cb_t write_callback,
+                        void* user_data);
+
 }  // namespace pulse
 
 }  // namespace media