Bug 1238038 - Create an input-only audio source r=padenot

This patch creates an class managing an input-only audio stream within a
task thread. This class will be used as a source generating the audio
data to a NonNativeInputTrack in the following patch.

Depends on D116535

Differential Revision: https://phabricator.services.mozilla.com/D137911

dom/media/AudioInputSource.cpp

@@ -0,0 +1,211 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+#include "AudioInputSource.h"
+
+#include "AudioThreadRegistry.h"
+#include "GraphDriver.h"
+
+namespace mozilla {
+
+extern mozilla::LazyLogModule gMediaTrackGraphLog;
+
+#ifdef LOG_INTERNAL
+#  undef LOG_INTERNAL
+#endif  // LOG_INTERNAL
+#define LOG_INTERNAL(level, msg, ...) \
+  MOZ_LOG(gMediaTrackGraphLog, LogLevel::level, (msg, ##__VA_ARGS__))
+
+#ifdef LOG
+#  undef LOG
+#endif  // LOG
+#define LOG(msg, ...) LOG_INTERNAL(Debug, msg, ##__VA_ARGS__)
+
+#ifdef LOGW
+#  undef LOGW
+#endif  // LOGW
+#define LOGW(msg, ...) LOG_INTERNAL(Warning, msg, ##__VA_ARGS__)
+
+#ifdef LOGE
+#  undef LOGE
+#endif  // LOGE
+#define LOGE(msg, ...) LOG_INTERNAL(Error, msg, ##__VA_ARGS__)
+
+#ifdef LOGV
+#  undef LOGV
+#endif  // LOGV
+#define LOGV(msg, ...) LOG_INTERNAL(Verbose, msg, ##__VA_ARGS__)
+
+AudioInputSource::AudioInputSource(RefPtr<EventListener>&& aListener,
+                                   Id aSourceId,
+                                   CubebUtils::AudioDeviceID aDeviceId,
+                                   uint32_t aChannelCount, bool aIsVoice,
+                                   const PrincipalHandle& aPrincipalHandle,
+                                   TrackRate aSourceRate, TrackRate aTargetRate,
+                                   uint32_t aBufferMs)
+    : mId(aSourceId),
+      mDeviceId(aDeviceId),
+      mChannelCount(aChannelCount),
+      mRate(aSourceRate),
+      mIsVoice(aIsVoice),
+      mPrincipalHandle(aPrincipalHandle),
+      mSandboxed(CubebUtils::SandboxEnabled()),
+      mAudioThreadId(ProfilerThreadId{}),
+      mEventListener(std::move(aListener)),
+      mTaskThread(CUBEB_TASK_THREAD),
+      mDriftCorrector(static_cast<uint32_t>(aSourceRate),
+                      static_cast<uint32_t>(aTargetRate), aBufferMs,
+                      aPrincipalHandle) {
+  MOZ_ASSERT(mChannelCount > 0);
+  MOZ_ASSERT(mEventListener);
+}
+
+void AudioInputSource::Start() {
+  // This is called on MediaTrackGraph's graph thread, which can be the cubeb
+  // stream's callback thread. Running cubeb operations within cubeb stream
+  // callback thread can cause the deadlock on Linux, so we dispatch those
+  // operations to the task thread.
+  MOZ_ASSERT(mTaskThread);
+
+  LOG("AudioInputSource %p, start", this);
+  MOZ_ALWAYS_SUCCEEDS(mTaskThread->Dispatch(
+      NS_NewRunnableFunction(__func__, [self = RefPtr(this)]() mutable {
+        self->mStream = CubebInputStream::Create(
+            self->mDeviceId, self->mChannelCount,
+            static_cast<uint32_t>(self->mRate), self->mIsVoice, self.get());
+        if (!self->mStream) {
+          LOGE("AudioInputSource %p, cannot create an audio input stream!",
+               self.get());
+          return;
+        }
+        if (int r = self->mStream->Start(); r != CUBEB_OK) {
+          LOGE(
+              "AudioInputSource %p, cannot start its audio input stream! The "
+              "stream is destroyed directly!",
+              self.get());
+          self->mStream = nullptr;
+        }
+      })));
+}
+
+void AudioInputSource::Stop() {
+  // This is called on MediaTrackGraph's graph thread, which can be the cubeb
+  // stream's callback thread. Running cubeb operations within cubeb stream
+  // callback thread can cause the deadlock on Linux, so we dispatch those
+  // operations to the task thread.
+  MOZ_ASSERT(mTaskThread);
+
+  LOG("AudioInputSource %p, stop", this);
+  MOZ_ALWAYS_SUCCEEDS(mTaskThread->Dispatch(
+      NS_NewRunnableFunction(__func__, [self = RefPtr(this)]() mutable {
+        if (!self->mStream) {
+          LOGE("AudioInputSource %p, has no audio input stream to stop!",
+               self.get());
+          return;
+        }
+        if (int r = self->mStream->Stop(); r != CUBEB_OK) {
+          LOGE(
+              "AudioInputSource %p, cannot stop its audio input stream! The "
+              "stream is going to be destroyed forcefully",
+              self.get());
+        }
+        self->mStream = nullptr;
+      })));
+}
+
+AudioSegment AudioInputSource::GetAudioSegment(TrackTime aDuration) {
+  AudioSegment raw;
+  while (mSPSCQueue.AvailableRead()) {
+    AudioChunk chunk;
+    DebugOnly<int> reads = mSPSCQueue.Dequeue(&chunk, 1);
+    MOZ_ASSERT(reads);
+    raw.AppendAndConsumeChunk(std::move(chunk));
+  }
+
+  return mDriftCorrector.RequestFrames(raw, static_cast<uint32_t>(aDuration));
+}
+
+long AudioInputSource::DataCallback(const void* aBuffer, long aFrames) {
+  const AudioDataValue* source =
+      reinterpret_cast<const AudioDataValue*>(aBuffer);
+
+  AudioChunk c = AudioChunk::FromInterleavedBuffer(
+      source, static_cast<size_t>(aFrames), mChannelCount, mPrincipalHandle);
+
+  // Reset queue's producer to avoid hitting the assertion for checking the
+  // consistency of mSPSCQueue's mProducerId in Enqueue. This can happen when:
+  // 1) cubeb stream is reinitialized behind the scenes for the device changed
+  // events, e.g., users plug/unplug a TRRS mic into/from the built-in jack port
+  // of some old macbooks.
+  // 2) After Start() to Stop() cycle finishes, user call Start() again.
+  if (CheckThreadIdChanged()) {
+    mSPSCQueue.ResetProducerThreadId();
+    if (!mSandboxed) {
+      CubebUtils::GetAudioThreadRegistry()->Register(mAudioThreadId);
+    }
+  }
+
+  int writes = mSPSCQueue.Enqueue(c);
+  if (writes == 0) {
+    LOGW("AudioInputSource %p, buffer is full. Dropping %ld frames", this,
+         aFrames);
+  } else {
+    LOGV("AudioInputSource %p, enqueue %ld frames (%d AudioChunks)", this,
+         aFrames, writes);
+  }
+  return aFrames;
+}
+
+void AudioInputSource::StateCallback(cubeb_state aState) {
+  EventListener::State state;
+  if (aState == CUBEB_STATE_STARTED) {
+    LOG("AudioInputSource %p, stream started", this);
+    state = EventListener::State::Started;
+  } else if (aState == CUBEB_STATE_STOPPED) {
+    LOG("AudioInputSource %p, stream stopped", this);
+    state = EventListener::State::Stopped;
+  } else if (aState == CUBEB_STATE_DRAINED) {
+    LOG("AudioInputSource %p, stream is drained", this);
+    state = EventListener::State::Drained;
+  } else {
+    MOZ_ASSERT(aState == CUBEB_STATE_ERROR);
+    LOG("AudioInputSource %p, error happend", this);
+    state = EventListener::State::Error;
+  }
+  // This can be called on any thread, so we forward the event to main thread
+  // first.
+  NS_DispatchToMainThread(
+      NS_NewRunnableFunction(__func__, [self = RefPtr(this), s = state] {
+        self->mEventListener->AudioStateCallback(self->mId, s);
+      }));
+}
+
+void AudioInputSource::DeviceChangedCallback() {
+  LOG("AudioInputSource %p, device changed", this);
+  // This can be called on any thread, so we forward the event to main thread
+  // first.
+  NS_DispatchToMainThread(
+      NS_NewRunnableFunction(__func__, [self = RefPtr(this)] {
+        self->mEventListener->AudioDeviceChanged(self->mId);
+      }));
+}
+
+bool AudioInputSource::CheckThreadIdChanged() {
+  ProfilerThreadId id = profiler_current_thread_id();
+  if (id != mAudioThreadId) {
+    mAudioThreadId = id;
+    return true;
+  }
+  return false;
+}
+
+#undef LOG_INTERNAL
+#undef LOG
+#undef LOGW
+#undef LOGE
+#undef LOGV
+
+}  // namespace mozilla

dom/media/AudioInputSource.h

@@ -7,29 +7,67 @@
 #ifndef DOM_MEDIA_AudioInputSource_H_
 #define DOM_MEDIA_AudioInputSource_H_
 
+#include "AudioDriftCorrection.h"
 #include "AudioSegment.h"
+#include "CubebInputStream.h"
 #include "CubebUtils.h"
+#include "mozilla/ProfilerUtils.h"
+#include "mozilla/RefPtr.h"
+#include "mozilla/SPSCQueue.h"
+#include "mozilla/SharedThreadPool.h"
 
 namespace mozilla {
 
-// This is an interface to operate an input-only audio stream. Once users call
-// Start(), the audio data can be read by GetAudioSegment(). GetAudioSegment()
-// is not thread-safe, which means it can only be call in one specific thread.
-// The audio data is periodically produced by the underlying audio stream on its
-// callback thread.
-class AudioInputSource {
+// This is an interface to operate an input-only audio stream within a
+// cubeb-task thread on a specific thread. Once the class instance is created,
+// all its operations must be called on the same thread.
+//
+// The audio data is periodically produced by the underlying audio stream on the
+// stream's callback thread, and can be safely read by GetAudioSegment() on a
+// specific thread.
+class AudioInputSource : public CubebInputStream::Listener {
  public:
-  NS_INLINE_DECL_PURE_VIRTUAL_REFCOUNTING;
+  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(AudioInputSource, override);
 
   using Id = uint32_t;
 
-  // These functions should always be called in the same thread:
+  class EventListener {
+   public:
+    NS_INLINE_DECL_PURE_VIRTUAL_REFCOUNTING;
+
+    // This two events will be fired on main thread.
+    virtual void AudioDeviceChanged(Id aId) = 0;
+    enum class State { Started, Stopped, Drained, Error };
+    virtual void AudioStateCallback(Id aId, State aState) = 0;
+
+   protected:
+    EventListener() = default;
+    virtual ~EventListener() = default;
+  };
+
+  AudioInputSource(RefPtr<EventListener>&& aListener, Id aSourceId,
+                   CubebUtils::AudioDeviceID aDeviceId, uint32_t aChannelCount,
+                   bool aIsVoice, const PrincipalHandle& aPrincipalHandle,
+                   TrackRate aSourceRate, TrackRate aTargetRate,
+                   uint32_t aBufferMs);
+
+  // The following functions should always be called in the same thread: They
+  // are always run on MediaTrackGraph's graph thread.
   // Starts producing audio data.
-  virtual void Start() = 0;
+  virtual void Start();
   // Stops producing audio data.
-  virtual void Stop() = 0;
+  virtual void Stop();
   // Returns the AudioSegment with aDuration of data inside.
-  virtual AudioSegment GetAudioSegment(TrackTime aDuration) = 0;
+  virtual AudioSegment GetAudioSegment(TrackTime aDuration);
+
+  // CubebInputStream::Listener interface: These are used only for the
+  // underlying audio stream. No user should call these APIs.
+  // This will be fired on audio callback thread.
+  long DataCallback(const void* aBuffer, long aFrames) override;
+  // This can be fired on any thread.
+  void StateCallback(cubeb_state aState) override;
+  // This can be fired on any thread.
+  void DeviceChangedCallback() override;
 
   // Any threads:
   // The unique id of this source.
@@ -46,16 +84,38 @@ class AudioInputSource {
   const PrincipalHandle mPrincipalHandle;
 
  protected:
-  AudioInputSource(Id aId, CubebUtils::AudioDeviceID aDeviceId,
-                   uint32_t aChannelCount, TrackRate aRate, bool aIsVoice,
-                   const PrincipalHandle& aPrincipalHandle)
-      : mId(aId),
-        mDeviceId(aDeviceId),
-        mChannelCount(aChannelCount),
-        mRate(aRate),
-        mIsVoice(aIsVoice),
-        mPrincipalHandle(aPrincipalHandle) {}
-  virtual ~AudioInputSource() = default;
+  ~AudioInputSource() = default;
+
+ private:
+  // Underlying audio thread only.
+  bool CheckThreadIdChanged();
+
+  // Any thread.
+  const bool mSandboxed;
+
+  // Thread id of the underlying audio thread. Underlying audio thread only.
+  std::atomic<ProfilerThreadId> mAudioThreadId;
+
+  // Forward the underlying event from main thread.
+  const RefPtr<EventListener> mEventListener;
+
+  // Shared thread pool containing only one thread for cubeb operations.
+  // The cubeb operations: Start() and Stop() will be called on
+  // MediaTrackGraph's graph thread, which can be the cubeb stream's callback
+  // thread. Running cubeb operations within cubeb stream callback thread can
+  // cause the deadlock on Linux, so we dispatch those operations to the task
+  // thread.
+  const RefPtr<SharedThreadPool> mTaskThread;
+
+  // Correct the drift between the underlying audio stream and its reader.
+  AudioDriftCorrection mDriftCorrector;
+
+  // An input-only cubeb stream operated within mTaskThread.
+  UniquePtr<CubebInputStream> mStream;
+
+  // A single-producer-single-consumer lock-free queue whose data is produced by
+  // the audio callback thread and consumed by AudioInputSource's data reader.
+  SPSCQueue<AudioChunk> mSPSCQueue{30};
 };
 
 }  // namespace mozilla

dom/media/GraphDriver.cpp

@@ -502,7 +502,7 @@ AudioCallbackDriver::AudioCallbackDriver(
       mInputDeviceID(aInputDeviceID),
       mIterationDurationMS(MEDIA_GRAPH_TARGET_PERIOD_MS),
       mStarted(false),
-      mInitShutdownThread(SharedThreadPool::Get("CubebOperation"_ns, 1)),
+      mInitShutdownThread(CUBEB_TASK_THREAD),
       mAudioThreadId(ProfilerThreadId{}),
       mAudioThreadIdInCb(std::thread::id()),
       mAudioStreamState(AudioStreamState::None),

dom/media/GraphDriver.h

@@ -31,6 +31,11 @@ class nsAutoRefTraits<cubeb_stream> : public nsPointerRefTraits<cubeb_stream> {
 
 namespace mozilla {
 
+// A thread pool containing only one thread to execute the cubeb operations. We
+// should always use this thread to init, destroy, start, or stop cubeb streams,
+// to avoid data racing or deadlock issues across platforms.
+#define CUBEB_TASK_THREAD SharedThreadPool::Get("CubebOperation"_ns, 1)
+
 /**
  * Assume we can run an iteration of the MediaTrackGraph loop in this much time
  * or less.

dom/media/gtest/TestAudioInputSource.cpp

@@ -0,0 +1,265 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
+/* vim: set ts=2 et sw=2 tw=80: */
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+#include "AudioInputSource.h"
+
+#include "gmock/gmock.h"
+#include "gtest/gtest.h"
+
+#include "MockCubeb.h"
+#include "WaitFor.h"
+#include "nsContentUtils.h"
+
+using namespace mozilla;
+
+namespace {
+#define DispatchFunction(f) \
+  NS_DispatchToCurrentThread(NS_NewRunnableFunction(__func__, f))
+}  // namespace
+
+class MockEventListener : public AudioInputSource::EventListener {
+ public:
+  NS_INLINE_DECL_THREADSAFE_REFCOUNTING(MockEventListener, override);
+  MOCK_METHOD1(AudioDeviceChanged, void(AudioInputSource::Id));
+  MOCK_METHOD2(AudioStateCallback,
+               void(AudioInputSource::Id,
+                    AudioInputSource::EventListener::State));
+
+ private:
+  ~MockEventListener() = default;
+};
+
+TEST(TestAudioInputSource, StartAndStop)
+{
+  MockCubeb* cubeb = new MockCubeb();
+  CubebUtils::ForceSetCubebContext(cubeb->AsCubebContext());
+
+  const AudioInputSource::Id sourceId = 1;
+  const CubebUtils::AudioDeviceID deviceId = (CubebUtils::AudioDeviceID)1;
+  const uint32_t channels = 2;
+  const PrincipalHandle testPrincipal =
+      MakePrincipalHandle(nsContentUtils::GetSystemPrincipal());
+  const TrackRate sourceRate = 44100;
+  const TrackRate targetRate = 48000;
+  const uint32_t bufferingMs = 50;  // ms
+
+  auto listener = MakeRefPtr<MockEventListener>();
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Started))
+      .Times(2);
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Stopped))
+      .Times(2);
+
+  RefPtr<AudioInputSource> ais = MakeRefPtr<AudioInputSource>(
+      std::move(listener), sourceId, deviceId, channels, true, testPrincipal,
+      sourceRate, targetRate, bufferingMs);
+  ASSERT_TRUE(ais);
+
+  // Make sure start and stop works.
+  {
+    DispatchFunction([&] { ais->Start(); });
+    RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+    EXPECT_TRUE(stream->mHasInput);
+    EXPECT_FALSE(stream->mHasOutput);
+    EXPECT_EQ(stream->GetInputDeviceID(), deviceId);
+    EXPECT_EQ(stream->InputChannels(), channels);
+    EXPECT_EQ(stream->InputSampleRate(), static_cast<uint32_t>(sourceRate));
+
+    Unused << WaitFor(stream->FramesProcessedEvent());
+
+    DispatchFunction([&] { ais->Stop(); });
+    Unused << WaitFor(cubeb->StreamDestroyEvent());
+  }
+
+  // Make sure restart is ok.
+  {
+    DispatchFunction([&] { ais->Start(); });
+    RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+    EXPECT_TRUE(stream->mHasInput);
+    EXPECT_FALSE(stream->mHasOutput);
+    EXPECT_EQ(stream->GetInputDeviceID(), deviceId);
+    EXPECT_EQ(stream->InputChannels(), channels);
+    EXPECT_EQ(stream->InputSampleRate(), static_cast<uint32_t>(sourceRate));
+
+    Unused << WaitFor(stream->FramesProcessedEvent());
+
+    DispatchFunction([&] { ais->Stop(); });
+    Unused << WaitFor(cubeb->StreamDestroyEvent());
+  }
+
+  ais = nullptr;  // Drop the SharedThreadPool here.
+}
+
+TEST(TestAudioInputSource, DataOutputBeforeStartAndAfterStop)
+{
+  MockCubeb* cubeb = new MockCubeb();
+  CubebUtils::ForceSetCubebContext(cubeb->AsCubebContext());
+
+  const AudioInputSource::Id sourceId = 1;
+  const CubebUtils::AudioDeviceID deviceId = (CubebUtils::AudioDeviceID)1;
+  const uint32_t channels = 2;
+  const PrincipalHandle testPrincipal =
+      MakePrincipalHandle(nsContentUtils::GetSystemPrincipal());
+  const TrackRate sourceRate = 44100;
+  const TrackRate targetRate = 48000;
+  const uint32_t bufferingMs = 50;  // ms
+
+  const TrackTime requestFrames = 2 * WEBAUDIO_BLOCK_SIZE;
+
+  auto listener = MakeRefPtr<MockEventListener>();
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Started));
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Stopped));
+
+  RefPtr<AudioInputSource> ais = MakeRefPtr<AudioInputSource>(
+      std::move(listener), sourceId, deviceId, channels, true, testPrincipal,
+      sourceRate, targetRate, bufferingMs);
+  ASSERT_TRUE(ais);
+
+  // It's ok to call GetAudioSegment before starting
+  {
+    AudioSegment data = ais->GetAudioSegment(requestFrames);
+    EXPECT_EQ(data.GetDuration(), requestFrames);
+    EXPECT_TRUE(data.IsNull());
+  }
+
+  DispatchFunction([&] { ais->Start(); });
+  RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+  EXPECT_TRUE(stream->mHasInput);
+  EXPECT_FALSE(stream->mHasOutput);
+  EXPECT_EQ(stream->InputChannels(), channels);
+
+  stream->SetInputRecordingEnabled(true);
+
+  Unused << WaitFor(stream->FramesProcessedEvent());
+
+  DispatchFunction([&] { ais->Stop(); });
+  Unused << WaitFor(cubeb->StreamDestroyEvent());
+
+  // Check the data output
+  {
+    nsTArray<AudioDataValue> record = stream->TakeRecordedInput();
+    size_t frames = record.Length() / channels;
+    AudioSegment deinterleaved;
+    deinterleaved.AppendFromInterleavedBuffer(record.Elements(), frames,
+                                              channels, testPrincipal);
+    AudioDriftCorrection driftCorrector(sourceRate, targetRate, bufferingMs,
+                                        testPrincipal);
+    AudioSegment expectedSegment = driftCorrector.RequestFrames(
+        deinterleaved, static_cast<uint32_t>(requestFrames));
+
+    nsTArray<AudioDataValue> expected;
+    size_t expectedSamples =
+        expectedSegment.WriteToInterleavedBuffer(expected, channels);
+
+    AudioSegment actualSegment = ais->GetAudioSegment(requestFrames);
+    EXPECT_EQ(actualSegment.GetDuration(), requestFrames);
+    nsTArray<AudioDataValue> actual;
+    size_t actualSamples =
+        actualSegment.WriteToInterleavedBuffer(actual, channels);
+
+    EXPECT_EQ(actualSamples, expectedSamples);
+    EXPECT_EQ(actualSamples / channels, static_cast<size_t>(requestFrames));
+    EXPECT_EQ(actual, expected);
+  }
+
+  ais = nullptr;  // Drop the SharedThreadPool here.
+}
+
+TEST(TestAudioInputSource, ErrorCallback)
+{
+  MockCubeb* cubeb = new MockCubeb();
+  CubebUtils::ForceSetCubebContext(cubeb->AsCubebContext());
+
+  const AudioInputSource::Id sourceId = 1;
+  const CubebUtils::AudioDeviceID deviceId = (CubebUtils::AudioDeviceID)1;
+  const uint32_t channels = 2;
+  const PrincipalHandle testPrincipal =
+      MakePrincipalHandle(nsContentUtils::GetSystemPrincipal());
+  const TrackRate sourceRate = 44100;
+  const TrackRate targetRate = 48000;
+  const uint32_t bufferingMs = 50;  // ms
+
+  auto listener = MakeRefPtr<MockEventListener>();
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Started));
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Error));
+
+  RefPtr<AudioInputSource> ais = MakeRefPtr<AudioInputSource>(
+      std::move(listener), sourceId, deviceId, channels, true, testPrincipal,
+      sourceRate, targetRate, bufferingMs);
+  ASSERT_TRUE(ais);
+
+  DispatchFunction([&] { ais->Start(); });
+  RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+  EXPECT_TRUE(stream->mHasInput);
+  EXPECT_FALSE(stream->mHasOutput);
+  EXPECT_EQ(stream->InputChannels(), channels);
+
+  Unused << WaitFor(stream->FramesProcessedEvent());
+
+  DispatchFunction([&] { stream->ForceError(); });
+  WaitFor(stream->ErrorForcedEvent());
+
+  DispatchFunction([&] { ais->Stop(); });
+  Unused << WaitFor(cubeb->StreamDestroyEvent());
+
+  ais = nullptr;  // Drop the SharedThreadPool here.
+}
+
+TEST(TestAudioInputSource, DeviceChangedCallback)
+{
+  MockCubeb* cubeb = new MockCubeb();
+  CubebUtils::ForceSetCubebContext(cubeb->AsCubebContext());
+
+  const AudioInputSource::Id sourceId = 1;
+  const CubebUtils::AudioDeviceID deviceId = (CubebUtils::AudioDeviceID)1;
+  const uint32_t channels = 2;
+  const PrincipalHandle testPrincipal =
+      MakePrincipalHandle(nsContentUtils::GetSystemPrincipal());
+  const TrackRate sourceRate = 44100;
+  const TrackRate targetRate = 48000;
+  const uint32_t bufferingMs = 50;  // ms
+
+  auto listener = MakeRefPtr<MockEventListener>();
+  EXPECT_CALL(*listener, AudioDeviceChanged(sourceId));
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Started));
+  EXPECT_CALL(*listener,
+              AudioStateCallback(
+                  sourceId, AudioInputSource::EventListener::State::Stopped));
+
+  RefPtr<AudioInputSource> ais = MakeRefPtr<AudioInputSource>(
+      std::move(listener), sourceId, deviceId, channels, true, testPrincipal,
+      sourceRate, targetRate, bufferingMs);
+  ASSERT_TRUE(ais);
+
+  DispatchFunction([&] { ais->Start(); });
+  RefPtr<SmartMockCubebStream> stream = WaitFor(cubeb->StreamInitEvent());
+  EXPECT_TRUE(stream->mHasInput);
+  EXPECT_FALSE(stream->mHasOutput);
+  EXPECT_EQ(stream->InputChannels(), channels);
+
+  Unused << WaitFor(stream->FramesProcessedEvent());
+
+  DispatchFunction([&] { stream->ForceDeviceChanged(); });
+  WaitFor(stream->DeviceChangeForcedEvent());
+
+  DispatchFunction([&] { ais->Stop(); });
+  Unused << WaitFor(cubeb->StreamDestroyEvent());
+
+  ais = nullptr;  // Drop the SharedThreadPool here.
+}

dom/media/gtest/TestDeviceInputTrack.cpp

@@ -150,6 +150,7 @@ TEST_F(TestDeviceInputTrack, OpenTwiceWithoutCloseFirst) {
   NativeInputTrack::CloseAudio(std::move(track1), nullptr);
 }
 
+// TODO: Use AudioInputSource with MockCubeb for the test
 TEST_F(TestDeviceInputTrack, NonNativeInputTrackData) {
   // Graph settings
   const uint32_t flags = 0;
@@ -191,17 +192,29 @@ TEST_F(TestDeviceInputTrack, NonNativeInputTrackData) {
   next = MediaTrackGraphImpl::RoundUpToEndOfAudioBlock(2 * frames);
   ASSERT_NE(current, next);  // Make sure we have data produced in ProcessInput.
 
+  class MockEventListener : public AudioInputSource::EventListener {
+   public:
+    NS_INLINE_DECL_THREADSAFE_REFCOUNTING(MockEventListener, override);
+    MOCK_METHOD1(AudioDeviceChanged, void(AudioInputSource::Id));
+    MOCK_METHOD2(AudioStateCallback,
+                 void(AudioInputSource::Id,
+                      AudioInputSource::EventListener::State));
+
+   private:
+    ~MockEventListener() = default;
+  };
+
   class TestAudioSource final : public AudioInputSource {
    public:
-    NS_INLINE_DECL_THREADSAFE_REFCOUNTING(TestAudioSource, override);
-
-    TestAudioSource(uint32_t aSourceId, CubebUtils::AudioDeviceID aDeviceId,
-                    uint32_t aChannelCount, TrackRate aRate, bool aIsVoice,
-                    const PrincipalHandle& aPrincipalHandle,
-                    AudioSegment& aCurrentData)
-        : AudioInputSource(aSourceId, aDeviceId, aChannelCount, aRate, aIsVoice,
-                           aPrincipalHandle),
-          mGenerator(mChannelCount, mRate),
+    TestAudioSource(RefPtr<EventListener>&& aListener, Id aSourceId,
+                    CubebUtils::AudioDeviceID aDeviceId, uint32_t aChannelCount,
+                    bool aIsVoice, const PrincipalHandle& aPrincipalHandle,
+                    TrackRate aSourceRate, TrackRate aTargetRate,
+                    uint32_t aBufferMs, AudioSegment& aCurrentData)
+        : AudioInputSource(std::move(aListener), aSourceId, aDeviceId,
+                           aChannelCount, aIsVoice, aPrincipalHandle,
+                           aSourceRate, aTargetRate, aBufferMs),
+          mGenerator(mChannelCount, aTargetRate),
           mCurrentData(aCurrentData) {}
 
     void Start() override {}
@@ -224,9 +237,10 @@ TEST_F(TestDeviceInputTrack, NonNativeInputTrackData) {
 
   // We need to make sure the buffer lives longer than the TestAudioSource here
   // since TestAudioSource will access buffer by reference.
-  track->StartAudio(
-      MakeRefPtr<TestAudioSource>(1 /* Ignored */, deviceId, channelCount, rate,
-                                  true /* Ignored*/, testPrincipal, buffer));
+  track->StartAudio(MakeRefPtr<TestAudioSource>(
+      MakeRefPtr<MockEventListener>(), 1 /* Ignored */, deviceId, channelCount,
+      true /* Ignored*/, testPrincipal, rate, mGraph->GraphRate(),
+      50 /* Ignored */, buffer));
   track->ProcessInput(current, next, flags);
   {
     AudioSegment data;

dom/media/gtest/moz.build

@@ -25,6 +25,7 @@ UNIFIED_SOURCES += [
     "TestAudioCompactor.cpp",
     "TestAudioDecoderInputTrack.cpp",
     "TestAudioDriftCorrection.cpp",
+    "TestAudioInputSource.cpp",
     "TestAudioMixer.cpp",
     "TestAudioPacketizer.cpp",
     "TestAudioRingBuffer.cpp",

dom/media/moz.build

@@ -246,6 +246,7 @@ UNIFIED_SOURCES += [
     "AudioConfig.cpp",
     "AudioConverter.cpp",
     "AudioDeviceInfo.cpp",
+    "AudioInputSource.cpp",
     "AudioRingBuffer.cpp",
     "AudioSegment.cpp",
     "AudioStream.cpp",

mfbt/SPSCQueue.h

@@ -266,8 +266,19 @@ class SPSCRingBufferBase {
    * asserts are disabled.
    */
   void ResetThreadIds() {
+    ResetProducerThreadId();
+    ResetConsumerThreadId();
+  }
+
+  void ResetConsumerThreadId() {
 #ifdef DEBUG
-    mConsumerId = mProducerId = std::thread::id();
+    mConsumerId = std::thread::id();
+#endif
+  }
+
+  void ResetProducerThreadId() {
+#ifdef DEBUG
+    mProducerId = std::thread::id();
 #endif
   }