Peer transports and observers

 package io.livekit.android.room
 package io.livekit.android.room
 
 
 import dagger.assisted.Assisted
 import dagger.assisted.Assisted
+import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import dagger.assisted.AssistedInject
 import org.webrtc.*
 import org.webrtc.*
 
 
 class PeerConnectionTransport
 class PeerConnectionTransport
 @AssistedInject
 @AssistedInject
 constructor(
 constructor(
-    config: PeerConnection.RTCConfiguration,
-    listener: PeerConnection.Observer,
-    @Assisted connectionFactory: PeerConnectionFactory
+    @Assisted config: PeerConnection.RTCConfiguration,
+    @Assisted listener: PeerConnection.Observer,
+    connectionFactory: PeerConnectionFactory
 ) {
 ) {
     val peerConnection: PeerConnection = connectionFactory.createPeerConnection(
     val peerConnection: PeerConnection = connectionFactory.createPeerConnection(
         config,
         config,
@@ -49,4 +50,12 @@ constructor(
     fun close() {
     fun close() {
         peerConnection.close()
         peerConnection.close()
     }
     }
+
+    @AssistedFactory
+    interface Factory {
+        fun create(
+            config: PeerConnection.RTCConfiguration,
+            listener: PeerConnection.Observer
+        ): PeerConnectionTransport
+    }
 }
 }

+package io.livekit.android.room
+
+import livekit.Rtc
+import org.webrtc.*
+
+class PublisherTransportObserver(
+    private val engine: RTCEngine
+) : PeerConnection.Observer {
+
+    override fun onIceCandidate(candidate: IceCandidate?) {
+        val candidate = candidate ?: return
+        if (engine.rtcConnected) {
+            engine.client.sendCandidate(candidate, target = Rtc.SignalTarget.PUBLISHER)
+        } else {
+            engine.pendingCandidates.add(candidate)
+        }
+    }
+
+    override fun onRenegotiationNeeded() {
+        engine.negotiate()
+    }
+
+    override fun onIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
+        val newState = newState ?: throw NullPointerException("unexpected null new state, what do?")
+        if (newState == PeerConnection.IceConnectionState.CONNECTED && !engine.iceConnected) {
+            engine.iceConnected = true
+        } else if (newState == PeerConnection.IceConnectionState.DISCONNECTED) {
+            engine.iceConnected = false
+            engine.listener?.onDisconnect("Peer connection disconnected")
+        }
+    }
+
+    override fun onStandardizedIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
+    }
+
+    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
+    }
+
+    override fun onSelectedCandidatePairChanged(event: CandidatePairChangeEvent?) {
+    }
+
+
+    override fun onSignalingChange(p0: PeerConnection.SignalingState?) {
+    }
+
+    override fun onIceConnectionReceivingChange(p0: Boolean) {
+    }
+
+    override fun onIceGatheringChange(p0: PeerConnection.IceGatheringState?) {
+    }
+
+    override fun onIceCandidatesRemoved(p0: Array<out IceCandidate>?) {
+    }
+
+    override fun onAddStream(p0: MediaStream?) {
+    }
+
+    override fun onRemoveStream(p0: MediaStream?) {
+    }
+
+    override fun onDataChannel(p0: DataChannel?) {
+    }
+
+    override fun onTrack(transceiver: RtpTransceiver?) {
+    }
+
+    override fun onAddTrack(p0: RtpReceiver?, p1: Array<out MediaStream>?) {
+    }
+}

 package io.livekit.android.room
 package io.livekit.android.room
 
 
-import org.webrtc.PeerConnection
+import livekit.Model
+import livekit.Rtc
+import org.webrtc.*
 import javax.inject.Inject
 import javax.inject.Inject
 
 
 class RTCEngine
 class RTCEngine
 @Inject
 @Inject
 constructor(
 constructor(
-    private val client: RTCClient
+    val client: RTCClient,
+    pctFactory: PeerConnectionTransport.Factory,
 ) {
 ) {
 
 
+    var listener: Listener? = null
+    var rtcConnected: Boolean = false
+    var joinResponse: Rtc.JoinResponse? = null
+    var iceConnected: Boolean = false
+        set(value) {
+            field = value
+            if (field) {
+                listener?.onJoin(joinResponse)
+                joinResponse = null
+            }
+        }
+    val pendingCandidates = mutableListOf<IceCandidate>()
+
+    private val publisherObserver = PublisherTransportObserver(this)
+    private val subscriberObserver = SubscriberTransportObserver(this)
+    private val publisher: PeerConnectionTransport
+    private val subscriber: PeerConnectionTransport
+
     init {
     init {
         val rtcConfig = PeerConnection.RTCConfiguration(RTCClient.DEFAULT_ICE_SERVERS).apply {
         val rtcConfig = PeerConnection.RTCConfiguration(RTCClient.DEFAULT_ICE_SERVERS).apply {
             sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN
             sdpSemantics = PeerConnection.SdpSemantics.UNIFIED_PLAN
             continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
             continualGatheringPolicy = PeerConnection.ContinualGatheringPolicy.GATHER_CONTINUALLY
         }
         }
 
 
+        publisher = pctFactory.create(rtcConfig, publisherObserver)
+        subscriber = pctFactory.create(rtcConfig, subscriberObserver)
+
     }
     }
 
 
     suspend fun join(url: String, token: String, isSecure: Boolean) {
     suspend fun join(url: String, token: String, isSecure: Boolean) {
         client.join(url, token, isSecure)
         client.join(url, token, isSecure)
     }
     }
+
+    fun negotiate() {
+        TODO("Not yet implemented")
+    }
+
+    interface Listener {
+        fun onJoin(response: Rtc.JoinResponse?)
+        fun onAddTrack(track: MediaStreamTrack, streams: Array<out MediaStream>)
+        fun onPublishLocalTrack(cid: String, track: Model.TrackInfo)
+        fun onAddDataChannel(channel: DataChannel)
+        fun onUpdateParticipants(updates: Array<out Model.ParticipantInfo>)
+        fun onUpdateSpeakers(speakers: Array<out Rtc.SpeakerInfo>)
+        fun onDisconnect(reason: String)
+        fun onFailToConnect(error: Error)
+    }
 }
 }

@@ -11,7 +11,7 @@ constructor(
     @Assisted private val engine: RTCEngine,
     @Assisted private val engine: RTCEngine,
 ) {
 ) {
 
 
-    suspend fun connect() {
-        engine.join(connectOptions)
+    suspend fun connect(url: String, token: String, isSecure: Boolean) {
+        engine.join(url, token, isSecure)
     }
     }
 }
 }

+package io.livekit.android.room
+
+import com.github.ajalt.timberkt.Timber
+import livekit.Rtc
+import org.webrtc.*
+
+class SubscriberTransportObserver(
+    private val engine: RTCEngine
+) : PeerConnection.Observer {
+
+
+    override fun onIceCandidate(candidate: IceCandidate) {
+        engine.client.sendCandidate(candidate, Rtc.SignalTarget.SUBSCRIBER)
+    }
+
+    override fun onAddTrack(receiver: RtpReceiver, streams: Array<out MediaStream>) {
+        val track = receiver.track() ?: return
+        engine.listener?.onAddTrack(track, streams)
+    }
+
+    override fun onTrack(transceiver: RtpTransceiver) {
+        when (transceiver.mediaType) {
+            MediaStreamTrack.MediaType.MEDIA_TYPE_AUDIO -> Timber.v { "peerconn started receiving audio" }
+            MediaStreamTrack.MediaType.MEDIA_TYPE_VIDEO -> Timber.v { "peerconn started receiving video" }
+            else -> Timber.d { "peerconn started receiving unknown media type: ${transceiver.mediaType}" }
+        }
+    }
+
+    override fun onDataChannel(channel: DataChannel) {
+        engine.listener?.onAddDataChannel(channel)
+    }
+
+    override fun onStandardizedIceConnectionChange(newState: PeerConnection.IceConnectionState?) {
+    }
+
+    override fun onConnectionChange(newState: PeerConnection.PeerConnectionState?) {
+    }
+
+    override fun onSelectedCandidatePairChanged(event: CandidatePairChangeEvent?) {
+    }
+
+    override fun onSignalingChange(p0: PeerConnection.SignalingState?) {
+    }
+
+    override fun onIceConnectionChange(p0: PeerConnection.IceConnectionState?) {
+    }
+
+    override fun onIceConnectionReceivingChange(p0: Boolean) {
+    }
+
+    override fun onIceGatheringChange(p0: PeerConnection.IceGatheringState?) {
+    }
+
+    override fun onIceCandidatesRemoved(p0: Array<out IceCandidate>?) {
+    }
+
+    override fun onAddStream(p0: MediaStream?) {
+    }
+
+    override fun onRemoveStream(p0: MediaStream?) {
+    }
+
+    override fun onRenegotiationNeeded() {
+    }
+
+}