Support for remote audio into tracks

talk/app/webrtc/remoteaudiosource.cc

Index: talk/app/webrtc/remoteaudiosource.cc
diff --git a/talk/app/webrtc/remoteaudiosource.cc b/talk/app/webrtc/remoteaudiosource.cc
index 41f3d8798a912bb24e897f54c250ab3ed38494c6..e5af1e9487503f40c37a81a6315dd6f72a2657fe 100644
--- a/talk/app/webrtc/remoteaudiosource.cc
+++ b/talk/app/webrtc/remoteaudiosource.cc
@@ -29,44 +29,143 @@
 
 #include <algorithm>
 #include <functional>
+#include <utility>
 
+#include "talk/app/webrtc/mediastreamprovider.h"
+#include "webrtc/base/checks.h"
 #include "webrtc/base/logging.h"
+#include "webrtc/base/thread.h"
 
 namespace webrtc {
 
-rtc::scoped_refptr<RemoteAudioSource> RemoteAudioSource::Create() {
-  return new rtc::RefCountedObject<RemoteAudioSource>();
+class RemoteAudioSource::MessageHandler : public rtc::MessageHandler {
+ public:
+  explicit MessageHandler(RemoteAudioSource* source) : source_(source) {}
+
+ private:
+  ~MessageHandler() override {}
+
+  void OnMessage(rtc::Message* msg) override {
+    source_->OnMessage(msg);
+    delete this;
+  }
+
+  const rtc::scoped_refptr<RemoteAudioSource> source_;
+  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(MessageHandler);
+};
+
+class RemoteAudioSource::Sink : public AudioSinkInterface {
+ public:
+  explicit Sink(RemoteAudioSource* source) : source_(source) {}
+  ~Sink() override { source_->OnAudioProviderGone(); }
+
+ private:
+  void OnData(const AudioSinkInterface::Data& audio) override {
+    if (source_)
+      source_->OnData(audio);
+  }
+
+  const rtc::scoped_refptr<RemoteAudioSource> source_;
+  RTC_DISALLOW_IMPLICIT_CONSTRUCTORS(Sink);
+};
+
+rtc::scoped_refptr<RemoteAudioSource> RemoteAudioSource::Create(
+    uint32_t ssrc,
+    AudioProviderInterface* provider) {
+  rtc::scoped_refptr<RemoteAudioSource> ret(
+      new rtc::RefCountedObject<RemoteAudioSource>());
+  ret->Initialize(ssrc, provider);
+  return ret;
 }
 
-RemoteAudioSource::RemoteAudioSource() {
+RemoteAudioSource::RemoteAudioSource()
+    : main_thread_(rtc::Thread::Current()),
+      state_(MediaSourceInterface::kLive) {
+  RTC_DCHECK(main_thread_);
 }
 
 RemoteAudioSource::~RemoteAudioSource() {
-  ASSERT(audio_observers_.empty());
+  RTC_DCHECK(main_thread_->IsCurrent());
+  RTC_DCHECK(audio_observers_.empty());
+  RTC_DCHECK(sinks_.empty());
+}
+
+void RemoteAudioSource::Initialize(uint32_t ssrc,
+                                   AudioProviderInterface* provider) {
+  RTC_DCHECK(main_thread_->IsCurrent());
+  // To make sure we always get notified when the provider goes out of scope,
+  // we register for callbacks here and not on demand in AddSink.
+  if (provider) {  // May be null in tests.
+    provider->SetRawAudioSink(
+        ssrc, std::move(rtc::scoped_ptr<AudioSinkInterface>(new Sink(this))));
+  }
 }
 
 MediaSourceInterface::SourceState RemoteAudioSource::state() const {
-  return MediaSourceInterface::kLive;
+  RTC_DCHECK(main_thread_->IsCurrent());
+  return state_;
 }
 
 void RemoteAudioSource::SetVolume(double volume) {
-  ASSERT(volume >= 0 && volume <= 10);
-  for (AudioObserverList::iterator it = audio_observers_.begin();
-       it != audio_observers_.end(); ++it) {
-    (*it)->OnSetVolume(volume);
-  }
+  RTC_DCHECK(volume >= 0 && volume <= 10);
+  for (auto* observer : audio_observers_)
+    observer->OnSetVolume(volume);
 }
 
 void RemoteAudioSource::RegisterAudioObserver(AudioObserver* observer) {
-  ASSERT(observer != NULL);
-  ASSERT(std::find(audio_observers_.begin(), audio_observers_.end(),
-                   observer) == audio_observers_.end());
+  RTC_DCHECK(observer != NULL);
+  RTC_DCHECK(std::find(audio_observers_.begin(), audio_observers_.end(),
+                       observer) == audio_observers_.end());
   audio_observers_.push_back(observer);
 }
 
 void RemoteAudioSource::UnregisterAudioObserver(AudioObserver* observer) {
-  ASSERT(observer != NULL);
+  RTC_DCHECK(observer != NULL);
   audio_observers_.remove(observer);
 }
 
+void RemoteAudioSource::AddSink(AudioTrackSinkInterface* sink) {
+  RTC_DCHECK(main_thread_->IsCurrent());
+  RTC_DCHECK(sink);
+
+  if (state_ != MediaSourceInterface::kLive) {
+    LOG(LS_ERROR) << "Can't register sink as the source isn't live.";
+    return;
+  }
+
+  rtc::CritScope lock(&sink_lock_);
+  RTC_DCHECK(std::find(sinks_.begin(), sinks_.end(), sink) == sinks_.end());
+  sinks_.push_back(sink);
+}
+
+void RemoteAudioSource::RemoveSink(AudioTrackSinkInterface* sink) {
+  RTC_DCHECK(main_thread_->IsCurrent());
+  RTC_DCHECK(sink);
+
+  rtc::CritScope lock(&sink_lock_);
+  sinks_.remove(sink);
+}
+
+void RemoteAudioSource::OnData(const AudioSinkInterface::Data& audio) {
+  // Called on the externally-owned audio callback thread, via/from webrtc.
+  rtc::CritScope lock(&sink_lock_);
+  for (auto* sink : sinks_) {
+    sink->OnData(audio.data, 16, audio.sample_rate, audio.channels,
+                 audio.samples_per_channel);
+  }
+}
+
+void RemoteAudioSource::OnAudioProviderGone() {
+  // Called when the data provider is deleted.  It may be the worker thread
+  // in libjingle or may be a different worker thread.
+  main_thread_->Post(new MessageHandler(this));
+}
+
+void RemoteAudioSource::OnMessage(rtc::Message* msg) {
+  RTC_DCHECK(main_thread_->IsCurrent());
+  sinks_.clear();
+  state_ = MediaSourceInterface::kEnded;
+  FireOnChanged();
+}
+
 }  // namespace webrtc