VoiceBroadcastPlayer - Live listening

matrix-sdk-android/src/main/java/org/matrix/android/sdk/api/session/events/model/Event.kt

@@ -401,7 +401,7 @@ fun Event.getRelationContent(): RelationDefaultContent? {
             when (getClearType()) {
                 EventType.STICKER -> getClearContent().toModel<MessageStickerContent>()?.relatesTo
                 in EventType.BEACON_LOCATION_DATA -> getClearContent().toModel<MessageBeaconLocationDataContent>()?.relatesTo
-                else -> null
+                else -> getClearContent()?.get("m.relates_to")?.toContent().toModel()
             }
         }
     }

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/session/room/timeline/TimelineEventDataSource.kt

@@ -22,6 +22,8 @@ import io.realm.Sort
 import org.matrix.android.sdk.api.session.events.model.getRelationContent
 import org.matrix.android.sdk.api.session.events.model.isImageMessage
 import org.matrix.android.sdk.api.session.events.model.isVideoMessage
+import org.matrix.android.sdk.api.session.events.model.toModel
+import org.matrix.android.sdk.api.session.room.model.message.MessageContent
 import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
 import org.matrix.android.sdk.api.util.Optional
 import org.matrix.android.sdk.internal.database.RealmSessionProvider
@@ -74,7 +76,13 @@ internal class TimelineEventDataSource @Inject constructor(
                     .distinct(TimelineEventEntityFields.EVENT_ID)
                     .findAll()
                     .mapNotNull {
-                        timelineEventMapper.map(it).takeIf { it.root.getRelationContent()?.takeIf { it.type == eventType && it.eventId == eventId } != null }
+                        timelineEventMapper.map(it)
+                                .takeIf {
+                                    val isEventRelatedTo = it.root.getRelationContent()?.takeIf { it.type == eventType && it.eventId == eventId } != null
+                                    val isContentRelatedTo = it.root.getClearContent()?.toModel<MessageContent>()
+                                            ?.relatesTo?.takeIf { it.type == eventType && it.eventId == eventId } != null
+                                    isEventRelatedTo || isContentRelatedTo
+                                }
                     }
         }
     }

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

@@ -40,7 +40,7 @@ class VoiceBroadcastHelper @Inject constructor(
 
     suspend fun stopVoiceBroadcast(roomId: String) = stopVoiceBroadcastUseCase.execute(roomId)
 
-    fun playOrResumePlayback(roomId: String, eventId: String) = voiceBroadcastPlayer.play(roomId, eventId)
+    fun playOrResumePlayback(roomId: String, eventId: String) = voiceBroadcastPlayer.playOrResume(roomId, eventId)
 
     fun pausePlayback() = voiceBroadcastPlayer.pause()
 

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

@@ -20,13 +20,19 @@ import android.media.AudioAttributes
 import android.media.MediaPlayer
 import im.vector.app.core.di.ActiveSessionHolder
 import im.vector.app.features.home.room.detail.timeline.helper.AudioMessagePlaybackTracker
-import im.vector.app.features.home.room.detail.timeline.helper.AudioMessagePlaybackTracker.Listener.State
 import im.vector.app.features.voice.VoiceFailure
+import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
+import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
+import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastStateUseCase
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.mapNotNull
+import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
-import org.matrix.android.sdk.api.extensions.orFalse
+import org.matrix.android.sdk.api.query.QueryStringValue
 import org.matrix.android.sdk.api.session.events.model.RelationType
 import org.matrix.android.sdk.api.session.events.model.getRelationContent
 import org.matrix.android.sdk.api.session.getRoom
@@ -34,6 +40,11 @@ import org.matrix.android.sdk.api.session.room.Room
 import org.matrix.android.sdk.api.session.room.model.message.MessageAudioContent
 import org.matrix.android.sdk.api.session.room.model.message.MessageAudioEvent
 import org.matrix.android.sdk.api.session.room.model.message.asMessageAudioEvent
+import org.matrix.android.sdk.api.session.room.timeline.Timeline
+import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
+import org.matrix.android.sdk.api.session.room.timeline.TimelineSettings
+import org.matrix.android.sdk.flow.flow
+import org.matrix.android.sdk.flow.unwrap
 import timber.log.Timber
 import javax.inject.Inject
 import javax.inject.Singleton
@@ -42,62 +53,117 @@ import javax.inject.Singleton
 class VoiceBroadcastPlayer @Inject constructor(
         private val sessionHolder: ActiveSessionHolder,
         private val playbackTracker: AudioMessagePlaybackTracker,
+        private val getVoiceBroadcastStateUseCase: GetVoiceBroadcastStateUseCase,
 ) {
-    private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
     private val session
         get() = sessionHolder.getActiveSession()
 
+    private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+    private var voiceBroadcastStateJob: Job? = null
+    private var currentTimeline: Timeline? = null
+        set(value) {
+            field?.removeAllListeners()
+            field?.dispose()
+            field = value
+        }
+
+    private val mediaPlayerListener = MediaPlayerListener()
+    private var timelineListener: TimelineListener? = null
+
     private var currentMediaPlayer: MediaPlayer? = null
-    private var currentPlayingIndex: Int = -1
+    private var nextMediaPlayer: MediaPlayer? = null
+        set(value) {
+            field = value
+            currentMediaPlayer?.setNextMediaPlayer(value)
+        }
+    private var currentSequence: Int? = null
 
     private var playlist = emptyList<MessageAudioEvent>()
     private val currentVoiceBroadcastId
-        get() = playlist.getOrNull(currentPlayingIndex)?.root?.getRelationContent()?.eventId
+        get() = playlist.firstOrNull()?.root?.getRelationContent()?.eventId
 
-    private val mediaPlayerListener = MediaPlayerListener()
+    private var state: State = State.IDLE
-
+        set(value) {
-    fun play(roomId: String, eventId: String) {
+            Timber.w("## VoiceBroadcastPlayer state: $field -> $value")
-        val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
+            field = value
+        }
 
+    fun playOrResume(roomId: String, eventId: String) {
+        val hasChanged = currentVoiceBroadcastId != eventId
         when {
-            currentVoiceBroadcastId != eventId -> {
+            hasChanged -> startPlayback(roomId, eventId)
-                stop()
+            state == State.PAUSED -> resumePlayback()
-                updatePlaylist(room, eventId)
+            else -> Unit
-                startPlayback()
-            }
-            playbackTracker.getPlaybackState(eventId) is State.Playing -> pause()
-            else -> resumePlayback()
         }
     }
 
     fun pause() {
         currentMediaPlayer?.pause()
         currentVoiceBroadcastId?.let { playbackTracker.pausePlayback(it) }
+        state = State.PAUSED
     }
 
     fun stop() {
+        // Stop playback
         currentMediaPlayer?.stop()
         currentVoiceBroadcastId?.let { playbackTracker.stopPlayback(it) }
+
+        // Release current player
         release(currentMediaPlayer)
+        currentMediaPlayer = null
+
+        // Release next player
+        release(nextMediaPlayer)
+        nextMediaPlayer = null
+
+        // Do not observe anymore voice broadcast state changes
+        voiceBroadcastStateJob?.cancel()
+        voiceBroadcastStateJob = null
+
+        // In case of live broadcast, stop observing new chunks
+        currentTimeline?.dispose()
+        currentTimeline?.removeAllListeners()
+        currentTimeline = null
+        timelineListener = null
+
+        // Update state
+        state = State.IDLE
+
+        // Clear playlist
         playlist = emptyList()
-        currentPlayingIndex = -1
+        currentSequence = null
     }
 
-    private fun updatePlaylist(room: Room, eventId: String) {
+    private fun startPlayback(roomId: String, eventId: String) {
-        val timelineEvents = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, eventId)
+        val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
-        val audioEvents = timelineEvents.mapNotNull { it.root.asMessageAudioEvent() }
+
-        playlist = audioEvents.sortedBy { it.getVoiceBroadcastChunk()?.sequence?.toLong() ?: it.root.originServerTs }
+        // Stop listening previous voice broadcast if any
+        if (state != State.IDLE) stop()
+
+        state = State.BUFFERING
+
+        val voiceBroadcastState = getVoiceBroadcastStateUseCase.execute(roomId, eventId)
+        if (voiceBroadcastState == VoiceBroadcastState.STOPPED) {
+            // Get static playlist
+            updatePlaylist(getExistingChunks(room, eventId))
+            startPlayback(false)
+        } else {
+            playLiveVoiceBroadcast(room, eventId)
+        }
     }
 
-    private fun startPlayback() {
+    private fun startPlayback(isLive: Boolean) {
-        val content = playlist.firstOrNull()?.content ?: run { Timber.w("## VoiceBroadcastPlayer: No content to play"); return }
+        val event = if (isLive) playlist.lastOrNull() else playlist.firstOrNull()
+        val content = event?.content ?: run { Timber.w("## VoiceBroadcastPlayer: No content to play"); return }
+        val sequence = event.getVoiceBroadcastChunk()?.sequence
         coroutineScope.launch {
             try {
                 currentMediaPlayer = prepareMediaPlayer(content)
                 currentMediaPlayer?.start()
-                currentPlayingIndex = 0
                 currentVoiceBroadcastId?.let { playbackTracker.startPlayback(it) }
-                prepareNextFile()
+                currentSequence = sequence
+                state = State.PLAYING
+                nextMediaPlayer = prepareNextMediaPlayer()
             } catch (failure: Throwable) {
                 Timber.e(failure, "Unable to start playback")
                 throw VoiceFailure.UnableToPlay(failure)
@@ -105,19 +171,68 @@ class VoiceBroadcastPlayer @Inject constructor(
         }
     }
 
+    private fun playLiveVoiceBroadcast(room: Room, eventId: String) {
+        val voiceBroadcastEvent = room.timelineService().getTimelineEvent(eventId)?.root?.asVoiceBroadcastEvent()
+                ?: error("Cannot retrieve voice broadcast $eventId")
+        updatePlaylist(getExistingChunks(room, eventId))
+        startPlayback(true)
+        room.flow()
+                .liveStateEvent(VoiceBroadcastConstants.STATE_ROOM_VOICE_BROADCAST_INFO, QueryStringValue.Equals(voiceBroadcastEvent.root.stateKey!!))
+                .unwrap()
+                .mapNotNull { it.asVoiceBroadcastEvent()?.content?.voiceBroadcastState }
+                .onEach { state ->
+                    when (state) {
+                        VoiceBroadcastState.STARTED,
+                        VoiceBroadcastState.PAUSED,
+                        VoiceBroadcastState.RESUMED -> {
+                            observeIncomingChunks(room, eventId)
+                        }
+                        VoiceBroadcastState.STOPPED -> {
+                            currentTimeline?.dispose()
+                            currentTimeline?.removeAllListeners()
+                            currentTimeline = null
+                        }
+                    }
+                }
+                .launchIn(coroutineScope)
+    }
+
+    private fun getExistingChunks(room: Room, eventId: String): List<MessageAudioEvent> {
+        return room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, eventId)
+                .mapNotNull { it.root.asMessageAudioEvent() }
+                .filter { it.isVoiceBroadcast() }
+    }
+
+    private fun observeIncomingChunks(room: Room, eventId: String) {
+        // Fixme this is probably not necessary here
+        currentTimeline?.dispose()
+        currentTimeline?.removeAllListeners()
+        currentTimeline = room.timelineService().createTimeline(null, TimelineSettings(5)).also { timeline ->
+            timelineListener = TimelineListener(eventId).also { timeline.addListener(it) }
+            timeline.start()
+        }
+    }
+
     private fun resumePlayback() {
         currentMediaPlayer?.start()
         currentVoiceBroadcastId?.let { playbackTracker.startPlayback(it) }
+        state = State.PLAYING
     }
 
-    private suspend fun prepareNextFile() {
+    private fun updatePlaylist(playlist: List<MessageAudioEvent>) {
-        val nextContent = playlist.getOrNull(currentPlayingIndex + 1)?.content
+        this.playlist = playlist.sortedBy { it.getVoiceBroadcastChunk()?.sequence?.toLong() ?: it.root.originServerTs }
-        if (nextContent == null) {
+    }
-            currentMediaPlayer?.setOnCompletionListener(mediaPlayerListener)
+
-        } else {
+    private fun getNextAudioContent(): MessageAudioContent? {
-            val nextMediaPlayer = prepareMediaPlayer(nextContent)
+        val nextSequence = currentSequence?.plus(1)
-            currentMediaPlayer?.setNextMediaPlayer(nextMediaPlayer)
+                ?: timelineListener?.let { playlist.lastOrNull()?.sequence }
-        }
+                ?: 1
+        return playlist.find { it.getVoiceBroadcastChunk()?.sequence == nextSequence }?.content
+    }
+
+    private suspend fun prepareNextMediaPlayer(): MediaPlayer? {
+        val nextContent = getNextAudioContent() ?: return null
+        return prepareMediaPlayer(nextContent)
     }
 
     private suspend fun prepareMediaPlayer(messageAudioContent: MessageAudioContent): MediaPlayer {
@@ -141,6 +256,7 @@ class VoiceBroadcastPlayer @Inject constructor(
                 setDataSource(fis.fd)
                 setOnInfoListener(mediaPlayerListener)
                 setOnErrorListener(mediaPlayerListener)
+                setOnCompletionListener(mediaPlayerListener)
                 prepare()
             }
         }
@@ -155,24 +271,59 @@ class VoiceBroadcastPlayer @Inject constructor(
         }
     }
 
-    inner class MediaPlayerListener : MediaPlayer.OnInfoListener, MediaPlayer.OnCompletionListener, MediaPlayer.OnErrorListener {
+    private inner class TimelineListener(private val voiceBroadcastId: String) : Timeline.Listener {
+        override fun onTimelineUpdated(snapshot: List<TimelineEvent>) {
+            val currentSequences = playlist.map { it.sequence }
+            val newChunks = snapshot
+                    .mapNotNull { timelineEvent ->
+                        timelineEvent.root.asMessageAudioEvent()
+                                ?.takeIf { it.isVoiceBroadcast() && it.getVoiceBroadcastEventId() == voiceBroadcastId && it.sequence !in currentSequences }
+                    }
+            if (newChunks.isEmpty()) return
+            updatePlaylist(playlist + newChunks)
+
+            when (state) {
+                State.PLAYING -> {
+                    if (nextMediaPlayer == null) {
+                        coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
+                    }
+                }
+                State.PAUSED -> {
+                    if (nextMediaPlayer == null) {
+                        coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
+                    }
+                }
+                State.BUFFERING -> {
+                    val newMediaContent = getNextAudioContent()
+                    if (newMediaContent != null) startPlayback(true)
+                }
+                State.IDLE -> startPlayback(true)
+            }
+        }
+    }
+
+    private inner class MediaPlayerListener : MediaPlayer.OnInfoListener, 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
-                    currentPlayingIndex++
+                    currentSequence = currentSequence?.plus(1)
-                    coroutineScope.launch { prepareNextFile() }
+                    coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
                 }
             }
             return false
         }
 
         override fun onCompletion(mp: MediaPlayer) {
-            // Verify that a new media has not been set in the mean time
+            when {
-            if (!currentMediaPlayer?.isPlaying.orFalse()) {
+                timelineListener == null && nextMediaPlayer == null -> {
-                stop()
+                    stop()
+                }
+                nextMediaPlayer == null -> {
+                    state = State.BUFFERING
+                }
             }
         }
 
@@ -181,4 +332,11 @@ class VoiceBroadcastPlayer @Inject constructor(
             return true
         }
     }
+
+    enum class State {
+        PLAYING,
+        PAUSED,
+        BUFFERING,
+        IDLE
+    }
 }

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

@@ -0,0 +1,41 @@
+/*
+ * 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.model.VoiceBroadcastState
+import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
+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 timber.log.Timber
+import javax.inject.Inject
+
+class GetVoiceBroadcastStateUseCase @Inject constructor(
+        private val session: Session,
+) {
+
+    fun execute(roomId: String, eventId: String): VoiceBroadcastState? {
+        val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
+
+        Timber.d("## GetVoiceBroadcastStateUseCase: get voice broadcast state requested for $eventId")
+
+        val initialEvent = room.timelineService().getTimelineEvent(eventId)?.root?.asVoiceBroadcastEvent() // Fallback to initial event
+        val relatedEvents = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, eventId).sortedBy { it.root.originServerTs }
+        val lastVoiceBroadcastEvent = relatedEvents.mapNotNull { it.root.asVoiceBroadcastEvent() }.lastOrNull() ?: initialEvent
+        return lastVoiceBroadcastEvent?.content?.voiceBroadcastState
+    }
+}