Improve player seek

vector/src/main/java/im/vector/app/features/home/room/detail/timeline/item/MessageVoiceBroadcastListeningItem.kt

@@ -71,18 +71,14 @@ abstract class MessageVoiceBroadcastListeningItem : AbsMessageVoiceBroadcastItem
                     playPauseButton.setImageResource(R.drawable.ic_play_pause_pause)
                     playPauseButton.contentDescription = view.resources.getString(R.string.a11y_pause_voice_broadcast)
                     playPauseButton.onClick { callback?.onTimelineItemAction(VoiceBroadcastAction.Listening.Pause) }
-                    seekBar.isEnabled = true
                 }
                 VoiceBroadcastPlayer.State.IDLE,
                 VoiceBroadcastPlayer.State.PAUSED -> {
                     playPauseButton.setImageResource(R.drawable.ic_play_pause_play)
                     playPauseButton.contentDescription = view.resources.getString(R.string.a11y_play_voice_broadcast)
                     playPauseButton.onClick { callback?.onTimelineItemAction(VoiceBroadcastAction.Listening.PlayOrResume(voiceBroadcast)) }
-                    seekBar.isEnabled = false
-                }
-                VoiceBroadcastPlayer.State.BUFFERING -> {
-                    seekBar.isEnabled = true
                 }
+                VoiceBroadcastPlayer.State.BUFFERING -> Unit
             }
         }
     }
@@ -112,6 +108,7 @@ abstract class MessageVoiceBroadcastListeningItem : AbsMessageVoiceBroadcastItem
                     }
                     is AudioMessagePlaybackTracker.Listener.State.Playing -> {
                         if (!isUserSeeking) {
+//                            Timber.d("Voice Broadcast | AudioMessagePlaybackTracker.Listener.onUpdate - duration: $duration, playbackTime: ${state.playbackTime}")
                             holder.seekBar.progress = state.playbackTime
                         }
                     }

vector/src/main/java/im/vector/app/features/voicebroadcast/VoiceBroadcastExtensions.kt

@@ -17,6 +17,8 @@
 package im.vector.app.features.voicebroadcast
 
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcastChunk
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
 import org.matrix.android.sdk.api.session.events.model.Content
 import org.matrix.android.sdk.api.session.events.model.getRelationContent
 import org.matrix.android.sdk.api.session.events.model.toModel
@@ -34,3 +36,6 @@ fun MessageAudioEvent.getVoiceBroadcastChunk(): VoiceBroadcastChunk? {
 val MessageAudioEvent.sequence: Int? get() = getVoiceBroadcastChunk()?.sequence
 
 val MessageAudioEvent.duration get() = content.audioInfo?.duration ?: content.audioWaveformInfo?.duration ?: 0
+
+val VoiceBroadcastEvent.isLive
+    get() = content?.voiceBroadcastState != null && content?.voiceBroadcastState != VoiceBroadcastState.STOPPED

vector/src/main/java/im/vector/app/features/voicebroadcast/VoiceBroadcastHelper.kt

@@ -49,8 +49,6 @@ class VoiceBroadcastHelper @Inject constructor(
     fun stopPlayback() = voiceBroadcastPlayer.stop()
 
     fun seekTo(voiceBroadcast: VoiceBroadcast, positionMillis: Int) {
-        if (voiceBroadcastPlayer.currentVoiceBroadcast == voiceBroadcast) {
+        voiceBroadcastPlayer.seekTo(voiceBroadcast, positionMillis)
-            voiceBroadcastPlayer.seekTo(positionMillis)
-        }
     }
 }

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

@@ -46,9 +46,9 @@ interface VoiceBroadcastPlayer {
     fun stop()
 
     /**
-     * Seek to the given playback position, is milliseconds.
+     * Seek the given voice broadcast playback to the given position, is milliseconds.
      */
-    fun seekTo(positionMillis: Int)
+    fun seekTo(voiceBroadcast: VoiceBroadcast, positionMillis: Int)
 
     /**
      * Add a [Listener] to the given voice broadcast.

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

@@ -18,16 +18,18 @@ package im.vector.app.features.voicebroadcast.listening
 
 import android.media.AudioAttributes
 import android.media.MediaPlayer
+import android.media.MediaPlayer.OnPreparedListener
 import androidx.annotation.MainThread
 import im.vector.app.core.di.ActiveSessionHolder
 import im.vector.app.features.home.room.detail.timeline.helper.AudioMessagePlaybackTracker
 import im.vector.app.features.voice.VoiceFailure
 import im.vector.app.features.voicebroadcast.duration
+import im.vector.app.features.voicebroadcast.isLive
 import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer.Listener
 import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer.State
 import im.vector.app.features.voicebroadcast.listening.usecase.GetLiveVoiceBroadcastChunksUseCase
 import im.vector.app.features.voicebroadcast.model.VoiceBroadcast
-import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
 import im.vector.app.features.voicebroadcast.sequence
 import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastEventUseCase
 import im.vector.lib.core.utils.timer.CountUpTimer
@@ -38,7 +40,6 @@ import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
-import kotlinx.coroutines.withContext
 import org.matrix.android.sdk.api.extensions.orFalse
 import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.session.room.model.message.MessageAudioContent
@@ -47,6 +48,7 @@ import timber.log.Timber
 import java.util.concurrent.CopyOnWriteArrayList
 import javax.inject.Inject
 import javax.inject.Singleton
+import kotlin.math.absoluteValue
 
 @Singleton
 class VoiceBroadcastPlayerImpl @Inject constructor(
@@ -60,19 +62,20 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         get() = sessionHolder.getActiveSession()
 
     private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-    private var voiceBroadcastStateJob: Job? = null
+    private var fetchPlaylistTask: Job? = null
+    private var voiceBroadcastStateTask: Job? = null
 
     private val mediaPlayerListener = MediaPlayerListener()
     private val playbackTicker = PlaybackTicker()
 
     private var currentMediaPlayer: MediaPlayer? = null
     private var nextMediaPlayer: MediaPlayer? = null
-    private var currentSequence: Int? = null
 
-    private var fetchPlaylistJob: Job? = null
     private var playlist = emptyList<PlaylistItem>()
-
+    private var currentSequence: Int? = null
-    private var isLive: Boolean = false
+    private var currentVoiceBroadcastEvent: VoiceBroadcastEvent? = null
+    private val isLive get() = currentVoiceBroadcastEvent?.isLive.orFalse()
+    private val lastSequence get() = currentVoiceBroadcastEvent?.content?.lastChunkSequence
 
     override var currentVoiceBroadcast: VoiceBroadcast? = null
 
@@ -81,33 +84,10 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         set(value) {
             Timber.w("## VoiceBroadcastPlayer state: $field -> $value")
             field = value
-            // Notify state change to all the listeners attached to the current voice broadcast id
+            onPlayingStateChanged(value)
-            currentVoiceBroadcast?.voiceBroadcastId?.let { voiceBroadcastId ->
-                when (value) {
-                    State.PLAYING -> {
-                        playbackTracker.startPlayback(voiceBroadcastId)
-                        playbackTicker.startPlaybackTicker(voiceBroadcastId)
-                    }
-                    State.PAUSED -> {
-                        playbackTracker.pausePlayback(voiceBroadcastId)
-                        playbackTicker.stopPlaybackTicker()
-                    }
-                    State.BUFFERING -> {
-                        playbackTracker.pausePlayback(voiceBroadcastId)
-                        playbackTicker.stopPlaybackTicker()
-                    }
-                    State.IDLE -> {
-                        playbackTracker.stopPlayback(voiceBroadcastId)
-                        playbackTicker.stopPlaybackTicker()
-                    }
-                }
-                listeners[voiceBroadcastId]?.forEach { listener -> listener.onStateChanged(value) }
-            }
         }
 
-    /**
+    /** Map voiceBroadcastId to listeners.*/
-     * Map voiceBroadcastId to listeners.
-     */
     private val listeners: MutableMap<String, CopyOnWriteArrayList<Listener>> = mutableMapOf()
 
     override fun playOrResume(voiceBroadcast: VoiceBroadcast) {
@@ -120,38 +100,28 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
     }
 
     override fun pause() {
-        playingState = State.PAUSED
         currentMediaPlayer?.pause()
+        playingState = State.PAUSED
     }
 
     override fun stop() {
         // Update state
         playingState = State.IDLE
 
-        // Stop playback
+        // Stop and release media players
-        currentMediaPlayer?.stop()
+        stopPlayer()
-        isLive = false
 
-        // Release current player
+        // Do not observe anymore voice broadcast changes
-        release(currentMediaPlayer)
+        fetchPlaylistTask?.cancel()
-        currentMediaPlayer = null
+        fetchPlaylistTask = null
-
+        voiceBroadcastStateTask?.cancel()
-        // Release next player
+        voiceBroadcastStateTask = null
-        release(nextMediaPlayer)
-        nextMediaPlayer = null
-
-        // Do not observe anymore voice broadcast state changes
-        voiceBroadcastStateJob?.cancel()
-        voiceBroadcastStateJob = null
-
-        // Do not fetch the playlist anymore
-        fetchPlaylistJob?.cancel()
-        fetchPlaylistJob = null
 
         // Clear playlist
         playlist = emptyList()
         currentSequence = null
 
+        currentVoiceBroadcastEvent = null
         currentVoiceBroadcast = null
     }
 
@@ -174,13 +144,18 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
 
         playingState = State.BUFFERING
 
-        val voiceBroadcastState = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)?.content?.voiceBroadcastState
+        observeVoiceBroadcastLiveState(voiceBroadcast)
-        isLive = voiceBroadcastState != null && voiceBroadcastState != VoiceBroadcastState.STOPPED
         fetchPlaylistAndStartPlayback(voiceBroadcast)
     }
 
+    private fun observeVoiceBroadcastLiveState(voiceBroadcast: VoiceBroadcast) {
+        voiceBroadcastStateTask = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)
+                .onEach { currentVoiceBroadcastEvent = it.getOrNull() }
+                .launchIn(coroutineScope)
+    }
+
     private fun fetchPlaylistAndStartPlayback(voiceBroadcast: VoiceBroadcast) {
-        fetchPlaylistJob = getLiveVoiceBroadcastChunksUseCase.execute(voiceBroadcast)
+        fetchPlaylistTask = getLiveVoiceBroadcastChunksUseCase.execute(voiceBroadcast)
                 .onEach(this::updatePlaylist)
                 .launchIn(coroutineScope)
     }
@@ -204,40 +179,51 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         when (playingState) {
             State.PLAYING -> {
                 if (nextMediaPlayer == null) {
-                    coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
+                    prepareNextMediaPlayer()
                 }
             }
             State.PAUSED -> {
                 if (nextMediaPlayer == null) {
-                    coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
+                    prepareNextMediaPlayer()
                 }
             }
             State.BUFFERING -> {
                 val newMediaContent = getNextAudioContent()
-                if (newMediaContent != null) startPlayback()
+                if (newMediaContent != null) {
+                    val savedPosition = currentVoiceBroadcast?.let { playbackTracker.getPlaybackTime(it.voiceBroadcastId) }
+                    startPlayback(savedPosition)
+                }
+            }
+            State.IDLE -> {
+                val savedPosition = currentVoiceBroadcast?.let { playbackTracker.getPlaybackTime(it.voiceBroadcastId) }
+                startPlayback(savedPosition)
             }
-            State.IDLE -> startPlayback()
         }
     }
 
-    private fun startPlayback(sequence: Int? = null, position: Int = 0) {
+    private fun startPlayback(position: Int? = null) {
+        stopPlayer()
+
         val playlistItem = when {
-            sequence != null -> playlist.find { it.audioEvent.sequence == sequence }
+            position != null -> playlist.lastOrNull { it.startTime <= position }
             isLive -> playlist.lastOrNull()
             else -> playlist.firstOrNull()
         }
         val content = playlistItem?.audioEvent?.content ?: run { Timber.w("## VoiceBroadcastPlayer: No content to play"); return }
-        val computedSequence = playlistItem.audioEvent.sequence
+        val sequence = playlistItem.audioEvent.sequence
+        val sequencePosition = position?.let { it - playlistItem.startTime } ?: 0
         coroutineScope.launch {
             try {
-                currentMediaPlayer = prepareMediaPlayer(content)
+                prepareMediaPlayer(content) { mp ->
-                currentMediaPlayer?.start()
+                    currentMediaPlayer = mp
-                if (position > 0) {
+                    currentSequence = sequence
-                    currentMediaPlayer?.seekTo(position)
+                    mp.start()
+                    if (sequencePosition > 0) {
+                        mp.seekTo(sequencePosition)
+                    }
+                    playingState = State.PLAYING
+                    prepareNextMediaPlayer()
                 }
-                currentSequence = computedSequence
-                withContext(Dispatchers.Main) { playingState = State.PLAYING }
-                nextMediaPlayer = prepareNextMediaPlayer()
             } catch (failure: Throwable) {
                 Timber.e(failure, "Unable to start playback")
                 throw VoiceFailure.UnableToPlay(failure)
@@ -250,20 +236,12 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         playingState = State.PLAYING
     }
 
-    override fun seekTo(positionMillis: Int) {
+    override fun seekTo(voiceBroadcast: VoiceBroadcast, positionMillis: Int) {
-        val duration = getVoiceBroadcastDuration()
+        if (voiceBroadcast != currentVoiceBroadcast) {
-        val playlistItem = playlist.lastOrNull { it.startTime <= positionMillis } ?: return
+            playbackTracker.updatePausedAtPlaybackTime(voiceBroadcast.voiceBroadcastId, positionMillis, 0f)
-        val audioEvent = playlistItem.audioEvent
+        } else {
-        val eventPosition = positionMillis - playlistItem.startTime
+            startPlayback(positionMillis)
-
+        }
-        Timber.d("## Voice Broadcast | seekTo - duration=$duration, position=$positionMillis, sequence=${audioEvent.sequence}, sequencePosition=$eventPosition")
-
-        tryOrNull { currentMediaPlayer?.stop() }
-        release(currentMediaPlayer)
-        tryOrNull { nextMediaPlayer?.stop() }
-        release(nextMediaPlayer)
-
-        startPlayback(audioEvent.sequence, eventPosition)
     }
 
     private fun getNextAudioContent(): MessageAudioContent? {
@@ -273,12 +251,24 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
         return playlist.find { it.audioEvent.sequence == nextSequence }?.audioEvent?.content
     }
 
-    private suspend fun prepareNextMediaPlayer(): MediaPlayer? {
+    private fun prepareNextMediaPlayer() {
-        val nextContent = getNextAudioContent() ?: return null
+        nextMediaPlayer = null
-        return prepareMediaPlayer(nextContent)
+        val nextContent = getNextAudioContent()
+        if (nextContent != null) {
+            coroutineScope.launch {
+                prepareMediaPlayer(nextContent) { mp ->
+                    if (nextMediaPlayer == null) {
+                        nextMediaPlayer = mp
+                        currentMediaPlayer?.setNextMediaPlayer(mp)
+                    } else {
+                        mp.release()
+                    }
+                }
+            }
+        }
     }
 
-    private suspend fun prepareMediaPlayer(messageAudioContent: MessageAudioContent): MediaPlayer {
+    private suspend fun prepareMediaPlayer(messageAudioContent: MessageAudioContent, onPreparedListener: OnPreparedListener): MediaPlayer {
         // Download can fail
         val audioFile = try {
             session.fileService().downloadFile(messageAudioContent)
@@ -299,57 +289,55 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
                 setDataSource(fis.fd)
                 setOnInfoListener(mediaPlayerListener)
                 setOnErrorListener(mediaPlayerListener)
+                setOnPreparedListener(onPreparedListener)
                 setOnCompletionListener(mediaPlayerListener)
                 prepare()
             }
         }
     }
 
-    private fun release(mp: MediaPlayer?) {
+    private fun stopPlayer() {
-        mp?.apply {
+        tryOrNull { currentMediaPlayer?.stop() }
-            release()
+        currentMediaPlayer?.release()
-            setOnInfoListener(null)
+        currentMediaPlayer = null
-            setOnCompletionListener(null)
+
-            setOnErrorListener(null)
+        nextMediaPlayer?.release()
+        nextMediaPlayer = null
+    }
+
+    private fun onPlayingStateChanged(playingState: State) {
+        // Notify state change to all the listeners attached to the current voice broadcast id
+        currentVoiceBroadcast?.voiceBroadcastId?.let { voiceBroadcastId ->
+            when (playingState) {
+                State.PLAYING -> playbackTicker.startPlaybackTicker(voiceBroadcastId)
+                State.PAUSED,
+                State.BUFFERING,
+                State.IDLE -> playbackTicker.stopPlaybackTicker(voiceBroadcastId)
+            }
+            listeners[voiceBroadcastId]?.forEach { listener -> listener.onStateChanged(playingState) }
         }
     }
 
     private inner class MediaPlayerListener :
             MediaPlayer.OnInfoListener,
-            MediaPlayer.OnPreparedListener,
             MediaPlayer.OnCompletionListener,
             MediaPlayer.OnErrorListener {
 
         override fun onInfo(mp: MediaPlayer, what: Int, extra: Int): Boolean {
             when (what) {
                 MediaPlayer.MEDIA_INFO_STARTED_AS_NEXT -> {
-                    release(currentMediaPlayer)
-                    currentMediaPlayer = mp
                     currentSequence = currentSequence?.plus(1)
-                    coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
+                    currentMediaPlayer = mp
+                    prepareNextMediaPlayer()
                 }
             }
             return false
         }
 
-        override fun onPrepared(mp: MediaPlayer) {
-            when (mp) {
-                currentMediaPlayer -> {
-                    nextMediaPlayer?.let { mp.setNextMediaPlayer(it) }
-                }
-                nextMediaPlayer -> {
-                    tryOrNull { currentMediaPlayer?.setNextMediaPlayer(mp) }
-                }
-            }
-        }
-
         override fun onCompletion(mp: MediaPlayer) {
             if (nextMediaPlayer != null) return
-            val voiceBroadcast = currentVoiceBroadcast ?: return
-            val voiceBroadcastEventContent = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)?.content ?: return
-            isLive = voiceBroadcastEventContent.voiceBroadcastState != null && voiceBroadcastEventContent.voiceBroadcastState != VoiceBroadcastState.STOPPED
 
-            if (!isLive && voiceBroadcastEventContent.lastChunkSequence == currentSequence) {
+            if (!isLive && lastSequence == currentSequence) {
                 // We'll not receive new chunks anymore so we can stop the live listening
                 stop()
             } else {
@@ -388,23 +376,31 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
             if (currentMediaPlayer?.isPlaying.orFalse()) {
                 val itemStartPosition = currentSequence?.let { seq -> playlist.find { it.audioEvent.sequence == seq } }?.startTime
                 val currentVoiceBroadcastPosition = itemStartPosition?.plus(currentMediaPlayer?.currentPosition ?: 0)
+                Timber.d("Voice Broadcast | VoiceBroadcastPlayerImpl - sequence: $currentSequence, itemStartPosition $itemStartPosition, currentMediaPlayer=$currentMediaPlayer, currentMediaPlayer?.currentPosition: ${currentMediaPlayer?.currentPosition}")
                 if (currentVoiceBroadcastPosition != null) {
                     val totalDuration = getVoiceBroadcastDuration()
                     val percentage = currentVoiceBroadcastPosition.toFloat() / totalDuration
                     playbackTracker.updatePlayingAtPlaybackTime(id, currentVoiceBroadcastPosition, percentage)
                 } else {
-                    playbackTracker.stopPlayback(id)
+                    stopPlaybackTicker(id)
-                    stopPlaybackTicker()
                 }
             } else {
-                playbackTracker.stopPlayback(id)
+                stopPlaybackTicker(id)
-                stopPlaybackTicker()
             }
         }
 
-        fun stopPlaybackTicker() {
+        fun stopPlaybackTicker(id: String) {
             playbackTicker?.stop()
             playbackTicker = null
+
+            val totalDuration = getVoiceBroadcastDuration()
+            val playbackTime = playbackTracker.getPlaybackTime(id)
+            val remainingTime = totalDuration - playbackTime
+            if (remainingTime < 1000) {
+                playbackTracker.updatePausedAtPlaybackTime(id, 0, 0f)
+            } else {
+                playbackTracker.pausePlayback(id)
+            }
         }
     }
 }

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

@@ -29,9 +29,12 @@ import kotlinx.coroutines.channels.awaitClose
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.callbackFlow
 import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.firstOrNull
 import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.lastOrNull
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.runningReduce
+import kotlinx.coroutines.runBlocking
 import org.matrix.android.sdk.api.session.events.model.RelationType
 import org.matrix.android.sdk.api.session.room.model.message.MessageAudioEvent
 import org.matrix.android.sdk.api.session.room.model.message.asMessageAudioEvent
@@ -57,7 +60,7 @@ class GetLiveVoiceBroadcastChunksUseCase @Inject constructor(
         val existingChunks = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId)
                 .mapNotNull { timelineEvent -> timelineEvent.root.asMessageAudioEvent().takeIf { it.isVoiceBroadcast() } }
 
-        val voiceBroadcastEvent = getVoiceBroadcastEventUseCase.execute(voiceBroadcast)
+        val voiceBroadcastEvent = runBlocking { getVoiceBroadcastEventUseCase.execute(voiceBroadcast).firstOrNull()?.getOrNull() }
         val voiceBroadcastState = voiceBroadcastEvent?.content?.voiceBroadcastState
 
         return if (voiceBroadcastState == null || voiceBroadcastState == VoiceBroadcastState.STOPPED) {

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

@@ -16,12 +16,26 @@
 
 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 kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.mapNotNull
+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.unwrap
 import timber.log.Timber
 import javax.inject.Inject
 
@@ -29,14 +43,27 @@ class GetVoiceBroadcastEventUseCase @Inject constructor(
         private val session: Session,
 ) {
 
-    fun execute(voiceBroadcast: VoiceBroadcast): VoiceBroadcastEvent? {
+    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 relatedEvents = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId)
+        val latestEvent = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcast.voiceBroadcastId)
-                .sortedBy { it.root.originServerTs }
+                .mapNotNull { it.root.asVoiceBroadcastEvent() }
-        return relatedEvents.mapNotNull { it.root.asVoiceBroadcastEvent() }.lastOrNull() ?: initialEvent
+                .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()))
+                        .unwrap()
+                        .mapNotNull { it.asVoiceBroadcastEvent() }
+                        .filter { it.reference?.eventId == voiceBroadcast.voiceBroadcastId }
+                        .map { it.toOptional() }
+            }
+        }
     }
 }