Protocol 7 support: server unpublish local track and permission change updates (#72)

 package io.livekit.android.events
 
-import io.livekit.android.room.Room
 import io.livekit.android.room.participant.LocalParticipant
 import io.livekit.android.room.participant.Participant
+import io.livekit.android.room.participant.ParticipantPermission
 import io.livekit.android.room.participant.RemoteParticipant
 import io.livekit.android.room.track.LocalTrackPublication
 import io.livekit.android.room.track.RemoteTrackPublication
@@ -118,4 +118,15 @@ sealed class ParticipantEvent(open val participant: Participant) : Event() {
         val trackPublication: RemoteTrackPublication,
         val subscriptionAllowed: Boolean
     ) : ParticipantEvent(participant)
+
+    /**
+     * A participant's permissions have changed.
+     *
+     * Currently only fires for the local participant.
+     */
+    class ParticipantPermissionsChanged(
+        override val participant: Participant,
+        val newPermissions: ParticipantPermission?,
+        val oldPermissions: ParticipantPermission?,
+    ) : ParticipantEvent(participant)
 }
\ No newline at end of file

 package io.livekit.android.events
 
 import io.livekit.android.room.Room
-import io.livekit.android.room.participant.ConnectionQuality
-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.participant.*
 import io.livekit.android.room.track.LocalTrackPublication
 import io.livekit.android.room.track.RemoteTrackPublication
 import io.livekit.android.room.track.Track
@@ -157,4 +154,15 @@ sealed class RoomEvent(val room: Room) : Event() {
 
     class FailedToConnect(room: Room, val error: Throwable) : RoomEvent(room)
 
+    /**
+     * A participant's permissions have changed.
+     *
+     * Currently only fires for the local participant.
+     */
+    class ParticipantPermissionsChanged(
+        room: Room,
+        val participant: Participant,
+        val newPermissions: ParticipantPermission?,
+        val oldPermissions: ParticipantPermission?,
+    ) : RoomEvent(room)
 }
\ No newline at end of file

@@ -519,6 +519,7 @@ internal constructor(
         fun onSignalConnected(isResume: Boolean)
         fun onFullReconnecting()
         suspend fun onPostReconnect(isFullReconnect: Boolean)
+        fun onLocalTrackUnpublished(trackUnpublished: LivekitRtc.TrackUnpublishedResponse)
     }
 
     companion object {
@@ -671,6 +672,10 @@ internal constructor(
         sessionToken = token
     }
 
+    override fun onLocalTrackUnpublished(trackUnpublished: LivekitRtc.TrackUnpublishedResponse) {
+        listener?.onLocalTrackUnpublished(trackUnpublished)
+    }
+
     //--------------------------------- DataChannel.Observer ------------------------------------//
 
     override fun onBufferedAmountChange(previousAmount: Long) {

@@ -197,6 +197,23 @@ constructor(
         if (_localParticipant == null) {
             val lp = localParticipantFactory.create(response.participant, dynacast)
             lp.internalListener = this
+            coroutineScope.launch {
+                lp.events.collect {
+                    when (it) {
+                        is ParticipantEvent.ParticipantPermissionsChanged -> eventBus.postEvent(
+                            RoomEvent.ParticipantPermissionsChanged(
+                                room = this@Room,
+                                participant = it.participant,
+                                newPermissions = it.newPermissions,
+                                oldPermissions = it.oldPermissions,
+                            )
+                        )
+                        else -> {
+                            /* do nothing */
+                        }
+                    }
+                }
+            }
             _localParticipant = lp
         } else {
             localParticipant.updateFromInfo(response.participant)
@@ -264,6 +281,17 @@ constructor(
                             it.subscriptionAllowed
                         )
                     )
+                    is ParticipantEvent.ParticipantPermissionsChanged -> eventBus.postEvent(
+                        RoomEvent.ParticipantPermissionsChanged(
+                            room = this@Room,
+                            participant = it.participant,
+                            newPermissions = it.newPermissions,
+                            oldPermissions = it.oldPermissions,
+                        )
+                    )
+                    else -> {
+                        /* do nothing */
+                    }
                 }
             }
         }
@@ -651,7 +679,7 @@ constructor(
                 val pubs = participant.tracks.values.toList()
                 for (pub in pubs) {
                     val remotePub = pub as? RemoteTrackPublication ?: continue
-                    if(remotePub.subscribed) {
+                    if (remotePub.subscribed) {
                         remotePub.sendUpdateTrackSettings.invoke()
                     }
                 }
@@ -659,6 +687,13 @@ constructor(
         }
     }
 
+    /**
+     * @suppress
+     */
+    override fun onLocalTrackUnpublished(trackUnpublished: LivekitRtc.TrackUnpublishedResponse) {
+        localParticipant.handleLocalTrackUnpublished(trackUnpublished)
+    }
+
     //------------------------------- ParticipantListener --------------------------------//
     /**
      * This is called for both Local and Remote participants

@@ -525,6 +525,9 @@ constructor(
             LivekitRtc.SignalResponse.MessageCase.REFRESH_TOKEN -> {
                 listener?.onRefreshToken(response.refreshToken)
             }
+            LivekitRtc.SignalResponse.MessageCase.TRACK_UNPUBLISHED -> {
+                listener?.onLocalTrackUnpublished(response.trackUnpublished)
+            }
             LivekitRtc.SignalResponse.MessageCase.MESSAGE_NOT_SET,
             null -> {
                 LKLog.v { "empty messageCase!" }
@@ -567,6 +570,7 @@ constructor(
         fun onSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate)
         fun onSubscriptionPermissionUpdate(subscriptionPermissionUpdate: LivekitRtc.SubscriptionPermissionUpdate)
         fun onRefreshToken(token: String)
+        fun onLocalTrackUnpublished(trackUnpublished: LivekitRtc.TrackUnpublishedResponse)
     }
 
     companion object {
@@ -583,7 +587,7 @@ constructor(
         const val SD_TYPE_ANSWER = "answer"
         const val SD_TYPE_OFFER = "offer"
         const val SD_TYPE_PRANSWER = "pranswer"
-        const val PROTOCOL_VERSION = 6
+        const val PROTOCOL_VERSION = 7
         const val SDK_TYPE = "android"
 
         private val skipQueueTypes = listOf(

@@ -474,15 +474,12 @@ internal constructor(
         }
     }
 
-    /**
-     * @suppress
-     */
-    fun onRemoteMuteChanged(trackSid: String, muted: Boolean) {
+    internal fun onRemoteMuteChanged(trackSid: String, muted: Boolean) {
         val pub = tracks[trackSid]
         pub?.muted = muted
     }
 
-    fun handleSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate) {
+    internal fun handleSubscribedQualityUpdate(subscribedQualityUpdate: LivekitRtc.SubscribedQualityUpdate) {
         if (!dynacast) {
             return
         }
@@ -515,6 +512,17 @@ internal constructor(
         }
     }
 
+    internal fun handleLocalTrackUnpublished(unpublishedResponse: LivekitRtc.TrackUnpublishedResponse) {
+        val pub = tracks[unpublishedResponse.trackSid]
+        val track = pub?.track
+        if (track == null) {
+            LKLog.w { "Received unpublished track response for unknown or non-published track: ${unpublishedResponse.trackSid}" }
+            return
+        }
+
+        unpublishTrack(track)
+    }
+
     fun prepareForFullReconnect() {
         val pubs = localTrackPublications // creates a copy, so is safe from the following removal.
         tracks = tracks.toMutableMap().apply { clear() }

@@ -9,6 +9,7 @@ import io.livekit.android.room.track.RemoteTrackPublication
 import io.livekit.android.room.track.Track
 import io.livekit.android.room.track.TrackPublication
 import io.livekit.android.util.FlowObservable
+import io.livekit.android.util.LKLog
 import io.livekit.android.util.flow
 import io.livekit.android.util.flowDelegate
 import kotlinx.coroutines.CoroutineDispatcher
@@ -18,6 +19,7 @@ import kotlinx.coroutines.flow.SharingStarted
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.stateIn
 import livekit.LivekitModels
+import livekit.LivekitRtc
 import javax.inject.Named
 
 open class Participant(
@@ -88,6 +90,25 @@ open class Participant(
         internal set
 
     /**
+     *
+     */
+    @FlowObservable
+    @get:FlowObservable
+    var permissions: ParticipantPermission? by flowDelegate(null) { newPermissions, oldPermissions ->
+        if (newPermissions != oldPermissions) {
+            eventBus.postEvent(
+                ParticipantEvent.ParticipantPermissionsChanged(
+                    this,
+                    newPermissions,
+                    oldPermissions
+                ),
+                scope
+            )
+        }
+    }
+        internal set
+
+    /**
      * Changes can be observed by using [io.livekit.android.util.flow]
      */
     @FlowObservable
@@ -222,6 +243,9 @@ open class Participant(
         participantInfo = info
         metadata = info.metadata
         name = info.name
+        if (info.hasPermission()) {
+            permissions = ParticipantPermission.fromProto(info.permission)
+        }
     }
 
     override fun equals(other: Any?): Boolean {
@@ -239,7 +263,6 @@ open class Participant(
         return sid.hashCode()
     }
 
-
     // Internal methods just for posting events.
     internal fun onTrackMuted(trackPublication: TrackPublication) {
         listener?.onTrackMuted(trackPublication, this)
@@ -365,3 +388,23 @@ enum class ConnectionQuality {
         }
     }
 }
+
+data class ParticipantPermission(
+    val canPublish: Boolean,
+    val canSubscribe: Boolean,
+    val canPublishData: Boolean,
+    val hidden: Boolean,
+    val recorder: Boolean,
+) {
+    companion object {
+        fun fromProto(proto: LivekitModels.ParticipantPermission): ParticipantPermission {
+            return ParticipantPermission(
+                canPublish = proto.canPublish,
+                canSubscribe = proto.canSubscribe,
+                canPublishData = proto.canPublishData,
+                hidden = proto.hidden,
+                recorder = proto.recorder,
+            )
+        }
+    }
+}
\ No newline at end of file

@@ -2,6 +2,7 @@ package io.livekit.android
 
 import android.content.Context
 import androidx.test.core.app.ApplicationProvider
+import com.google.protobuf.MessageLite
 import io.livekit.android.mock.MockPeerConnection
 import io.livekit.android.mock.MockWebSocketFactory
 import io.livekit.android.mock.dagger.DaggerTestLiveKitComponent
@@ -16,6 +17,7 @@ import kotlinx.coroutines.launch
 import okhttp3.Protocol
 import okhttp3.Request
 import okhttp3.Response
+import okio.ByteString
 import org.junit.Before
 import org.webrtc.PeerConnection
 
@@ -73,4 +75,18 @@ abstract class MockE2ETest : BaseTest() {
             .message("")
             .build()
     }
+
+    /**
+     * Simulates receiving [message] from the server
+     */
+    fun simulateMessageFromServer(message: MessageLite) {
+        simulateMessageFromServer(message.toOkioByteString())
+    }
+
+    /**
+     * Simulates receiving [message] from the server
+     */
+    fun simulateMessageFromServer(message: ByteString) {
+        wsFactory.listener.onMessage(wsFactory.ws, message)
+    }
 }
\ No newline at end of file

@@ -20,6 +20,13 @@ object TestData {
         sid = "local_participant_sid"
         identity = "local_participant_identity"
         state = LivekitModels.ParticipantInfo.State.ACTIVE
+        permission = LivekitModels.ParticipantPermission.newBuilder()
+            .setCanPublish(true)
+            .setCanSubscribe(true)
+            .setCanPublishData(true)
+            .setHidden(true)
+            .setRecorder(false)
+            .build()
         build()
     }
 

@@ -260,6 +260,34 @@ class SignalClientTest : BaseTest() {
             build()
         }
 
+        val LOCAL_TRACK_UNPUBLISHED = with(LivekitRtc.SignalResponse.newBuilder()) {
+            trackUnpublished = with(LivekitRtc.TrackUnpublishedResponse.newBuilder()) {
+                trackSid = TestData.LOCAL_AUDIO_TRACK.sid
+                build()
+            }
+            build()
+        }
+
+        val PERMISSION_CHANGE = with(LivekitRtc.SignalResponse.newBuilder()) {
+            update = with(LivekitRtc.ParticipantUpdate.newBuilder()) {
+                addParticipants(
+                    TestData.LOCAL_PARTICIPANT.toBuilder()
+                        .setPermission(
+                            LivekitModels.ParticipantPermission.newBuilder()
+                                .setCanPublish(false)
+                                .setCanSubscribe(false)
+                                .setCanPublishData(false)
+                                .setHidden(false)
+                                .setRecorder(false)
+                                .build()
+                        )
+                        .build()
+                )
+                build()
+            }
+            build()
+        }
+
         val PARTICIPANT_JOIN = with(LivekitRtc.SignalResponse.newBuilder()) {
             update = with(LivekitRtc.ParticipantUpdate.newBuilder()) {
                 addParticipants(TestData.REMOTE_PARTICIPANT)

+package io.livekit.android.room.participant
+
+import io.livekit.android.MockE2ETest
+import io.livekit.android.events.EventCollector
+import io.livekit.android.events.RoomEvent
+import io.livekit.android.mock.MockAudioStreamTrack
+import io.livekit.android.room.SignalClientTest
+import io.livekit.android.room.track.LocalAudioTrack
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import org.junit.Assert
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+
+
+@ExperimentalCoroutinesApi
+@RunWith(RobolectricTestRunner::class)
+class ParticipantMockE2ETest : MockE2ETest() {
+
+    @Test
+    fun trackUnpublished() = runTest {
+        connect()
+
+        // publish track
+        val publishJob = launch {
+            room.localParticipant.publishAudioTrack(
+                LocalAudioTrack(
+                    "",
+                    MockAudioStreamTrack(id = SignalClientTest.LOCAL_TRACK_PUBLISHED.trackPublished.cid)
+                )
+            )
+        }
+        simulateMessageFromServer(SignalClientTest.LOCAL_TRACK_PUBLISHED)
+        publishJob.join()
+
+        val eventCollector = EventCollector(room.events, coroutineRule.scope)
+        // remote unpublish
+        simulateMessageFromServer(SignalClientTest.LOCAL_TRACK_UNPUBLISHED)
+        val events = eventCollector.stopCollecting()
+
+        Assert.assertEquals(1, events.size)
+        Assert.assertEquals(true, events[0] is RoomEvent.TrackUnpublished)
+        Assert.assertEquals(0, room.localParticipant.tracks.size)
+    }
+
+    @Test
+    fun participantPermissions() = runTest {
+        connect()
+
+        val eventCollector = EventCollector(room.events, coroutineRule.scope)
+        simulateMessageFromServer(SignalClientTest.PERMISSION_CHANGE)
+        val events = eventCollector.stopCollecting()
+
+        Assert.assertEquals(1, events.size)
+        Assert.assertEquals(true, events[0] is RoomEvent.ParticipantPermissionsChanged)
+    }
+}
\ No newline at end of file

- Subproject commit 6f2a49e449143a01b8c63803198b7e9d1112e77b
+ Subproject commit 3c712ad5c941c0d2ddb5631c44239fbe525c0391