flesh out room

@@ -69,6 +69,7 @@ dependencies {
     kapt 'com.google.dagger:dagger-compiler:2.33'
 
     implementation 'com.github.ajalt:timberkt:1.5.1'
+    implementation 'com.vdurmont:semver4j:3.1.0'
 
     testImplementation 'junit:junit:4.12'
     androidTestImplementation 'androidx.test.ext:junit:1.1.2'

@@ -237,7 +237,7 @@ constructor(
         fun onParticipantUpdate(updates: List<Model.ParticipantInfo>)
         fun onActiveSpeakersChanged(speakers: List<Rtc.SpeakerInfo>)
         fun onClose(reason: String, code: Int)
-        fun onError(error: Error)
+        fun onError(error: Exception)
     }
 
     companion object {

@@ -37,7 +37,8 @@ constructor(
         set(value) {
             field = value
             if (field) {
-                listener?.onJoin(joinResponse)
+                // TODO get rid of this assertion
+                listener?.onJoin(joinResponse!!)
                 joinResponse = null
             }
         }
@@ -127,14 +128,14 @@ constructor(
     }
 
     interface Listener {
-        fun onJoin(response: Rtc.JoinResponse?)
+        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: List<Model.ParticipantInfo>)
         fun onUpdateSpeakers(speakers: List<Rtc.SpeakerInfo>)
         fun onDisconnect(reason: String)
-        fun onFailToConnect(error: Error)
+        fun onFailToConnect(error: Exception)
     }
 
     companion object {
@@ -280,7 +281,7 @@ constructor(
         listener?.onDisconnect(reason)
     }
 
-    override fun onError(error: Error) {
+    override fun onError(error: Exception) {
         listener?.onFailToConnect(error)
     }
 }
\ No newline at end of file

 package io.livekit.android.room
 
+import com.github.ajalt.timberkt.Timber
+import com.vdurmont.semver4j.Semver
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
 import io.livekit.android.ConnectOptions
 import io.livekit.android.room.participant.LocalParticipant
+import io.livekit.android.room.participant.Participant
 import io.livekit.android.room.participant.RemoteParticipant
+import io.livekit.android.room.track.Track
+import io.livekit.android.room.util.unpackedTrackLabel
+import livekit.Model
+import livekit.Rtc
+import org.webrtc.DataChannel
+import org.webrtc.MediaStream
+import org.webrtc.MediaStreamTrack
 
 class Room
 @AssistedInject
 constructor(
     @Assisted private val connectOptions: ConnectOptions,
     private val engine: RTCEngine,
-) {
+) : RTCEngine.Listener {
+    init {
+        engine.listener = this
+    }
 
     enum class State {
         CONNECTING,
@@ -31,13 +44,89 @@ constructor(
         private set
     var state: State = State.DISCONNECTED
         private set
-    var localParticipant: LocalParticipant? = TODO()
+    var localParticipant: LocalParticipant? = null
+        private set
+    private val mutableRemoteParticipants = mutableMapOf<Participant.Sid, RemoteParticipant>()
+    val remoteParticipants: Map<Participant.Sid, RemoteParticipant>
+        get() = mutableRemoteParticipants
 
+    private val mutableActiveSpeakers = mutableListOf<Participant>()
+    val activeSpeakers: List<Participant>
+        get() = mutableActiveSpeakers
 
     suspend fun connect(url: String, token: String, isSecure: Boolean) {
+        if (localParticipant != null) {
+            Timber.d { "Attempting to connect to room when already connected." }
+            return
+        }
         engine.join(url, token, isSecure)
     }
 
+    fun disconnect() {
+        engine.close()
+        state = State.DISCONNECTED
+        listener?.onDisconnect(this, null)
+    }
+
+    private fun handleParticipantDisconnect(sid: Participant.Sid, participant: RemoteParticipant) {
+        val removedParticipant = mutableRemoteParticipants.remove(sid) ?: return
+        removedParticipant.tracks.values.forEach { publication ->
+            removedParticipant.unpublishTrack(publication.trackSid)
+        }
+
+        listener?.onparticipantDisconnected(this, removedParticipant)
+    }
+
+    private fun getOrCreateRemoteParticipant(
+        sid: Participant.Sid,
+        info: Model.ParticipantInfo? = null
+    ): RemoteParticipant {
+        var participant = remoteParticipants[sid]
+        if (participant != null) {
+            return participant
+        }
+
+        participant = if (info != null) {
+            RemoteParticipant(info)
+        } else {
+            RemoteParticipant(sid, null)
+        }
+        mutableRemoteParticipants[sid] = participant
+        return participant
+    }
+
+    private fun handleSpeakerUpdate(speakerInfos: List<Rtc.SpeakerInfo>) {
+        val speakers = mutableListOf<Participant>()
+        val seenSids = mutableSetOf<Participant.Sid>()
+        val localParticipant = localParticipant
+        speakerInfos.forEach { speakerInfo ->
+            val speakerSid = Participant.Sid(speakerInfo.sid)
+            seenSids.add(speakerSid)
+
+            if (speakerSid == localParticipant?.sid) {
+                localParticipant.audioLevel = speakerInfo.level
+                speakers.add(localParticipant)
+            } else {
+                val participant = remoteParticipants[speakerSid]
+                if (participant != null) {
+                    participant.audioLevel = speakerInfo.level
+                    speakers.add(participant)
+                }
+            }
+        }
+
+        if (localParticipant != null && seenSids.contains(localParticipant.sid)) {
+            localParticipant.audioLevel = 0.0f
+        }
+        remoteParticipants.values
+            .filterNot { seenSids.contains(it.sid) }
+            .forEach { it.audioLevel = 0.0f }
+
+        mutableActiveSpeakers.clear()
+        mutableActiveSpeakers.addAll(speakers)
+        listener?.onActiveSpeakersChanged(speakers, this)
+    }
+
     @AssistedFactory
     interface Factory {
         fun create(connectOptions: ConnectOptions): Room
@@ -45,17 +134,95 @@ constructor(
 
     interface Listener {
         fun onConnect(room: Room)
-        fun onDisconnect(room: Room, error: Exception)
-        fun onParticipantDidConnect(room: Room, participant: RemoteParticipant)
-    }
-//    func didConnect(room: Room)
-//    func didDisconnect(room: Room, error: Error?)
-//    func participantDidConnect(room: Room, participant: RemoteParticipant)
-//    func participantDidDisconnect(room: Room, participant: RemoteParticipant)
-//    func didFailToConnect(room: Room, error: Error)
-//    func isReconnecting(room: Room, error: Error)
-//    func didReconnect(room: Room)
-//    func didStartRecording(room: Room)
-//    func didStopRecording(room: Room)
-//    func activeSpeakersDidChange(speakers: [Participant], room: Room)
+        fun onDisconnect(room: Room, error: Exception?)
+        fun onParticipantConnected(room: Room, participant: RemoteParticipant)
+        fun onparticipantDisconnected(room: Room, participant: RemoteParticipant)
+        fun onFailedToConnect(room: Room, error: Exception)
+        fun onReconnecting(room: Room, error: Exception)
+        fun onReconnect(room: Room)
+        fun onStartRecording(room: Room)
+        fun onStopRecording(room: Room)
+        fun onActiveSpeakersChanged(speakers: List<Participant>, room: Room)
+    }
+
+    override fun onJoin(response: Rtc.JoinResponse) {
+        Timber.v { "engine did join, version: ${response.serverVersion}" }
+
+        try {
+            val serverVersion = Semver(response.serverVersion)
+            if (serverVersion.major == 0 && serverVersion.minor < 5) {
+                Timber.e { "This version of livekit requires server version >= 0.5.x" }
+                return
+            }
+        } catch (e: Exception) {
+            Timber.e { "Unable to parse server version!" }
+            return
+        }
+        state = State.CONNECTED
+        sid = Sid(response.room.sid)
+        name = response.room.name
+
+        if (response.hasParticipant()) {
+            localParticipant = LocalParticipant(response.participant, engine)
+        }
+        if (response.otherParticipantsList.isNotEmpty()) {
+            response.otherParticipantsList.forEach {
+                getOrCreateRemoteParticipant(Participant.Sid(it.sid), it)
+            }
+        }
+
+        listener?.onConnect(this)
+    }
+
+    override fun onAddTrack(track: MediaStreamTrack, streams: Array<out MediaStream>) {
+        if (streams.count() < 0) {
+            Timber.i { "add track with empty streams?" }
+            return
+        }
+
+        val participantSid = Participant.Sid(streams.first().id)
+        val trackSid = Track.Sid(track.id())
+        val participant = getOrCreateRemoteParticipant(participantSid)
+        participant.addSubscribedMediaTrack(track, trackSid)
+    }
+
+    override fun onAddDataChannel(channel: DataChannel) {
+        val unpackedTrackLabel = channel.unpackedTrackLabel()
+        val (participantSid, trackSid, name) = unpackedTrackLabel
+        val participant = getOrCreateRemoteParticipant(participantSid)
+        participant.addSubscribedDataTrack(channel, trackSid, name)
+    }
+
+    override fun onPublishLocalTrack(cid: String, track: Model.TrackInfo) {
+    }
+
+
+    override fun onUpdateParticipants(updates: List<Model.ParticipantInfo>) {
+        for (info in updates) {
+            val participantSid = Participant.Sid(info.sid)
+            val isNewParticipant = remoteParticipants.contains(participantSid)
+            val participant = getOrCreateRemoteParticipant(participantSid, info)
+
+            if (info.state == Model.ParticipantInfo.State.DISCONNECTED) {
+                handleParticipantDisconnect(participantSid, participant)
+            } else if (isNewParticipant) {
+                listener?.onParticipantConnected(this, participant)
+            } else {
+                participant.updateFromInfo(info)
+            }
+        }
+    }
+
+    override fun onUpdateSpeakers(speakers: List<Rtc.SpeakerInfo>) {
+        handleSpeakerUpdate(speakers)
+    }
+
+    override fun onDisconnect(reason: String) {
+        Timber.v { "engine did disconnect: $reason" }
+        listener?.onDisconnect(this, null)
+    }
+
+    override fun onFailToConnect(error: Exception) {
+        listener?.onFailedToConnect(this, error)
+    }
 }
\ No newline at end of file

@@ -173,7 +173,7 @@ class RemoteParticipant(
         listener?.onSubscribe(dataTrack = publication, participant = this)
     }
 
-    private fun unpublishTrack(trackSid: Track.Sid, sendUnpublish: Boolean) {
+    fun unpublishTrack(trackSid: Track.Sid, sendUnpublish: Boolean = false) {
         val publication = tracks.remove(trackSid) ?: return
 
         when (publication) {

+package io.livekit.android.room.util
+
+import io.livekit.android.room.participant.Participant
+import io.livekit.android.room.track.Track
+import org.webrtc.DataChannel
+
+fun DataChannel.unpackedTrackLabel(): Triple<Participant.Sid, Track.Sid, String> {
+    val parts = label().split("|")
+    val participantSid: Participant.Sid
+    val trackSid: Track.Sid
+    val name: String
+
+    if (parts.count() == 3) {
+        participantSid = Participant.Sid(parts[0])
+        trackSid = Track.Sid(parts[1])
+        name = parts[2]
+    } else {
+        participantSid = Participant.Sid("")
+        trackSid = Track.Sid("")
+        name = ""
+    }
+    
+    return Triple(participantSid, trackSid, name)
+}
\ No newline at end of file