Merge pull request #7419 from vector-im/feature/fre/voice_broadcast_live_listening

Voice broadcast - live listening

changelog.d/7419.wip

@@ -0,0 +1 @@
+[Voice Broadcast] Live listening support

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

@@ -18,14 +18,17 @@ package im.vector.app.features.voicebroadcast
 
 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.GetVoiceBroadcastUseCase
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.SupervisorJob
 import kotlinx.coroutines.launch
-import org.matrix.android.sdk.api.extensions.orFalse
-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.events.model.getRelationContent
 import org.matrix.android.sdk.api.session.getRoom
@@ -33,70 +36,129 @@ 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 timber.log.Timber
 import javax.inject.Inject
 import javax.inject.Singleton
 
 @Singleton
 class VoiceBroadcastPlayer @Inject constructor(
-        private val session: Session,
+        private val sessionHolder: ActiveSessionHolder,
         private val playbackTracker: AudioMessagePlaybackTracker,
+        private val getVoiceBroadcastUseCase: GetVoiceBroadcastUseCase,
 ) {
+    private val session
+        get() = sessionHolder.getActiveSession()
 
-    private val mediaPlayerScope = CoroutineScope(Dispatchers.IO)
+    private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-
+    private var voiceBroadcastStateJob: Job? = null
-    private var currentMediaPlayer: MediaPlayer? = null
+    private var currentTimeline: Timeline? = null
-    private var currentPlayingIndex: Int = -1
+        set(value) {
-    private var playlist = emptyList<MessageAudioEvent>()
+            field?.removeAllListeners()
-    private val currentVoiceBroadcastEventId
+            field?.dispose()
-        get() = playlist.firstOrNull()?.root?.getRelationContent()?.eventId
+            field = value
+        }
 
     private val mediaPlayerListener = MediaPlayerListener()
+    private var timelineListener: TimelineListener? = null
 
-    fun play(roomId: String, eventId: String) {
+    private var currentMediaPlayer: MediaPlayer? = null
-        val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
+    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.firstOrNull()?.root?.getRelationContent()?.eventId
+
+    private var state: State = State.IDLE
+        set(value) {
+            Timber.w("## VoiceBroadcastPlayer state: $field -> $value")
+            field = value
+        }
+    private var currentRoomId: String? = null
+
+    fun playOrResume(roomId: String, eventId: String) {
+        val hasChanged = currentVoiceBroadcastId != eventId
         when {
-            currentVoiceBroadcastEventId != 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()
-        currentVoiceBroadcastEventId?.let { playbackTracker.pausePlayback(it) }
+        currentVoiceBroadcastId?.let { playbackTracker.pausePlayback(it) }
+        state = State.PAUSED
     }
 
     fun stop() {
+        // Stop playback
         currentMediaPlayer?.stop()
-        currentMediaPlayer?.release()
+        currentVoiceBroadcastId?.let { playbackTracker.stopPlayback(it) }
-        currentMediaPlayer?.setOnInfoListener(null)
+
+        // Release current player
+        release(currentMediaPlayer)
         currentMediaPlayer = null
-        currentVoiceBroadcastEventId?.let { playbackTracker.stopPlayback(it) }
+
+        // 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 = null
+        timelineListener = null
+
+        // Update state
+        state = State.IDLE
+
+        // Clear playlist
         playlist = emptyList()
-        currentPlayingIndex = -1
+        currentSequence = null
+        currentRoomId = 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() }
+        currentRoomId = roomId
-        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 = getVoiceBroadcastUseCase.execute(roomId, eventId)?.content?.voiceBroadcastState
+        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()
-        mediaPlayerScope.launch {
+        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) }
-                currentVoiceBroadcastEventId?.let { playbackTracker.startPlayback(it) }
+                currentSequence = sequence
-                prepareNextFile()
+                state = State.PLAYING
+                nextMediaPlayer = prepareNextMediaPlayer()
             } catch (failure: Throwable) {
                 Timber.e(failure, "Unable to start playback")
                 throw VoiceFailure.UnableToPlay(failure)
@@ -104,21 +166,48 @@ class VoiceBroadcastPlayer @Inject constructor(
         }
     }
 
-    private fun resumePlayback() {
+    private fun playLiveVoiceBroadcast(room: Room, eventId: String) {
-        currentMediaPlayer?.start()
+        room.timelineService().getTimelineEvent(eventId)?.root?.asVoiceBroadcastEvent() ?: error("Cannot retrieve voice broadcast $eventId")
-        currentVoiceBroadcastEventId?.let { playbackTracker.startPlayback(it) }
+        updatePlaylist(getExistingChunks(room, eventId))
+        startPlayback(true)
+        observeIncomingEvents(room, eventId)
     }
 
-    private suspend fun prepareNextFile() {
+    private fun getExistingChunks(room: Room, eventId: String): List<MessageAudioEvent> {
-        val nextContent = playlist.getOrNull(currentPlayingIndex + 1)?.content
+        return room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, eventId)
-        if (nextContent == null) {
+                .mapNotNull { it.root.asMessageAudioEvent() }
-            currentMediaPlayer?.setOnCompletionListener(mediaPlayerListener)
+                .filter { it.isVoiceBroadcast() }
-        } else {
+    }
-            val nextMediaPlayer = prepareMediaPlayer(nextContent)
+
-            currentMediaPlayer?.setNextMediaPlayer(nextMediaPlayer)
+    private fun observeIncomingEvents(room: Room, eventId: String) {
+        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 fun updatePlaylist(playlist: List<MessageAudioEvent>) {
+        this.playlist = playlist.sortedBy { it.getVoiceBroadcastChunk()?.sequence?.toLong() ?: it.root.originServerTs }
+    }
+
+    private fun getNextAudioContent(): MessageAudioContent? {
+        val nextSequence = currentSequence?.plus(1)
+                ?: 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 {
         // Download can fail
         val audioFile = try {
@@ -140,28 +229,78 @@ class VoiceBroadcastPlayer @Inject constructor(
                 setDataSource(fis.fd)
                 setOnInfoListener(mediaPlayerListener)
                 setOnErrorListener(mediaPlayerListener)
+                setOnCompletionListener(mediaPlayerListener)
                 prepare()
             }
         }
     }
 
-    inner class MediaPlayerListener : MediaPlayer.OnInfoListener, MediaPlayer.OnCompletionListener, MediaPlayer.OnErrorListener {
+    private fun release(mp: MediaPlayer?) {
+        mp?.apply {
+            release()
+            setOnInfoListener(null)
+            setOnCompletionListener(null)
+            setOnErrorListener(null)
+        }
+    }
+
+    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)
-                    mediaPlayerScope.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
+            if (nextMediaPlayer != null) return
-            if (!currentMediaPlayer?.isPlaying.orFalse()) {
+            val roomId = currentRoomId ?: return
+            val voiceBroadcastId = currentVoiceBroadcastId ?: return
+            val voiceBroadcastEventContent = getVoiceBroadcastUseCase.execute(roomId, voiceBroadcastId)?.content ?: return
+            val isLive = voiceBroadcastEventContent.voiceBroadcastState != null && voiceBroadcastEventContent.voiceBroadcastState != VoiceBroadcastState.STOPPED
+
+            if (!isLive && voiceBroadcastEventContent.lastChunkSequence == currentSequence) {
+                // We'll not receive new chunks anymore so we can stop the live listening
                 stop()
+            } else {
+                state = State.BUFFERING
             }
         }
 
@@ -170,4 +309,11 @@ class VoiceBroadcastPlayer @Inject constructor(
             return true
         }
     }
+
+    enum class State {
+        PLAYING,
+        PAUSED,
+        BUFFERING,
+        IDLE
+    }
 }

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

@@ -23,6 +23,7 @@ import java.io.File
 interface VoiceBroadcastRecorder : VoiceRecorder {
 
     var listener: Listener?
+    var currentSequence: Int
 
     fun startRecord(roomId: String, chunkLength: Int)
 

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

@@ -21,6 +21,7 @@ import android.media.MediaRecorder
 import android.os.Build
 import androidx.annotation.RequiresApi
 import im.vector.app.features.voice.AbstractVoiceRecorderQ
+import org.matrix.android.sdk.api.extensions.tryOrNull
 import org.matrix.android.sdk.api.session.content.ContentAttachmentData
 
 @RequiresApi(Build.VERSION_CODES.Q)
@@ -29,7 +30,8 @@ class VoiceBroadcastRecorderQ(
 ) : AbstractVoiceRecorderQ(context), VoiceBroadcastRecorder {
 
     private var maxFileSize = 0L // zero or negative for no limit
-    private var currentSequence = 0
+    private var currentRoomId: String? = null
+    override var currentSequence = 0
 
     override var listener: VoiceBroadcastRecorder.Listener? = null
 
@@ -51,11 +53,23 @@ class VoiceBroadcastRecorderQ(
     }
 
     override fun startRecord(roomId: String, chunkLength: Int) {
+        currentRoomId = roomId
         maxFileSize = (chunkLength * audioEncodingBitRate / 8).toLong()
         currentSequence = 1
         startRecord(roomId)
     }
 
+    override fun pauseRecord() {
+        tryOrNull { mediaRecorder?.stop() }
+        mediaRecorder?.reset()
+        notifyOutputFileCreated()
+    }
+
+    override fun resumeRecord() {
+        currentSequence++
+        currentRoomId?.let { startRecord(it) }
+    }
+
     override fun stopRecord() {
         super.stopRecord()
         notifyOutputFileCreated()

vector/src/main/java/im/vector/app/features/voicebroadcast/model/MessageVoiceBroadcastInfoContent.kt

@@ -44,6 +44,8 @@ data class MessageVoiceBroadcastInfoContent(
         @Json(name = "state") val voiceBroadcastStateStr: String = "",
         /** The length of the voice chunks in seconds. **/
         @Json(name = "chunk_length") val chunkLength: Int? = null,
+        /** The sequence of the last sent chunk. **/
+        @Json(name = "last_chunk_sequence") val lastChunkSequence: Int? = null,
 ) : MessageContent {
 
     val voiceBroadcastState: VoiceBroadcastState? = VoiceBroadcastState.values()

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

@@ -0,0 +1,40 @@
+/*
+ * 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.VoiceBroadcastEvent
+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 GetVoiceBroadcastUseCase @Inject constructor(
+        private val session: Session,
+) {
+
+    fun execute(roomId: String, eventId: String): VoiceBroadcastEvent? {
+        val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
+
+        Timber.d("## GetVoiceBroadcastUseCase: get voice broadcast $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 }
+        return relatedEvents.mapNotNull { it.root.asVoiceBroadcastEvent() }.lastOrNull() ?: initialEvent
+    }
+}

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

@@ -59,6 +59,7 @@ class PauseVoiceBroadcastUseCase @Inject constructor(
                 body = MessageVoiceBroadcastInfoContent(
                         relatesTo = reference,
                         voiceBroadcastStateStr = VoiceBroadcastState.PAUSED.value,
+                        lastChunkSequence = voiceBroadcastRecorder?.currentSequence,
                 ).toContent(),
         )
 

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

@@ -55,7 +55,7 @@ class StartVoiceBroadcastUseCase @Inject constructor(
                 QueryStringValue.IsNotEmpty
         )
                 .mapNotNull { it.asVoiceBroadcastEvent() }
-                .filter { it.content?.voiceBroadcastState != VoiceBroadcastState.STOPPED }
+                .filter { it.content?.voiceBroadcastState != null && it.content?.voiceBroadcastState != VoiceBroadcastState.STOPPED }
 
         if (onGoingVoiceBroadcastEvents.isEmpty()) {
             startVoiceBroadcast(room)

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

@@ -60,6 +60,7 @@ class StopVoiceBroadcastUseCase @Inject constructor(
                 body = MessageVoiceBroadcastInfoContent(
                         relatesTo = reference,
                         voiceBroadcastStateStr = VoiceBroadcastState.STOPPED.value,
+                        lastChunkSequence = voiceBroadcastRecorder?.currentSequence,
                 ).toContent(),
         )