Bug 1195632. Part 2 - Have DecodedStream listen to push events of the media queues and call SendData() on its own without the help of MDSM. r=roc.

dom/media/DecodedStream.cpp

@@ -380,6 +380,7 @@ DecodedStream::StartPlayback(int64_t aStartTime, const MediaInfo& aInfo)
 
   mStartTime.emplace(aStartTime);
   mInfo = aInfo;
+  ConnectListener();
 
   class R : public nsRunnable {
     typedef MozPromiseHolder<GenericPromise> Promise;
@@ -417,7 +418,9 @@ void DecodedStream::StopPlayback()
   if (mStartTime.isNothing()) {
     return;
   }
+
   mStartTime.reset();
+  DisconnectListener();
 
   // Clear mData immediately when this playback session ends so we won't
   // send data to the wrong stream in SendData() in next playback session.
@@ -455,6 +458,19 @@ DecodedStream::CreateData(MozPromiseHolder<GenericPromise>&& aPromise)
   auto source = mOutputStreamManager.Graph()->CreateSourceStream(nullptr);
   mData.reset(new DecodedStreamData(source, mPlaying, Move(aPromise)));
   mOutputStreamManager.Connect(mData->mStream);
+
+  // Start to send data to the stream immediately
+  nsRefPtr<DecodedStream> self = this;
+  nsCOMPtr<nsIRunnable> r = NS_NewRunnableFunction([=] () {
+    ReentrantMonitorAutoEnter mon(self->GetReentrantMonitor());
+    // Don't send data if playback has ended.
+    if (self->mStartTime.isSome()) {
+      self->SendData();
+    }
+  });
+  // Don't assert success because the owner thread might have begun shutdown
+  // while we are still dealing with jobs on the main thread.
+  mOwnerThread->Dispatch(r.forget(), AbstractThread::DontAssertDispatchSuccess);
 }
 
 bool
@@ -824,4 +840,32 @@ DecodedStream::IsFinished() const
   return mData && mData->IsFinished();
 }
 
+void
+DecodedStream::ConnectListener()
+{
+  AssertOwnerThread();
+  ReentrantMonitorAutoEnter mon(GetReentrantMonitor());
+
+  mAudioPushListener = mAudioQueue.PushEvent().Connect(
+    mOwnerThread, this, &DecodedStream::SendData);
+  mAudioFinishListener = mAudioQueue.FinishEvent().Connect(
+    mOwnerThread, this, &DecodedStream::SendData);
+  mVideoPushListener = mVideoQueue.PushEvent().Connect(
+    mOwnerThread, this, &DecodedStream::SendData);
+  mVideoFinishListener = mVideoQueue.FinishEvent().Connect(
+    mOwnerThread, this, &DecodedStream::SendData);
+}
+
+void
+DecodedStream::DisconnectListener()
+{
+  AssertOwnerThread();
+  ReentrantMonitorAutoEnter mon(GetReentrantMonitor());
+
+  mAudioPushListener.Disconnect();
+  mVideoPushListener.Disconnect();
+  mAudioFinishListener.Disconnect();
+  mVideoFinishListener.Disconnect();
+}
+
 } // namespace mozilla

dom/media/DecodedStream.h

@@ -8,6 +8,7 @@
 #define DecodedStream_h_
 
 #include "nsTArray.h"
+#include "MediaEventSource.h"
 #include "MediaInfo.h"
 
 #include "mozilla/AbstractThread.h"
@@ -127,8 +128,6 @@ public:
   bool IsFinished() const;
   bool HasConsumers() const;
 
-  void SendData();
-
 protected:
   virtual ~DecodedStream();
 
@@ -139,11 +138,15 @@ private:
   void AdvanceTracks();
   void SendAudio(double aVolume, bool aIsSameOrigin);
   void SendVideo(bool aIsSameOrigin);
+  void SendData();
 
   void AssertOwnerThread() const {
     MOZ_ASSERT(mOwnerThread->IsCurrentThreadIn());
   }
 
+  void ConnectListener();
+  void DisconnectListener();
+
   const nsRefPtr<AbstractThread> mOwnerThread;
 
   UniquePtr<DecodedStreamData> mData;
@@ -168,6 +171,11 @@ private:
 
   MediaQueue<MediaData>& mAudioQueue;
   MediaQueue<MediaData>& mVideoQueue;
+
+  MediaEventListener mAudioPushListener;
+  MediaEventListener mVideoPushListener;
+  MediaEventListener mAudioFinishListener;
+  MediaEventListener mVideoFinishListener;
 };
 
 } // namespace mozilla

dom/media/MediaDecoderStateMachine.cpp

@@ -369,14 +369,12 @@ int64_t MediaDecoderStateMachine::GetDecodedAudioDuration()
   return audioDecoded;
 }
 
-void MediaDecoderStateMachine::SendStreamData()
+void MediaDecoderStateMachine::DiscardStreamData()
 {
   MOZ_ASSERT(OnTaskQueue());
   AssertCurrentThreadInMonitor();
   MOZ_ASSERT(!mAudioSink, "Should've been stopped in RunStateMachine()");
 
-  mDecodedStream->SendData();
-
   const auto clockTime = GetClock();
   while (true) {
     const MediaData* a = AudioQueue().PeekFront();
@@ -568,10 +566,6 @@ MediaDecoderStateMachine::OnAudioDecoded(AudioData* aAudioSample)
       if (mIsAudioPrerolling && DonePrerollingAudio()) {
         StopPrerollingAudio();
       }
-      // Schedule the state machine to send stream data as soon as possible.
-      if (mAudioCaptured) {
-        ScheduleStateMachine();
-      }
       return;
     }
 
@@ -767,10 +761,6 @@ MediaDecoderStateMachine::OnNotDecoded(MediaData::Type aType,
         return;
       }
       CheckIfDecodeComplete();
-      // Schedule the state machine to notify track ended as soon as possible.
-      if (mAudioCaptured) {
-        ScheduleStateMachine();
-      }
       return;
     }
     case DECODER_STATE_SEEKING: {
@@ -857,7 +847,7 @@ MediaDecoderStateMachine::OnVideoDecoded(VideoData* aVideoSample)
       // frame pushed in the queue, schedule the state machine as soon as
       // possible to render the video frame or delay the state machine thread
       // accurately.
-      if (mAudioCaptured || VideoQueue().GetSize() == 1) {
+      if (VideoQueue().GetSize() == 1) {
         ScheduleStateMachine();
       }
 
@@ -2651,7 +2641,7 @@ void MediaDecoderStateMachine::UpdateRenderedVideoFrames()
   }
 
   if (mAudioCaptured) {
-    SendStreamData();
+    DiscardStreamData();
   }
 
   TimeStamp nowTime;

dom/media/MediaDecoderStateMachine.h

@@ -323,10 +323,8 @@ public:
     mDecoder = nullptr;
   }
 
-  // Copy queued audio/video data in the reader to any output MediaStreams that
-  // need it.
-  void SendStreamData();
-  void FinishStreamData();
+  // Discard audio/video data that are already played by MSG.
+  void DiscardStreamData();
   bool HaveEnoughDecodedAudio(int64_t aAmpleAudioUSecs);
   bool HaveEnoughDecodedVideo();