Merge pull request #7629 from vector-im/feature/fre/voice_broadcast_handle_event_deletion

Voice Broadcast - Handle event deletion when listening or recording

changelog.d/7629.wip

@@ -0,0 +1 @@
+Voice Broadcast - Handle redaction of the state events on the listener and recorder sides

vector/src/main/java/im/vector/app/core/di/VoiceModule.kt

@@ -27,6 +27,7 @@ import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer
 import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayerImpl
 import im.vector.app.features.voicebroadcast.recording.VoiceBroadcastRecorder
 import im.vector.app.features.voicebroadcast.recording.VoiceBroadcastRecorderQ
+import im.vector.app.features.voicebroadcast.usecase.GetMostRecentVoiceBroadcastStateEventUseCase
 import javax.inject.Singleton
 
 @InstallIn(SingletonComponent::class)
@@ -36,9 +37,17 @@ abstract class VoiceModule {
     companion object {
         @Provides
         @Singleton
-        fun providesVoiceBroadcastRecorder(context: Context): VoiceBroadcastRecorder? {
+        fun providesVoiceBroadcastRecorder(
+                context: Context,
+                sessionHolder: ActiveSessionHolder,
+                getMostRecentVoiceBroadcastStateEventUseCase: GetMostRecentVoiceBroadcastStateEventUseCase,
+        ): VoiceBroadcastRecorder? {
             return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.Q) {
-                VoiceBroadcastRecorderQ(context)
+                VoiceBroadcastRecorderQ(
+                        context = context,
+                        sessionHolder = sessionHolder,
+                        getVoiceBroadcastEventUseCase = getMostRecentVoiceBroadcastStateEventUseCase
+                )
             } else {
                 null
             }

vector/src/main/java/im/vector/app/features/voicebroadcast/listening/VoiceBroadcastPlayerImpl.kt

@@ -30,7 +30,7 @@ import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer.Stat
 import im.vector.app.features.voicebroadcast.listening.usecase.GetLiveVoiceBroadcastChunksUseCase
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
-import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastEventUseCase
+import im.vector.app.features.voicebroadcast.usecase.GetMostRecentVoiceBroadcastStateEventUseCase
 import im.vector.lib.core.utils.timer.CountUpTimer
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.launchIn
@@ -48,7 +48,7 @@ import javax.inject.Singleton
 class VoiceBroadcastPlayerImpl @Inject constructor(
         private val sessionHolder: ActiveSessionHolder,
         private val playbackTracker: AudioMessagePlaybackTracker,
-        private val getVoiceBroadcastEventUseCase: GetVoiceBroadcastEventUseCase,
+        private val getVoiceBroadcastEventUseCase: GetMostRecentVoiceBroadcastStateEventUseCase,
         private val getLiveVoiceBroadcastChunksUseCase: GetLiveVoiceBroadcastChunksUseCase
 ) : VoiceBroadcastPlayer {
 
@@ -66,7 +66,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
     private var nextMediaPlayer: MediaPlayer? = null
     private var isPreparingNextPlayer: Boolean = false
 
-    private var currentVoiceBroadcastEvent: VoiceBroadcastEvent? = null
+    private var mostRecentVoiceBroadcastEvent: VoiceBroadcastEvent? = null
 
     override var currentVoiceBroadcast: VoiceBroadcast? = null
     override var isLiveListening: Boolean = false
@@ -121,7 +121,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         // Clear playlist
         playlist.reset()
 
-        currentVoiceBroadcastEvent = null
+        mostRecentVoiceBroadcastEvent = null
         currentVoiceBroadcast = null
     }
 
@@ -145,19 +145,25 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
 
         playingState = State.BUFFERING
 
-        observeVoiceBroadcastLiveState(voiceBroadcast)
+        observeVoiceBroadcastStateEvent(voiceBroadcast)
         fetchPlaylistAndStartPlayback(voiceBroadcast)
     }
 
-    private fun observeVoiceBroadcastLiveState(voiceBroadcast: VoiceBroadcast) {
+    private fun observeVoiceBroadcastStateEvent(voiceBroadcast: VoiceBroadcast) {
         voiceBroadcastStateObserver = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)
-                .onEach {
+                .onEach { onVoiceBroadcastStateEventUpdated(it.getOrNull()) }
-                    currentVoiceBroadcastEvent = it.getOrNull()
-                    updateLiveListeningMode()
-                }
                 .launchIn(sessionScope)
     }
 
+    private fun onVoiceBroadcastStateEventUpdated(event: VoiceBroadcastEvent?) {
+        if (event == null) {
+            stop()
+        } else {
+            mostRecentVoiceBroadcastEvent = event
+            updateLiveListeningMode()
+        }
+    }
+
     private fun fetchPlaylistAndStartPlayback(voiceBroadcast: VoiceBroadcast) {
         fetchPlaylistTask = getLiveVoiceBroadcastChunksUseCase.execute(voiceBroadcast)
                 .onEach {
@@ -198,7 +204,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
 
         val playlistItem = when {
             position != null -> playlist.findByPosition(position)
-            currentVoiceBroadcastEvent?.isLive.orFalse() -> playlist.lastOrNull()
+            mostRecentVoiceBroadcastEvent?.isLive.orFalse() -> playlist.lastOrNull()
             else -> playlist.firstOrNull()
         }
         val content = playlistItem?.audioEvent?.content ?: run { Timber.w("## VoiceBroadcastPlayer: No content to play"); return }
@@ -340,7 +346,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
     private fun updateLiveListeningMode(seekPosition: Int? = null) {
         isLiveListening = when {
             // the current voice broadcast is not live (ended)
-            currentVoiceBroadcastEvent?.isLive?.not().orFalse() -> false
+            mostRecentVoiceBroadcastEvent?.isLive != true -> false
             // the player is stopped or paused
             playingState == State.IDLE || playingState == State.PAUSED -> false
             seekPosition != null -> {
@@ -406,13 +412,11 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         override fun onCompletion(mp: MediaPlayer) {
             if (nextMediaPlayer != null) return
 
-            val content = currentVoiceBroadcastEvent?.content
+            if (isLiveListening || mostRecentVoiceBroadcastEvent?.content?.lastChunkSequence == playlist.currentSequence) {
-            val isLive = content?.isLive.orFalse()
+                playingState = State.BUFFERING
-            if (!isLive && content?.lastChunkSequence == playlist.currentSequence) {
+            } else {
                 // We'll not receive new chunks anymore so we can stop the live listening
                 stop()
-            } else {
-                playingState = State.BUFFERING
             }
         }
 

vector/src/main/java/im/vector/app/features/voicebroadcast/listening/usecase/GetLiveVoiceBroadcastChunksUseCase.kt

@@ -24,7 +24,7 @@ import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
 import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
 import im.vector.app.features.voicebroadcast.sequence
-import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastEventUseCase
+import im.vector.app.features.voicebroadcast.usecase.GetMostRecentVoiceBroadcastStateEventUseCase
 import im.vector.app.features.voicebroadcast.voiceBroadcastId
 import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
@@ -48,7 +48,7 @@ import javax.inject.Inject
  */
 class GetLiveVoiceBroadcastChunksUseCase @Inject constructor(
         private val activeSessionHolder: ActiveSessionHolder,
-        private val getVoiceBroadcastEventUseCase: GetVoiceBroadcastEventUseCase,
+        private val getVoiceBroadcastEventUseCase: GetMostRecentVoiceBroadcastStateEventUseCase,
 ) {
 
     fun execute(voiceBroadcast: VoiceBroadcast): Flow<List<MessageAudioEvent>> {

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/VoiceBroadcastRecorder.kt

@@ -18,6 +18,7 @@ package im.vector.app.features.voicebroadcast.recording
 
 import androidx.annotation.IntRange
 import im.vector.app.features.voice.VoiceRecorder
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
 import java.io.File
 
 interface VoiceBroadcastRecorder : VoiceRecorder {
@@ -31,7 +32,7 @@ interface VoiceBroadcastRecorder : VoiceRecorder {
     /** Current remaining time of recording, in seconds, if any. */
     val currentRemainingTime: Long?
 
-    fun startRecord(roomId: String, chunkLength: Int, maxLength: Int)
+    fun startRecordVoiceBroadcast(voiceBroadcast: VoiceBroadcast, chunkLength: Int, maxLength: Int)
     fun addListener(listener: Listener)
     fun removeListener(listener: Listener)
 

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/VoiceBroadcastRecorderQ.kt

@@ -20,8 +20,17 @@ import android.content.Context
 import android.media.MediaRecorder
 import android.os.Build
 import androidx.annotation.RequiresApi
+import im.vector.app.core.di.ActiveSessionHolder
+import im.vector.app.features.session.coroutineScope
 import im.vector.app.features.voice.AbstractVoiceRecorderQ
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
+import im.vector.app.features.voicebroadcast.usecase.GetMostRecentVoiceBroadcastStateEventUseCase
 import im.vector.lib.core.utils.timer.CountUpTimer
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.session.content.ContentAttachmentData
 import java.util.concurrent.CopyOnWriteArrayList
@@ -30,10 +39,17 @@ import java.util.concurrent.TimeUnit
 @RequiresApi(Build.VERSION_CODES.Q)
 class VoiceBroadcastRecorderQ(
         context: Context,
+        private val sessionHolder: ActiveSessionHolder,
+        private val getVoiceBroadcastEventUseCase: GetMostRecentVoiceBroadcastStateEventUseCase
 ) : AbstractVoiceRecorderQ(context), VoiceBroadcastRecorder {
 
+    private val session get() = sessionHolder.getActiveSession()
+    private val sessionScope get() = session.coroutineScope
+
+    private var voiceBroadcastStateObserver: Job? = null
+
     private var maxFileSize = 0L // zero or negative for no limit
-    private var currentRoomId: String? = null
+    private var currentVoiceBroadcast: VoiceBroadcast? = null
     private var currentMaxLength: Int = 0
 
     override var currentSequence = 0
@@ -68,17 +84,20 @@ class VoiceBroadcastRecorderQ(
         }
     }
 
-    override fun startRecord(roomId: String, chunkLength: Int, maxLength: Int) {
+    override fun startRecordVoiceBroadcast(voiceBroadcast: VoiceBroadcast, chunkLength: Int, maxLength: Int) {
-        currentRoomId = roomId
+        // Stop recording previous voice broadcast if any
+        if (recordingState != VoiceBroadcastRecorder.State.Idle) stopRecord()
+
+        currentVoiceBroadcast = voiceBroadcast
         maxFileSize = (chunkLength * audioEncodingBitRate / 8).toLong()
         currentMaxLength = maxLength
         currentSequence = 1
-        startRecord(roomId)
+
-        recordingState = VoiceBroadcastRecorder.State.Recording
+        observeVoiceBroadcastStateEvent(voiceBroadcast)
-        recordingTicker.start()
     }
 
     override fun pauseRecord() {
+        if (recordingState != VoiceBroadcastRecorder.State.Recording) return
         tryOrNull { mediaRecorder?.stop() }
         mediaRecorder?.reset()
         recordingState = VoiceBroadcastRecorder.State.Paused
@@ -87,8 +106,9 @@ class VoiceBroadcastRecorderQ(
     }
 
     override fun resumeRecord() {
+        if (recordingState != VoiceBroadcastRecorder.State.Paused) return
         currentSequence++
-        currentRoomId?.let { startRecord(it) }
+        currentVoiceBroadcast?.let { startRecord(it.roomId) }
         recordingState = VoiceBroadcastRecorder.State.Recording
         recordingTicker.resume()
     }
@@ -104,11 +124,15 @@ class VoiceBroadcastRecorderQ(
         // Remove listeners
         listeners.clear()
 
+        // Do not observe anymore voice broadcast changes
+        voiceBroadcastStateObserver?.cancel()
+        voiceBroadcastStateObserver = null
+
         // Reset data
         currentSequence = 0
         currentMaxLength = 0
         currentRemainingTime = null
-        currentRoomId = null
+        currentVoiceBroadcast = null
     }
 
     override fun release() {
@@ -126,6 +150,26 @@ class VoiceBroadcastRecorderQ(
         listeners.remove(listener)
     }
 
+    private fun observeVoiceBroadcastStateEvent(voiceBroadcast: VoiceBroadcast) {
+        voiceBroadcastStateObserver = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)
+                .onEach { onVoiceBroadcastStateEventUpdated(voiceBroadcast, it.getOrNull()) }
+                .launchIn(sessionScope)
+    }
+
+    private fun onVoiceBroadcastStateEventUpdated(voiceBroadcast: VoiceBroadcast, event: VoiceBroadcastEvent?) {
+        when (event?.content?.voiceBroadcastState) {
+            VoiceBroadcastState.STARTED -> {
+                startRecord(voiceBroadcast.roomId)
+                recordingState = VoiceBroadcastRecorder.State.Recording
+                recordingTicker.start()
+            }
+            VoiceBroadcastState.PAUSED -> pauseRecord()
+            VoiceBroadcastState.RESUMED -> resumeRecord()
+            VoiceBroadcastState.STOPPED,
+            null -> stopRecord()
+        }
+    }
+
     private fun onMaxFileSizeApproaching(roomId: String) {
         setNextOutputFile(roomId)
     }

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/usecase/PauseVoiceBroadcastUseCase.kt

@@ -53,17 +53,20 @@ class PauseVoiceBroadcastUseCase @Inject constructor(
 
     private suspend fun pauseVoiceBroadcast(room: Room, reference: RelationDefaultContent?) {
         Timber.d("## PauseVoiceBroadcastUseCase: Send new voice broadcast info state event")
+
+        // save the last sequence number and immediately pause the recording
+        val lastSequence = voiceBroadcastRecorder?.currentSequence
+        pauseRecording()
+
         room.stateService().sendStateEvent(
                 eventType = VoiceBroadcastConstants.STATE_ROOM_VOICE_BROADCAST_INFO,
                 stateKey = session.myUserId,
                 body = MessageVoiceBroadcastInfoContent(
                         relatesTo = reference,
                         voiceBroadcastStateStr = VoiceBroadcastState.PAUSED.value,
-                        lastChunkSequence = voiceBroadcastRecorder?.currentSequence,
+                        lastChunkSequence = lastSequence,
                 ).toContent(),
         )
-
-        pauseRecording()
     }
 
     private fun pauseRecording() {

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/usecase/ResumeVoiceBroadcastUseCase.kt

@@ -20,7 +20,6 @@ import im.vector.app.features.voicebroadcast.VoiceBroadcastConstants
 import im.vector.app.features.voicebroadcast.model.MessageVoiceBroadcastInfoContent
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
 import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
-import im.vector.app.features.voicebroadcast.recording.VoiceBroadcastRecorder
 import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.Session
 import org.matrix.android.sdk.api.session.events.model.toContent
@@ -32,7 +31,6 @@ import javax.inject.Inject
 
 class ResumeVoiceBroadcastUseCase @Inject constructor(
         private val session: Session,
-        private val voiceBroadcastRecorder: VoiceBroadcastRecorder?,
 ) {
 
     suspend fun execute(roomId: String): Result<Unit> = runCatching {
@@ -66,11 +64,5 @@ class ResumeVoiceBroadcastUseCase @Inject constructor(
                         voiceBroadcastStateStr = VoiceBroadcastState.RESUMED.value,
                 ).toContent(),
         )
-
-        resumeRecording()
-    }
-
-    private fun resumeRecording() {
-        voiceBroadcastRecorder?.resumeRecord()
     }
 }

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/usecase/StartVoiceBroadcastUseCase.kt

@@ -24,11 +24,13 @@ import im.vector.app.features.session.coroutineScope
 import im.vector.app.features.voicebroadcast.VoiceBroadcastConstants
 import im.vector.app.features.voicebroadcast.VoiceBroadcastFailure
 import im.vector.app.features.voicebroadcast.model.MessageVoiceBroadcastInfoContent
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastChunk
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
 import im.vector.app.features.voicebroadcast.recording.VoiceBroadcastRecorder
 import im.vector.app.features.voicebroadcast.usecase.GetOngoingVoiceBroadcastsUseCase
 import im.vector.lib.multipicker.utils.toMultiPickerAudioType
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.launch
 import org.jetbrains.annotations.VisibleForTesting
 import org.matrix.android.sdk.api.query.QueryStringValue
@@ -43,6 +45,8 @@ import org.matrix.android.sdk.api.session.room.getStateEvent
 import org.matrix.android.sdk.api.session.room.model.PowerLevelsContent
 import org.matrix.android.sdk.api.session.room.model.relation.RelationDefaultContent
 import org.matrix.android.sdk.api.session.room.powerlevels.PowerLevelsHelper
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 import timber.log.Timber
 import java.io.File
 import javax.inject.Inject
@@ -63,6 +67,7 @@ class StartVoiceBroadcastUseCase @Inject constructor(
 
         assertCanStartVoiceBroadcast(room)
         startVoiceBroadcast(room)
+        return Result.success(Unit)
     }
 
     private suspend fun startVoiceBroadcast(room: Room) {
@@ -79,13 +84,18 @@ class StartVoiceBroadcastUseCase @Inject constructor(
                 ).toContent()
         )
 
-        startRecording(room, eventId, chunkLength, maxLength)
+        val voiceBroadcast = VoiceBroadcast(roomId = room.roomId, voiceBroadcastId = eventId)
+
+        // TODO Update unit test to cover the following line
+        room.flow().liveTimelineEvent(eventId).unwrap().first() // wait for the event come back from the sync
+
+        startRecording(room, voiceBroadcast, chunkLength, maxLength)
     }
 
-    private fun startRecording(room: Room, eventId: String, chunkLength: Int, maxLength: Int) {
+    private fun startRecording(room: Room, voiceBroadcast: VoiceBroadcast, chunkLength: Int, maxLength: Int) {
         voiceBroadcastRecorder?.addListener(object : VoiceBroadcastRecorder.Listener {
             override fun onVoiceMessageCreated(file: File, sequence: Int) {
-                sendVoiceFile(room, file, eventId, sequence)
+                sendVoiceFile(room, file, voiceBroadcast, sequence)
             }
 
             override fun onRemainingTimeUpdated(remainingTime: Long?) {
@@ -94,10 +104,10 @@ class StartVoiceBroadcastUseCase @Inject constructor(
                 }
             }
         })
-        voiceBroadcastRecorder?.startRecord(room.roomId, chunkLength, maxLength)
+        voiceBroadcastRecorder?.startRecordVoiceBroadcast(voiceBroadcast, chunkLength, maxLength)
     }
 
-    private fun sendVoiceFile(room: Room, voiceMessageFile: File, referenceEventId: String, sequence: Int) {
+    private fun sendVoiceFile(room: Room, voiceMessageFile: File, voiceBroadcast: VoiceBroadcast, sequence: Int) {
         val outputFileUri = FileProvider.getUriForFile(
                 context,
                 buildMeta.applicationId + ".fileProvider",
@@ -109,7 +119,7 @@ class StartVoiceBroadcastUseCase @Inject constructor(
                 attachment = audioType.toContentAttachmentData(isVoiceMessage = true),
                 compressBeforeSending = false,
                 roomIds = emptySet(),
-                relatesTo = RelationDefaultContent(RelationType.REFERENCE, referenceEventId),
+                relatesTo = RelationDefaultContent(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId),
                 additionalContent = mapOf(
                         VoiceBroadcastConstants.VOICE_BROADCAST_CHUNK_KEY to VoiceBroadcastChunk(sequence = sequence).toContent()
                 )

vector/src/main/java/im/vector/app/features/voicebroadcast/recording/usecase/StopVoiceBroadcastUseCase.kt

@@ -54,17 +54,20 @@ class StopVoiceBroadcastUseCase @Inject constructor(
 
     private suspend fun stopVoiceBroadcast(room: Room, reference: RelationDefaultContent?) {
         Timber.d("## StopVoiceBroadcastUseCase: Send new voice broadcast info state event")
+
+        // save the last sequence number and immediately stop the recording
+        val lastSequence = voiceBroadcastRecorder?.currentSequence
+        stopRecording()
+
         room.stateService().sendStateEvent(
                 eventType = VoiceBroadcastConstants.STATE_ROOM_VOICE_BROADCAST_INFO,
                 stateKey = session.myUserId,
                 body = MessageVoiceBroadcastInfoContent(
                         relatesTo = reference,
                         voiceBroadcastStateStr = VoiceBroadcastState.STOPPED.value,
-                        lastChunkSequence = voiceBroadcastRecorder?.currentSequence,
+                        lastChunkSequence = lastSequence,
                 ).toContent(),
         )
-
-        stopRecording()
     }
 
     private fun stopRecording() {

vector/src/main/java/im/vector/app/features/voicebroadcast/usecase/GetMostRecentVoiceBroadcastStateEventUseCase.kt

@@ -0,0 +1,160 @@
+/*
+ * Copyright (c) 2022 New Vector Ltd
+ *
+ * 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 im.vector.app.features.voicebroadcast.usecase
+
+import im.vector.app.features.voicebroadcast.VoiceBroadcastConstants
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
+import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
+import im.vector.app.features.voicebroadcast.voiceBroadcastId
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.distinctUntilChangedBy
+import kotlinx.coroutines.flow.drop
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flatMapLatest
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.onEach
+import kotlinx.coroutines.flow.onStart
+import kotlinx.coroutines.flow.transformWhile
+import org.matrix.android.sdk.api.query.QueryStringValue
+import org.matrix.android.sdk.api.session.Session
+import org.matrix.android.sdk.api.session.events.model.RelationType
+import org.matrix.android.sdk.api.session.getRoom
+import org.matrix.android.sdk.api.session.room.Room
+import org.matrix.android.sdk.api.util.Optional
+import org.matrix.android.sdk.api.util.toOptional
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.mapOptional
+import timber.log.Timber
+import javax.inject.Inject
+
+class GetMostRecentVoiceBroadcastStateEventUseCase @Inject constructor(
+        private val session: Session,
+) {
+
+    fun execute(voiceBroadcast: VoiceBroadcast): Flow<Optional<VoiceBroadcastEvent>> {
+        val room = session.getRoom(voiceBroadcast.roomId) ?: error("Unknown roomId: ${voiceBroadcast.roomId}")
+        return getMostRecentVoiceBroadcastEventFlow(room, voiceBroadcast)
+                .onEach { event ->
+                    Timber.d(
+                            "## VoiceBroadcast | " +
+                                    "voiceBroadcastId=${event.getOrNull()?.voiceBroadcastId}, " +
+                                    "state=${event.getOrNull()?.content?.voiceBroadcastState}"
+                    )
+                }
+    }
+
+    /**
+     * Get a flow of the most recent event for the given voice broadcast.
+     */
+    private fun getMostRecentVoiceBroadcastEventFlow(room: Room, voiceBroadcast: VoiceBroadcast): Flow<Optional<VoiceBroadcastEvent>> {
+        val startedEventFlow = room.flow().liveTimelineEvent(voiceBroadcast.voiceBroadcastId)
+        // observe started event changes
+        return startedEventFlow
+                .mapOptional { it.root.asVoiceBroadcastEvent() }
+                .flatMapLatest { startedEvent ->
+                    if (startedEvent.hasValue().not() || startedEvent.get().root.isRedacted()) {
+                        // if started event is null or redacted, send null
+                        flowOf(Optional.empty())
+                    } else {
+                        // otherwise, observe most recent event changes
+                        getMostRecentRelatedEventFlow(room, voiceBroadcast)
+                                .transformWhile { mostRecentEvent ->
+                                    val hasValue = mostRecentEvent.hasValue()
+                                    if (hasValue) {
+                                        // keep the most recent event
+                                        emit(mostRecentEvent)
+                                    } else {
+                                        // no most recent event, fallback to started event
+                                        emit(startedEvent)
+                                    }
+                                    hasValue
+                                }
+                    }
+                }
+                .distinctUntilChangedBy { it.getOrNull()?.content?.voiceBroadcastState }
+    }
+
+    /**
+     * Get a flow of the most recent related event.
+     */
+    private fun getMostRecentRelatedEventFlow(room: Room, voiceBroadcast: VoiceBroadcast): Flow<Optional<VoiceBroadcastEvent>> {
+        val mostRecentEvent = getMostRecentRelatedEvent(room, voiceBroadcast).toOptional()
+        return if (mostRecentEvent.hasValue()) {
+            val stateKey = mostRecentEvent.get().root.stateKey.orEmpty()
+            // observe incoming voice broadcast state events
+            room.flow()
+                    .liveStateEvent(VoiceBroadcastConstants.STATE_ROOM_VOICE_BROADCAST_INFO, QueryStringValue.Equals(stateKey))
+                    .mapOptional { it.asVoiceBroadcastEvent() }
+                    // drop first event sent by the matrix-sdk, we compute manually this first event
+                    .drop(1)
+                    // start with the computed most recent event
+                    .onStart { emit(mostRecentEvent) }
+                    // handle event if null or related to the given voice broadcast
+                    .filter { it.hasValue().not() || it.get().voiceBroadcastId == voiceBroadcast.voiceBroadcastId }
+                    // observe changes while event is not null
+                    .transformWhile { event ->
+                        emit(event)
+                        event.hasValue()
+                    }
+                    .flatMapLatest { newMostRecentEvent ->
+                        if (newMostRecentEvent.hasValue()) {
+                            // observe most recent event changes
+                            newMostRecentEvent.get().flow()
+                                    .transformWhile { event ->
+                                        // observe changes until event is null or redacted
+                                        emit(event)
+                                        event.hasValue() && event.get().root.isRedacted().not()
+                                    }
+                                    .flatMapLatest { event ->
+                                        val isRedactedOrNull = !event.hasValue() || event.get().root.isRedacted()
+                                        if (isRedactedOrNull) {
+                                            // event is null or redacted, switch to the latest not redacted event
+                                            getMostRecentRelatedEventFlow(room, voiceBroadcast)
+                                        } else {
+                                            // event is not redacted, send the event
+                                            flowOf(event)
+                                        }
+                                    }
+                        } else {
+                            // there is no more most recent event, just send it
+                            flowOf(newMostRecentEvent)
+                        }
+                    }
+        } else {
+            // there is no more most recent event, just send it
+            flowOf(mostRecentEvent)
+        }
+    }
+
+    /**
+     * Get the most recent event related to the given voice broadcast.
+     */
+    private fun getMostRecentRelatedEvent(room: Room, voiceBroadcast: VoiceBroadcast): VoiceBroadcastEvent? {
+        return room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId)
+                .mapNotNull { timelineEvent -> timelineEvent.root.asVoiceBroadcastEvent()?.takeUnless { it.root.isRedacted() } }
+                .maxByOrNull { it.root.originServerTs ?: 0 }
+    }
+
+    /**
+     * Get a flow of the given voice broadcast event changes.
+     */
+    private fun VoiceBroadcastEvent.flow(): Flow<Optional<VoiceBroadcastEvent>> {
+        val room = this.root.roomId?.let { session.getRoom(it) } ?: return flowOf(Optional.empty())
+        return room.flow().liveTimelineEvent(root.eventId!!).mapOptional { it.root.asVoiceBroadcastEvent() }
+    }
+}

vector/src/main/java/im/vector/app/features/voicebroadcast/usecase/GetVoiceBroadcastEventUseCase.kt

@@ -1,68 +0,0 @@
-/*
- * Copyright (c) 2022 New Vector Ltd
- *
- * 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 im.vector.app.features.voicebroadcast.usecase
-
-import im.vector.app.features.voicebroadcast.VoiceBroadcastConstants
-import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
-import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
-import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
-import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
-import im.vector.app.features.voicebroadcast.voiceBroadcastId
-import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.distinctUntilChanged
-import kotlinx.coroutines.flow.filter
-import kotlinx.coroutines.flow.flowOf
-import kotlinx.coroutines.flow.onStart
-import org.matrix.android.sdk.api.query.QueryStringValue
-import org.matrix.android.sdk.api.session.Session
-import org.matrix.android.sdk.api.session.events.model.RelationType
-import org.matrix.android.sdk.api.session.getRoom
-import org.matrix.android.sdk.api.util.Optional
-import org.matrix.android.sdk.api.util.toOptional
-import org.matrix.android.sdk.flow.flow
-import org.matrix.android.sdk.flow.mapOptional
-import timber.log.Timber
-import javax.inject.Inject
-
-class GetVoiceBroadcastEventUseCase @Inject constructor(
-        private val session: Session,
-) {
-
-    fun execute(voiceBroadcast: VoiceBroadcast): Flow<Optional<VoiceBroadcastEvent>> {
-        val room = session.getRoom(voiceBroadcast.roomId) ?: error("Unknown roomId: ${voiceBroadcast.roomId}")
-
-        Timber.d("## GetVoiceBroadcastUseCase: get voice broadcast $voiceBroadcast")
-
-        val initialEvent = room.timelineService().getTimelineEvent(voiceBroadcast.voiceBroadcastId)?.root?.asVoiceBroadcastEvent()
-        val latestEvent = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId)
-                .mapNotNull { it.root.asVoiceBroadcastEvent() }
-                .maxByOrNull { it.root.originServerTs ?: 0 }
-                ?: initialEvent
-
-        return when (latestEvent?.content?.voiceBroadcastState) {
-            null, VoiceBroadcastState.STOPPED -> flowOf(latestEvent.toOptional())
-            else -> {
-                room.flow()
-                        .liveStateEvent(VoiceBroadcastConstants.STATE_ROOM_VOICE_BROADCAST_INFO, QueryStringValue.Equals(latestEvent.root.stateKey.orEmpty()))
-                        .onStart { emit(latestEvent.root.toOptional()) }
-                        .distinctUntilChanged()
-                        .filter { !it.hasValue() || it.getOrNull()?.asVoiceBroadcastEvent()?.voiceBroadcastId == voiceBroadcast.voiceBroadcastId }
-                        .mapOptional { it.asVoiceBroadcastEvent() }
-            }
-        }
-    }
-}

vector/src/test/java/im/vector/app/features/voicebroadcast/usecase/ResumeVoiceBroadcastUseCaseTest.kt

@@ -19,7 +19,6 @@ package im.vector.app.features.voicebroadcast.usecase
 import im.vector.app.features.voicebroadcast.VoiceBroadcastConstants
 import im.vector.app.features.voicebroadcast.model.MessageVoiceBroadcastInfoContent
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
-import im.vector.app.features.voicebroadcast.recording.VoiceBroadcastRecorder
 import im.vector.app.features.voicebroadcast.recording.usecase.ResumeVoiceBroadcastUseCase
 import im.vector.app.test.fakes.FakeRoom
 import im.vector.app.test.fakes.FakeRoomService
@@ -27,7 +26,6 @@ import im.vector.app.test.fakes.FakeSession
 import io.mockk.clearAllMocks
 import io.mockk.coEvery
 import io.mockk.coVerify
-import io.mockk.mockk
 import io.mockk.slot
 import kotlinx.coroutines.test.runTest
 import org.amshove.kluent.shouldBe
@@ -47,8 +45,7 @@ class ResumeVoiceBroadcastUseCaseTest {
 
     private val fakeRoom = FakeRoom()
     private val fakeSession = FakeSession(fakeRoomService = FakeRoomService(fakeRoom))
-    private val fakeVoiceBroadcastRecorder = mockk<VoiceBroadcastRecorder>(relaxed = true)
+    private val resumeVoiceBroadcastUseCase = ResumeVoiceBroadcastUseCase(fakeSession)
-    private val resumeVoiceBroadcastUseCase = ResumeVoiceBroadcastUseCase(fakeSession, fakeVoiceBroadcastRecorder)
 
     @Test
     fun `given a room id with a potential existing voice broadcast state when calling execute then the voice broadcast is resumed or not`() = runTest {