Render video directly from capturer for local video tracks (#419)

* Render video directly from capturer for local video tracks

* spotless

@@ -121,7 +121,7 @@ internal constructor(
         options: LocalVideoTrackOptions = videoTrackCaptureDefaults.copy(),
         videoProcessor: VideoProcessor? = null,
     ): LocalVideoTrack {
-        return LocalVideoTrack.createTrack(
+        return LocalVideoTrack.createCameraTrack(
             peerConnectionFactory,
             context,
             name,

@@ -31,6 +31,7 @@ import io.livekit.android.room.track.video.CameraCapturerUtils.createCameraEnume
 import io.livekit.android.room.track.video.CameraCapturerUtils.findCamera
 import io.livekit.android.room.track.video.CameraCapturerUtils.getCameraPosition
 import io.livekit.android.room.track.video.CameraCapturerWithSize
+import io.livekit.android.room.track.video.CaptureDispatchObserver
 import io.livekit.android.room.track.video.VideoCapturerWithSize
 import io.livekit.android.room.util.EncodingUtils
 import io.livekit.android.util.FlowObservable
@@ -50,6 +51,7 @@ import livekit.org.webrtc.RtpTransceiver
 import livekit.org.webrtc.SurfaceTextureHelper
 import livekit.org.webrtc.VideoCapturer
 import livekit.org.webrtc.VideoProcessor
+import livekit.org.webrtc.VideoSink
 import livekit.org.webrtc.VideoSource
 import java.util.UUID
 import livekit.LivekitModels.VideoQuality as ProtoVideoQuality
@@ -72,6 +74,11 @@ constructor(
     private val eglBase: EglBase,
     private val defaultsManager: DefaultsManager,
     private val trackFactory: Factory,
+    /**
+     * If this is assigned, you must ensure that this observer is associated with the [capturer],
+     * as this will be used to receive frames in [addRenderer].
+     **/
+    @Assisted private var dispatchObserver: CaptureDispatchObserver? = null,
 ) : VideoTrack(name, rtcTrack) {
 
     override var rtcTrack: livekit.org.webrtc.VideoTrack = rtcTrack
@@ -126,6 +133,22 @@ constructor(
         closeableManager.close()
     }
 
+    override fun addRenderer(renderer: VideoSink) {
+        if (dispatchObserver != null) {
+            dispatchObserver?.registerSink(renderer)
+        } else {
+            super.addRenderer(renderer)
+        }
+    }
+
+    override fun removeRenderer(renderer: VideoSink) {
+        if (dispatchObserver != null) {
+            dispatchObserver?.unregisterSink(renderer)
+        } else {
+            super.removeRenderer(renderer)
+        }
+    }
+
     fun setDeviceId(deviceId: String) {
         restartTrack(options.copy(deviceId = deviceId))
     }
@@ -239,7 +262,7 @@ constructor(
         val oldCloseable = closeableManager.unregisterResource(oldRtcTrack)
         oldCloseable?.close()
 
-        val newTrack = createTrack(
+        val newTrack = createCameraTrack(
             peerConnectionFactory,
             context,
             name,
@@ -386,85 +409,78 @@ constructor(
             name: String,
             options: LocalVideoTrackOptions,
             rtcTrack: livekit.org.webrtc.VideoTrack,
+            dispatchObserver: CaptureDispatchObserver?,
         ): LocalVideoTrack
     }
 
     companion object {
 
-        internal fun createTrack(
+        internal fun createCameraTrack(
             peerConnectionFactory: PeerConnectionFactory,
             context: Context,
             name: String,
-            capturer: VideoCapturer,
-            options: LocalVideoTrackOptions = LocalVideoTrackOptions(),
+            options: LocalVideoTrackOptions,
             rootEglBase: EglBase,
             trackFactory: Factory,
             videoProcessor: VideoProcessor? = null,
         ): LocalVideoTrack {
-            val source = peerConnectionFactory.createVideoSource(false)
-            source.setVideoProcessor(videoProcessor)
-            val surfaceTextureHelper = SurfaceTextureHelper.create("VideoCaptureThread", rootEglBase.eglBaseContext)
-            capturer.initialize(
-                surfaceTextureHelper,
-                context,
-                source.capturerObserver,
-            )
-            val rtcTrack = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source)
+            if (ContextCompat.checkSelfPermission(context, Manifest.permission.CAMERA) !=
+                PackageManager.PERMISSION_GRANTED
+            ) {
+                throw SecurityException("Camera permissions are required to create a camera video track.")
+            }
 
-            val track = trackFactory.create(
-                capturer = capturer,
-                source = source,
-                options = options,
-                name = name,
-                rtcTrack = rtcTrack,
-            )
+            val (capturer, newOptions) = CameraCapturerUtils.createCameraCapturer(context, options) ?: TODO()
 
-            track.closeableManager.registerResource(
-                rtcTrack,
-                SurfaceTextureHelperCloser(surfaceTextureHelper),
+            return createTrack(
+                peerConnectionFactory = peerConnectionFactory,
+                context = context,
+                name = name,
+                capturer = capturer,
+                options = newOptions,
+                rootEglBase = rootEglBase,
+                trackFactory = trackFactory,
+                videoProcessor = videoProcessor,
             )
-            return track
         }
 
         internal fun createTrack(
             peerConnectionFactory: PeerConnectionFactory,
             context: Context,
             name: String,
-            options: LocalVideoTrackOptions,
+            capturer: VideoCapturer,
+            options: LocalVideoTrackOptions = LocalVideoTrackOptions(),
             rootEglBase: EglBase,
             trackFactory: Factory,
             videoProcessor: VideoProcessor? = null,
         ): LocalVideoTrack {
-            if (ContextCompat.checkSelfPermission(context, Manifest.permission.CAMERA) !=
-                PackageManager.PERMISSION_GRANTED
-            ) {
-                throw SecurityException("Camera permissions are required to create a camera video track.")
-            }
-
             val source = peerConnectionFactory.createVideoSource(options.isScreencast)
             source.setVideoProcessor(videoProcessor)
-            val (capturer, newOptions) = CameraCapturerUtils.createCameraCapturer(context, options) ?: TODO()
+
             val surfaceTextureHelper = SurfaceTextureHelper.create("VideoCaptureThread", rootEglBase.eglBaseContext)
+            val dispatchObserver = CaptureDispatchObserver()
+            dispatchObserver.registerObserver(source.capturerObserver)
+
             capturer.initialize(
                 surfaceTextureHelper,
                 context,
-                source.capturerObserver,
+                dispatchObserver,
             )
             val rtcTrack = peerConnectionFactory.createVideoTrack(UUID.randomUUID().toString(), source)
 
             val track = trackFactory.create(
                 capturer = capturer,
                 source = source,
-                options = newOptions,
+                options = options,
                 name = name,
                 rtcTrack = rtcTrack,
+                dispatchObserver = dispatchObserver,
             )
 
             track.closeableManager.registerResource(
                 rtcTrack,
                 SurfaceTextureHelperCloser(surfaceTextureHelper),
             )
-
             return track
         }
     }

+/*
+ * Copyright 2024 LiveKit, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.livekit.android.room.track.video
+
+import livekit.org.webrtc.CapturerObserver
+import livekit.org.webrtc.VideoFrame
+import livekit.org.webrtc.VideoSink
+
+class CaptureDispatchObserver : CapturerObserver {
+    private val observers = linkedSetOf<CapturerObserver>()
+    private val sinks = linkedSetOf<VideoSink>()
+
+    @Synchronized
+    fun registerObserver(observer: CapturerObserver) {
+        observers.add(observer)
+    }
+
+    @Synchronized
+    fun unregisterObserver(observer: CapturerObserver) {
+        observers.remove(observer)
+    }
+
+    @Synchronized
+    fun registerSink(sink: VideoSink) {
+        sinks.add(sink)
+    }
+
+    @Synchronized
+    fun unregisterSink(sink: VideoSink) {
+        sinks.remove(sink)
+    }
+
+    @Synchronized
+    override fun onCapturerStarted(success: Boolean) {
+        for (observer in observers) {
+            observer.onCapturerStarted(success)
+        }
+    }
+
+    @Synchronized
+    override fun onCapturerStopped() {
+        for (observer in observers) {
+            observer.onCapturerStopped()
+        }
+    }
+
+    @Synchronized
+    override fun onFrameCaptured(frame: VideoFrame) {
+        for (observer in observers) {
+            observer.onFrameCaptured(frame)
+        }
+
+        for (sink in sinks) {
+            sink.onFrame(frame)
+        }
+    }
+}