Bug 664918. Part 5: Create SourceMediaStream, a MediaStream with an API allowing data to be injected into it by some source. r=jesup

content/media/MediaStreamGraph.cpp

@@ -205,6 +205,14 @@ public:
    * will take effect.
    */
   void ChooseActionTime();
+  /**
+   * Extract any state updates pending in aStream, and apply them.
+   */
+  void ExtractPendingInput(SourceMediaStream* aStream);
+  /**
+   * Update "have enough data" flags in aStream.
+   */
+  void UpdateBufferSufficiencyState(SourceMediaStream* aStream);
   /**
    * Compute the blocking states of streams from mBlockingDecisionsMadeUntilTime
    * until the desired future time (determined by heuristic).
@@ -604,6 +612,72 @@ MediaStreamGraphImpl::ChooseActionTime()
   mLastActionTime = GetEarliestActionTime();
 }
 
+void
+MediaStreamGraphImpl::ExtractPendingInput(SourceMediaStream* aStream)
+{
+  bool finished;
+  {
+    MutexAutoLock lock(aStream->mMutex);
+    finished = aStream->mUpdateFinished;
+    for (PRInt32 i = aStream->mUpdateTracks.Length() - 1; i >= 0; --i) {
+      SourceMediaStream::TrackData* data = &aStream->mUpdateTracks[i];
+      if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
+        MediaSegment* segment = data->mData.forget();
+        LOG(PR_LOG_DEBUG, ("SourceMediaStream %p creating track %d, rate %d, start %lld, initial end %lld",
+                           aStream, data->mID, data->mRate, PRInt64(data->mStart),
+                           PRInt64(segment->GetDuration())));
+        aStream->mBuffer.AddTrack(data->mID, data->mRate, data->mStart, segment);
+        // The track has taken ownership of data->mData, so let's replace
+        // data->mData with an empty clone.
+        data->mData = segment->CreateEmptyClone();
+        data->mCommands &= ~SourceMediaStream::TRACK_CREATE;
+      } else if (data->mData->GetDuration() > 0) {
+        MediaSegment* dest = aStream->mBuffer.FindTrack(data->mID)->GetSegment();
+        LOG(PR_LOG_DEBUG, ("SourceMediaStream %p track %d, advancing end from %lld to %lld",
+                           aStream, data->mID,
+                           PRInt64(dest->GetDuration()),
+                           PRInt64(dest->GetDuration() + data->mData->GetDuration())));
+        dest->AppendFrom(data->mData);
+      }
+      if (data->mCommands & SourceMediaStream::TRACK_END) {
+        aStream->mBuffer.FindTrack(data->mID)->SetEnded();
+        aStream->mUpdateTracks.RemoveElementAt(i);
+      }
+    }
+    aStream->mBuffer.AdvanceKnownTracksTime(aStream->mUpdateKnownTracksTime);
+  }
+  if (finished) {
+    FinishStream(aStream);
+  }
+}
+
+void
+MediaStreamGraphImpl::UpdateBufferSufficiencyState(SourceMediaStream* aStream)
+{
+  StreamTime desiredEnd = GetDesiredBufferEnd(aStream);
+  nsTArray<SourceMediaStream::ThreadAndRunnable> runnables;
+
+  {
+    MutexAutoLock lock(aStream->mMutex);
+    for (PRUint32 i = 0; i < aStream->mUpdateTracks.Length(); ++i) {
+      SourceMediaStream::TrackData* data = &aStream->mUpdateTracks[i];
+      if (data->mCommands & SourceMediaStream::TRACK_CREATE) {
+        continue;
+      }
+      StreamBuffer::Track* track = aStream->mBuffer.FindTrack(data->mID);
+      data->mHaveEnough = track->GetEndTimeRoundDown() >= desiredEnd;
+      if (!data->mHaveEnough) {
+        runnables.MoveElementsFrom(data->mDispatchWhenNotEnough);
+      }
+    }
+  }
+
+  for (PRUint32 i = 0; i < runnables.Length(); ++i) {
+    runnables[i].mThread->Dispatch(runnables[i].mRunnable, 0);
+  }
+}
+
+
 StreamTime
 MediaStreamGraphImpl::GraphTimeToStreamTime(MediaStream* aStream,
                                             GraphTime aTime)
@@ -1186,6 +1260,14 @@ MediaStreamGraphImpl::RunThread()
     }
     messageQueue.Clear();
 
+    // Grab pending ProcessingEngine results.
+    for (PRUint32 i = 0; i < mStreams.Length(); ++i) {
+      SourceMediaStream* is = mStreams[i]->AsSourceStream();
+      if (is) {
+        ExtractPendingInput(is);
+      }
+    }
+
     GraphTime prevBlockingDecisionsMadeUntilTime = mBlockingDecisionsMadeUntilTime;
     RecomputeBlocking();
 
@@ -1202,6 +1284,10 @@ MediaStreamGraphImpl::RunThread()
         ++audioStreamsActive;
       }
       PlayVideo(stream);
+      SourceMediaStream* is = stream->AsSourceStream();
+      if (is) {
+        UpdateBufferSufficiencyState(is);
+      }
       GraphTime end;
       if (!stream->mBlocked.GetAt(mCurrentTime, &end) || end < GRAPH_TIME_MAX) {
         allBlockedForever = false;
@@ -1705,6 +1791,83 @@ MediaStream::RemoveListener(MediaStreamListener* aListener)
   GraphImpl()->AppendMessage(new Message(this, aListener));
 }
 
+void
+SourceMediaStream::AddTrack(TrackID aID, TrackRate aRate, TrackTicks aStart,
+                            MediaSegment* aSegment)
+{
+  {
+    MutexAutoLock lock(mMutex);
+    TrackData* data = mUpdateTracks.AppendElement();
+    data->mID = aID;
+    data->mRate = aRate;
+    data->mStart = aStart;
+    data->mCommands = TRACK_CREATE;
+    data->mData = aSegment;
+    data->mHaveEnough = false;
+  }
+  GraphImpl()->EnsureNextIteration();
+}
+
+void
+SourceMediaStream::AppendToTrack(TrackID aID, MediaSegment* aSegment)
+{
+  {
+    MutexAutoLock lock(mMutex);
+    FindDataForTrack(aID)->mData->AppendFrom(aSegment);
+  }
+  GraphImpl()->EnsureNextIteration();
+}
+
+bool
+SourceMediaStream::HaveEnoughBuffered(TrackID aID)
+{
+  MutexAutoLock lock(mMutex);
+  return FindDataForTrack(aID)->mHaveEnough;
+}
+
+void
+SourceMediaStream::DispatchWhenNotEnoughBuffered(TrackID aID,
+    nsIThread* aSignalThread, nsIRunnable* aSignalRunnable)
+{
+  MutexAutoLock lock(mMutex);
+  TrackData* data = FindDataForTrack(aID);
+  if (data->mHaveEnough) {
+    data->mDispatchWhenNotEnough.AppendElement()->Init(aSignalThread, aSignalRunnable);
+  } else {
+    aSignalThread->Dispatch(aSignalRunnable, 0);
+  }
+}
+
+void
+SourceMediaStream::EndTrack(TrackID aID)
+{
+  {
+    MutexAutoLock lock(mMutex);
+    FindDataForTrack(aID)->mCommands |= TRACK_END;
+  }
+  GraphImpl()->EnsureNextIteration();
+}
+
+void
+SourceMediaStream::AdvanceKnownTracksTime(StreamTime aKnownTime)
+{
+  {
+    MutexAutoLock lock(mMutex);
+    mUpdateKnownTracksTime = aKnownTime;
+  }
+  GraphImpl()->EnsureNextIteration();
+}
+
+void
+SourceMediaStream::Finish()
+{
+  {
+    MutexAutoLock lock(mMutex);
+    mUpdateFinished = true;
+  }
+  GraphImpl()->EnsureNextIteration();
+}
+
 static const PRUint32 kThreadLimit = 4;
 static const PRUint32 kIdleThreadLimit = 4;
 static const PRUint32 kIdleThreadTimeoutMs = 2000;
@@ -1769,4 +1932,13 @@ MediaStreamGraph::GetInstance()
   return gGraph;
 }
 
+SourceMediaStream*
+MediaStreamGraph::CreateInputStream(nsDOMMediaStream* aWrapper)
+{
+  SourceMediaStream* stream = new SourceMediaStream(aWrapper);
+  NS_ADDREF(stream);
+  static_cast<MediaStreamGraphImpl*>(this)->AppendMessage(new CreateMessage(stream));
+  return stream;
+}
+
 }

content/media/MediaStreamGraph.h

@@ -111,6 +111,7 @@ public:
 };
 
 class MediaStreamGraphImpl;
+class SourceMediaStream;
 
 /**
  * A stream of synchronized audio and video data. All (not blocked) streams
@@ -216,6 +217,8 @@ public:
 
   friend class MediaStreamGraphImpl;
 
+  virtual SourceMediaStream* AsSourceStream() { return nsnull; }
+
   // media graph thread only
   void Init();
   // These Impl methods perform the core functionality of the control methods
@@ -342,6 +345,119 @@ protected:
   bool mMainThreadFinished;
 };
 
+/**
+ * This is a stream into which a decoder can write audio and video.
+ *
+ * Audio and video can be written on any thread, but you probably want to
+ * always write from the same thread to avoid unexpected interleavings.
+ */
+class SourceMediaStream : public MediaStream {
+public:
+  SourceMediaStream(nsDOMMediaStream* aWrapper) :
+    MediaStream(aWrapper), mMutex("mozilla::media::SourceMediaStream"),
+    mUpdateKnownTracksTime(0), mUpdateFinished(false)
+  {}
+
+  virtual SourceMediaStream* AsSourceStream() { return this; }
+
+  // Call these on any thread.
+  /**
+   * Add a new track to the stream starting at the given base time (which
+   * must be greater than or equal to the last time passed to
+   * AdvanceKnownTracksTime). Takes ownership of aSegment. aSegment should
+   * contain data starting after aStart.
+   */
+  void AddTrack(TrackID aID, TrackRate aRate, TrackTicks aStart,
+                MediaSegment* aSegment);
+  /**
+   * Append media data to a track. Ownership of aSegment remains with the caller,
+   * but aSegment is emptied.
+   */
+  void AppendToTrack(TrackID aID, MediaSegment* aSegment);
+  /**
+   * Returns true if the buffer currently has enough data.
+   */
+  bool HaveEnoughBuffered(TrackID aID);
+  /**
+   * Ensures that aSignalRunnable will be dispatched to aSignalThread
+   * when we don't have enough buffered data in the track (which could be
+   * immediately).
+   */
+  void DispatchWhenNotEnoughBuffered(TrackID aID,
+      nsIThread* aSignalThread, nsIRunnable* aSignalRunnable);
+  /**
+   * Indicate that a track has ended. Do not do any more API calls
+   * affecting this track.
+   */
+  void EndTrack(TrackID aID);
+  /**
+   * Indicate that no tracks will be added starting before time aKnownTime.
+   * aKnownTime must be >= its value at the last call to AdvanceKnownTracksTime.
+   */
+  void AdvanceKnownTracksTime(StreamTime aKnownTime);
+  /**
+   * Indicate that this stream should enter the "finished" state. All tracks
+   * must have been ended via EndTrack. The finish time of the stream is
+   * when all tracks have ended and when latest time sent to
+   * AdvanceKnownTracksTime() has been reached.
+   */
+  void Finish();
+
+  // XXX need a Reset API
+
+  friend class MediaStreamGraph;
+  friend class MediaStreamGraphImpl;
+
+  struct ThreadAndRunnable {
+    void Init(nsIThread* aThread, nsIRunnable* aRunnable)
+    {
+      mThread = aThread;
+      mRunnable = aRunnable;
+    }
+
+    nsCOMPtr<nsIThread> mThread;
+    nsCOMPtr<nsIRunnable> mRunnable;
+  };
+  enum TrackCommands {
+    TRACK_CREATE = 0x01,
+    TRACK_END = 0x02
+  };
+  /**
+   * Data for each track that hasn't ended.
+   */
+  struct TrackData {
+    TrackID mID;
+    TrackRate mRate;
+    TrackTicks mStart;
+    // Each time the track updates are flushed to the media graph thread,
+    // this is cleared.
+    PRUint32 mCommands;
+    // Each time the track updates are flushed to the media graph thread,
+    // the segment buffer is emptied.
+    nsAutoPtr<MediaSegment> mData;
+    nsTArray<ThreadAndRunnable> mDispatchWhenNotEnough;
+    bool mHaveEnough;
+  };
+
+protected:
+  TrackData* FindDataForTrack(TrackID aID)
+  {
+    for (PRUint32 i = 0; i < mUpdateTracks.Length(); ++i) {
+      if (mUpdateTracks[i].mID == aID) {
+        return &mUpdateTracks[i];
+      }
+    }
+    NS_ERROR("Bad track ID!");
+    return nsnull;
+  }
+
+  Mutex mMutex;
+  // protected by mMutex
+  StreamTime mUpdateKnownTracksTime;
+  nsTArray<TrackData> mUpdateTracks;
+  bool mUpdateFinished;
+};
+
 /**
  * Initially, at least, we will have a singleton MediaStreamGraph per
  * process.
@@ -351,6 +467,11 @@ public:
   // Main thread only
   static MediaStreamGraph* GetInstance();
   // Control API.
+  /**
+   * Create a stream that a media decoder (or some other source of
+   * media data, such as a camera) can write to.
+   */
+  SourceMediaStream* CreateInputStream(nsDOMMediaStream* aWrapper);
   /**
    * Returns the number of graph updates sent. This can be used to track
    * whether a given update has been processed by the graph thread and reflected

content/media/nsDOMMediaStream.cpp

@@ -42,6 +42,15 @@ nsDOMMediaStream::GetCurrentTime(double *aCurrentTime)
   return NS_OK;
 }
 
+already_AddRefed<nsDOMMediaStream>
+nsDOMMediaStream::CreateInputStream()
+{
+  nsRefPtr<nsDOMMediaStream> stream = new nsDOMMediaStream();
+  MediaStreamGraph* gm = MediaStreamGraph::GetInstance();
+  stream->mStream = gm->CreateInputStream(stream);
+  return stream.forget();
+}
+
 bool
 nsDOMMediaStream::CombineWithPrincipal(nsIPrincipal* aPrincipal)
 {

content/media/nsDOMMediaStream.h

@@ -43,6 +43,11 @@ public:
    */
   bool CombineWithPrincipal(nsIPrincipal* aPrincipal);
 
+  /**
+   * Create an nsDOMMediaStream whose underlying stream is a SourceMediaStream.
+   */
+  static already_AddRefed<nsDOMMediaStream> CreateInputStream();
+
 protected:
   // MediaStream is owned by the graph, but we tell it when to die, and it won't
   // die until we let it.