Start/Stop receiving stream method for VideoTrack (#25)

* Initial implementation of start/stop receive * move set_state call to signalling thread as well. https://source.chromium.org/chromium/_/webrtc/src.git/+/dfd69c22100e1d2e83295f33d06fa9916caa5c53 * separate should_receive from enabled for now * ios hooks * Fix ios compile errors * code cleanup * clean up * reuse frame buffers instead
webrtc-sdk · Jul 17, 2022 · e88b47e · e88b47e
1 parent b585775
commit e88b47e
Show file tree

Hide file tree

Showing 17 changed files with 167 additions and 3 deletions.
diff --git a/api/media_stream_interface.cc b/api/media_stream_interface.cc
@@ -18,6 +18,10 @@ const char* const MediaStreamTrackInterface::kVideoKind =
 const char* const MediaStreamTrackInterface::kAudioKind =
     cricket::kMediaTypeAudio;
 
+bool VideoTrackInterface::should_receive() const {
+  return true;
+}
+
 VideoTrackInterface::ContentHint VideoTrackInterface::content_hint() const {
   return ContentHint::kNone;
 }

diff --git a/api/media_stream_interface.h b/api/media_stream_interface.h
@@ -188,6 +188,8 @@ class RTC_EXPORT VideoTrackInterface
 
   virtual VideoTrackSourceInterface* GetSource() const = 0;
 
+  virtual void set_should_receive(bool should_receive) {}
+  virtual bool should_receive() const;
   virtual ContentHint content_hint() const;
   virtual void set_content_hint(ContentHint hint) {}
 

diff --git a/media/base/media_channel.h b/media/base/media_channel.h
@@ -913,6 +913,9 @@ class VideoMediaChannel : public MediaChannel, public Delayable {
   virtual void GenerateKeyFrame(uint32_t ssrc) = 0;
 
   virtual std::vector<webrtc::RtpSource> GetSources(uint32_t ssrc) const = 0;
+
+  virtual void StartReceive(uint32_t ssrc) {}
+  virtual void StopReceive(uint32_t ssrc) {}
 };
 
 // Info about data received in DataMediaChannel.  For use in

diff --git a/media/engine/webrtc_video_engine.cc b/media/engine/webrtc_video_engine.cc
@@ -970,6 +970,24 @@ void WebRtcVideoChannel::RequestEncoderSwitch(
   }
 }
 
+void WebRtcVideoChannel::StartReceive(uint32_t ssrc) {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
+  WebRtcVideoReceiveStream* stream = FindReceiveStream(ssrc);
+  if(!stream) {
+    return;
+  }
+  stream->StartStream();
+}
+
+void WebRtcVideoChannel::StopReceive(uint32_t ssrc) {
+  RTC_DCHECK_RUN_ON(&thread_checker_);
+  WebRtcVideoReceiveStream* stream = FindReceiveStream(ssrc);
+  if(!stream) {
+    return;
+  }
+  stream->StopStream();
+}
+
 bool WebRtcVideoChannel::ApplyChangedParams(
     const ChangedSendParameters& changed_params) {
   RTC_DCHECK_RUN_ON(&thread_checker_);
@@ -3056,6 +3074,17 @@ void WebRtcVideoChannel::WebRtcVideoReceiveStream::SetRecvParameters(
   }
 }
 
+void WebRtcVideoChannel::WebRtcVideoReceiveStream::StartStream(){
+  if (stream_) {
+    stream_->Start();
+  }
+}
+void WebRtcVideoChannel::WebRtcVideoReceiveStream::StopStream(){
+  if (stream_) {
+    stream_->Stop();
+  }
+}
+
 void WebRtcVideoChannel::WebRtcVideoReceiveStream::RecreateReceiveStream() {
   absl::optional<int> base_minimum_playout_delay_ms;
   absl::optional<webrtc::VideoReceiveStreamInterface::RecordingState>

diff --git a/media/engine/webrtc_video_engine.h b/media/engine/webrtc_video_engine.h
@@ -254,7 +254,8 @@ class WebRtcVideoChannel : public VideoMediaChannel,
       uint32_t ssrc,
       rtc::scoped_refptr<webrtc::FrameTransformerInterface> frame_transformer)
       override;
-
+  void StartReceive(uint32_t ssrc) override;
+  void StopReceive(uint32_t ssrc) override;
  private:
   class WebRtcVideoReceiveStream;
 
@@ -501,6 +502,9 @@ class WebRtcVideoChannel : public VideoMediaChannel,
     void SetDepacketizerToDecoderFrameTransformer(
         rtc::scoped_refptr<webrtc::FrameTransformerInterface>
             frame_transformer);
+
+    void StartStream();
+    void StopStream();
 
     void SetLocalSsrc(uint32_t local_ssrc);
 

diff --git a/modules/video_coding/frame_buffer2.cc b/modules/video_coding/frame_buffer2.cc
@@ -332,6 +332,13 @@ void FrameBuffer::SetProtectionMode(VCMVideoProtection mode) {
   protection_mode_ = mode;
 }
 
+void FrameBuffer::Start() {
+  TRACE_EVENT0("webrtc", "FrameBuffer::Stop");
+  MutexLock lock(&mutex_);
+  if (!stopped_)
+    return;
+  stopped_ = false;
+}
 void FrameBuffer::Stop() {
   TRACE_EVENT0("webrtc", "FrameBuffer::Stop");
   MutexLock lock(&mutex_);

diff --git a/modules/video_coding/frame_buffer2.h b/modules/video_coding/frame_buffer2.h
@@ -78,6 +78,8 @@ class FrameBuffer {
   // Stop the frame buffer, causing any sleeping thread in NextFrame to
   // return immediately.
   void Stop();
+  // Unstop the frame buffer, re-allowing new frames to be inserted and read.
+  void Start();
 
   // Updates the RTT for jitter buffer estimation.
   void UpdateRtt(int64_t rtt_ms);

diff --git a/pc/media_stream_track_proxy.h b/pc/media_stream_track_proxy.h
@@ -55,6 +55,8 @@ PROXY_SECONDARY_METHOD2(void,
 PROXY_SECONDARY_METHOD1(void, RemoveSink, rtc::VideoSinkInterface<VideoFrame>*)
 PROXY_SECONDARY_METHOD0(void, RequestRefreshFrame)
 BYPASS_PROXY_CONSTMETHOD0(VideoTrackSourceInterface*, GetSource)
+PROXY_CONSTMETHOD0(bool, should_receive)
+PROXY_METHOD1(void, set_should_receive, bool)
 
 PROXY_METHOD1(void, RegisterObserver, ObserverInterface*)
 PROXY_METHOD1(void, UnregisterObserver, ObserverInterface*)

diff --git a/pc/video_rtp_receiver.cc b/pc/video_rtp_receiver.cc
@@ -22,6 +22,7 @@
 #include "rtc_base/location.h"
 #include "rtc_base/logging.h"
 #include "rtc_base/ref_counted_object.h"
+#include "rtc_base/task_utils/to_queued_task.h"
 
 namespace webrtc {
 
@@ -43,9 +44,12 @@ VideoRtpReceiver::VideoRtpReceiver(
           rtc::Thread::Current(),
           worker_thread,
           VideoTrack::Create(receiver_id, source_, worker_thread))),
-      attachment_id_(GenerateUniqueId()) {
+      cached_track_should_receive_(track_->should_receive()),
+      attachment_id_(GenerateUniqueId()),
+      worker_thread_safety_(PendingTaskSafetyFlag::CreateDetachedInactive()) {
   RTC_DCHECK(worker_thread_);
   SetStreams(streams);
+  track_->RegisterObserver(this);
   RTC_DCHECK_EQ(source_->state(), MediaSourceInterface::kInitializing);
 }
 
@@ -114,6 +118,40 @@ void VideoRtpReceiver::Stop() {
   track_->internal()->set_ended();
 }
 
+void VideoRtpReceiver::OnChanged() {
+  RTC_DCHECK_RUN_ON(&signaling_thread_checker_);
+  if (cached_track_should_receive_ != track_->should_receive()) {
+    cached_track_should_receive_ = track_->should_receive();
+    worker_thread_->PostTask(ToQueuedTask(
+        worker_thread_safety_,
+        [this, receive = cached_track_should_receive_]() {
+          RTC_DCHECK_RUN_ON(worker_thread_);
+          if(receive) {
+            StartMediaChannel();
+          } else {
+            StopMediaChannel();
+          }
+        }));
+  }
+}
+
+void VideoRtpReceiver::StartMediaChannel() {  
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  if (!media_channel_) {
+    return;
+  }
+  media_channel_->StartReceive(ssrc_.value_or(0));
+  OnGenerateKeyFrame();
+}
+
+void VideoRtpReceiver::StopMediaChannel() {
+  RTC_DCHECK_RUN_ON(worker_thread_);
+  if (!media_channel_) {
+    return;
+  }
+  media_channel_->StopReceive(ssrc_.value_or(0));
+}
+
 // RTC_RUN_ON(&signaling_thread_checker_)
 void VideoRtpReceiver::RestartMediaChannel(absl::optional<uint32_t> ssrc) {
   MediaSourceInterface::SourceState state = source_->state();
@@ -205,6 +243,7 @@ void VideoRtpReceiver::set_transport(
 void VideoRtpReceiver::SetStreams(
     const std::vector<rtc::scoped_refptr<MediaStreamInterface>>& streams) {
   RTC_DCHECK_RUN_ON(&signaling_thread_checker_);
+
   // Remove remote track from any streams that are going away.
   for (const auto& existing_stream : streams_) {
     bool removed = true;
@@ -276,6 +315,8 @@ void VideoRtpReceiver::SetMediaChannel_w(cricket::MediaChannel* media_channel) {
     SetEncodedSinkEnabled(false);
   }
 
+  media_channel ? worker_thread_safety_->SetAlive()
+                : worker_thread_safety_->SetNotAlive();
   media_channel_ = static_cast<cricket::VideoMediaChannel*>(media_channel);
 
   if (media_channel_) {

diff --git a/pc/video_rtp_receiver.h b/pc/video_rtp_receiver.h
@@ -43,7 +43,8 @@
 
 namespace webrtc {
 
-class VideoRtpReceiver : public RtpReceiverInternal {
+class VideoRtpReceiver : public RtpReceiverInternal,
+                         public ObserverInterface {
  public:
   // An SSRC of 0 will create a receiver that will match the first SSRC it
   // sees. Must be called on signaling thread.
@@ -61,6 +62,9 @@ class VideoRtpReceiver : public RtpReceiverInternal {
 
   rtc::scoped_refptr<VideoTrackInterface> video_track() const { return track_; }
 
+  // ObserverInterface implementation
+  void OnChanged() override;
+
   // RtpReceiverInterface implementation
   rtc::scoped_refptr<MediaStreamTrackInterface> track() const override {
     return track_;
@@ -115,6 +119,8 @@ class VideoRtpReceiver : public RtpReceiverInternal {
                          cricket::MediaChannel* media_channel);
 
  private:
+  void StartMediaChannel();
+  void StopMediaChannel();
   void RestartMediaChannel(absl::optional<uint32_t> ssrc)
       RTC_RUN_ON(&signaling_thread_checker_);
   void RestartMediaChannel_w(absl::optional<uint32_t> ssrc,
@@ -162,6 +168,8 @@ class VideoRtpReceiver : public RtpReceiverInternal {
       RTC_GUARDED_BY(&signaling_thread_checker_) = nullptr;
   bool received_first_packet_ RTC_GUARDED_BY(&signaling_thread_checker_) =
       false;
+
+  bool cached_track_should_receive_ RTC_GUARDED_BY(&signaling_thread_checker_);
   const int attachment_id_;
   rtc::scoped_refptr<FrameDecryptorInterface> frame_decryptor_
       RTC_GUARDED_BY(worker_thread_);
@@ -177,6 +185,7 @@ class VideoRtpReceiver : public RtpReceiverInternal {
   // or switched.
   bool saved_generate_keyframe_ RTC_GUARDED_BY(worker_thread_) = false;
   bool saved_encoded_sink_enabled_ RTC_GUARDED_BY(worker_thread_) = false;
+  const rtc::scoped_refptr<PendingTaskSafetyFlag> worker_thread_safety_;
 };
 
 }  // namespace webrtc

diff --git a/pc/video_track.cc b/pc/video_track.cc
@@ -18,6 +18,7 @@
 #include "rtc_base/checks.h"
 #include "rtc_base/location.h"
 #include "rtc_base/ref_counted_object.h"
+#include "rtc_base/logging.h"
 
 namespace webrtc {
 
@@ -78,6 +79,19 @@ VideoTrackSourceInterface* VideoTrack::GetSourceInternal() const {
   return video_source_->internal();
 }
 
+void VideoTrack::set_should_receive(bool receive) {
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
+  if (should_receive_ == receive)
+    return;
+  should_receive_ = receive;
+  Notifier<VideoTrackInterface>::FireOnChanged();
+}
+
+bool VideoTrack::should_receive() const {
+  RTC_DCHECK_RUN_ON(&signaling_thread_);
+  return should_receive_;
+}
+
 VideoTrackInterface::ContentHint VideoTrack::content_hint() const {
   RTC_DCHECK_RUN_ON(&signaling_thread_);
   return content_hint_;

diff --git a/pc/video_track.h b/pc/video_track.h
@@ -48,6 +48,9 @@ class VideoTrack : public MediaStreamTrack<VideoTrackInterface>,
   void RequestRefreshFrame() override;
   VideoTrackSourceInterface* GetSource() const override;
 
+  void set_should_receive(bool should_receive) override;
+  bool should_receive() const override;
+
   ContentHint content_hint() const override;
   void set_content_hint(ContentHint hint) override;
   bool set_enabled(bool enable) override;
@@ -81,6 +84,7 @@ class VideoTrack : public MediaStreamTrack<VideoTrackInterface>,
   // be queried without blocking on the worker thread by callers that don't
   // use an api proxy to call the `enabled()` method.
   bool enabled_w_ RTC_GUARDED_BY(worker_thread_) = true;
+  bool should_receive_ RTC_GUARDED_BY(signaling_thread_) = true;
 };
 
 }  // namespace webrtc

diff --git a/sdk/android/api/org/webrtc/VideoTrack.java b/sdk/android/api/org/webrtc/VideoTrack.java
@@ -54,6 +54,24 @@ public void removeSink(VideoSink sink) {
     }
   }
 
+  /**
+   * For a remote video track, starts/stops receiving the video stream.
+   * 
+   * If this is a local video track, this is a no-op.
+   */
+  public void setShouldReceive(boolean shouldReceive){
+    nativeSetShouldReceive(getNativeMediaStreamTrack(), shouldReceive);
+  }
+
+  /**
+   * The current receive status for a remote video track.
+   * 
+   * This has no meaning for a local video track.
+   */
+  public boolean shouldReceive(){
+    return nativeGetShouldReceive(getNativeMediaStreamTrack());
+  }
+
   @Override
   public void dispose() {
     for (long nativeSink : sinks.values()) {
@@ -73,4 +91,6 @@ long getNativeVideoTrack() {
   private static native void nativeRemoveSink(long track, long nativeSink);
   private static native long nativeWrapSink(VideoSink sink);
   private static native void nativeFreeSink(long sink);
+  private static native void nativeSetShouldReceive(long track, boolean shouldReceive);
+  private static native boolean nativeGetShouldReceive(long track);
 }
diff --git a/sdk/android/src/jni/video_track.cc b/sdk/android/src/jni/video_track.cc
@@ -45,5 +45,16 @@ static void JNI_VideoTrack_FreeSink(JNIEnv* jni,
   delete reinterpret_cast<rtc::VideoSinkInterface<VideoFrame>*>(j_native_sink);
 }
 
+static void JNI_VideoTrack_SetShouldReceive(JNIEnv* jni,
+                                      jlong j_native_track,
+                                      jboolean should_receive) {
+  reinterpret_cast<VideoTrackInterface*>(j_native_track)->set_should_receive(should_receive);
+}
+
+static jboolean JNI_VideoTrack_GetShouldReceive(JNIEnv* jni,
+                                      jlong j_native_track) {
+  return reinterpret_cast<VideoTrackInterface*>(j_native_track)->should_receive();
+}
+
 }  // namespace jni
 }  // namespace webrtc
diff --git a/sdk/objc/api/peerconnection/RTCVideoTrack.h b/sdk/objc/api/peerconnection/RTCVideoTrack.h
@@ -25,6 +25,9 @@ RTC_OBJC_EXPORT
 /** The video source for this video track. */
 @property(nonatomic, readonly) RTC_OBJC_TYPE(RTCVideoSource) *source;
 
+/** The receive state, if this is a remote video track. */
+@property(nonatomic, assign) BOOL shouldReceive;
+
 - (instancetype)init NS_UNAVAILABLE;
 
 /** Register a renderer that will render all frames received on this track. */

diff --git a/sdk/objc/api/peerconnection/RTCVideoTrack.mm b/sdk/objc/api/peerconnection/RTCVideoTrack.mm
@@ -68,6 +68,14 @@ - (void)dealloc {
   return _source;
 }
 
+- (BOOL)shouldReceive {
+  return self.nativeVideoTrack->should_receive();
+}
+
+- (void)setShouldReceive:(BOOL)shouldReceive {
+  self.nativeVideoTrack->set_should_receive(shouldReceive);
+}
+
 - (void)addRenderer:(id<RTC_OBJC_TYPE(RTCVideoRenderer)>)renderer {
   // Make sure we don't have this renderer yet.
   for (RTCVideoRendererAdapter *adapter in _adapters) {

diff --git a/video/video_receive_stream2.cc b/video/video_receive_stream2.cc
@@ -361,6 +361,7 @@ void VideoReceiveStream2::Start() {
     return;
   }
 
+  frame_buffer_->Start();
   const bool protected_by_fec = config_.rtp.protected_by_flexfec ||
                                 rtp_video_stream_receiver_.IsUlpfecEnabled();