Skip to content

Commit 3fcac09

Browse files
author
Florian Renaud
committed
VoiceBroadcastPlayer - Fetch playlist in dedicated use case and improve player
1 parent 174ba4f commit 3fcac09

File tree

2 files changed

+174
-86
lines changed

2 files changed

+174
-86
lines changed

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

Lines changed: 44 additions & 86 deletions
Original file line numberDiff line numberDiff line change
@@ -23,53 +23,42 @@ import im.vector.app.core.di.ActiveSessionHolder
2323
import im.vector.app.features.home.room.detail.timeline.helper.AudioMessagePlaybackTracker
2424
import im.vector.app.features.voice.VoiceFailure
2525
import im.vector.app.features.voicebroadcast.getVoiceBroadcastChunk
26-
import im.vector.app.features.voicebroadcast.getVoiceBroadcastEventId
27-
import im.vector.app.features.voicebroadcast.isVoiceBroadcast
2826
import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer.Listener
2927
import im.vector.app.features.voicebroadcast.listening.VoiceBroadcastPlayer.State
28+
import im.vector.app.features.voicebroadcast.listening.usecase.GetLiveVoiceBroadcastChunksUseCase
3029
import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
31-
import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
3230
import im.vector.app.features.voicebroadcast.sequence
3331
import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastUseCase
3432
import kotlinx.coroutines.CoroutineScope
3533
import kotlinx.coroutines.Dispatchers
3634
import kotlinx.coroutines.Job
3735
import kotlinx.coroutines.SupervisorJob
36+
import kotlinx.coroutines.flow.launchIn
37+
import kotlinx.coroutines.flow.onEach
3838
import kotlinx.coroutines.launch
3939
import kotlinx.coroutines.withContext
40-
import org.matrix.android.sdk.api.session.events.model.RelationType
41-
import org.matrix.android.sdk.api.session.getRoom
42-
import org.matrix.android.sdk.api.session.room.Room
4340
import org.matrix.android.sdk.api.session.room.model.message.MessageAudioContent
4441
import org.matrix.android.sdk.api.session.room.model.message.MessageAudioEvent
45-
import org.matrix.android.sdk.api.session.room.model.message.asMessageAudioEvent
46-
import org.matrix.android.sdk.api.session.room.timeline.Timeline
47-
import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
48-
import org.matrix.android.sdk.api.session.room.timeline.TimelineSettings
4942
import timber.log.Timber
5043
import java.util.concurrent.CopyOnWriteArrayList
5144
import javax.inject.Inject
45+
import javax.inject.Singleton
5246

5347
@Singleton
5448
class VoiceBroadcastPlayerImpl @Inject constructor(
5549
private val sessionHolder: ActiveSessionHolder,
5650
private val playbackTracker: AudioMessagePlaybackTracker,
5751
private val getVoiceBroadcastUseCase: GetVoiceBroadcastUseCase,
52+
private val getLiveVoiceBroadcastChunksUseCase: GetLiveVoiceBroadcastChunksUseCase
5853
) : VoiceBroadcastPlayer {
54+
5955
private val session
6056
get() = sessionHolder.getActiveSession()
6157

6258
private val coroutineScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
6359
private var voiceBroadcastStateJob: Job? = null
64-
private var currentTimeline: Timeline? = null
65-
set(value) {
66-
field?.removeAllListeners()
67-
field?.dispose()
68-
field = value
69-
}
7060

7161
private val mediaPlayerListener = MediaPlayerListener()
72-
private var timelineListener: TimelineListener? = null
7362

7463
private var currentMediaPlayer: MediaPlayer? = null
7564
private var nextMediaPlayer: MediaPlayer? = null
@@ -79,7 +68,10 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
7968
}
8069
private var currentSequence: Int? = null
8170

71+
private var fetchPlaylistJob: Job? = null
8272
private var playlist = emptyList<MessageAudioEvent>()
73+
private var isLive: Boolean = false
74+
8375
override var currentVoiceBroadcastId: String? = null
8476

8577
override var playingState = State.IDLE
@@ -118,6 +110,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
118110
// Stop playback
119111
currentMediaPlayer?.stop()
120112
currentVoiceBroadcastId?.let { playbackTracker.stopPlayback(it) }
113+
isLive = false
121114

122115
// Release current player
123116
release(currentMediaPlayer)
@@ -131,39 +124,33 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
131124
voiceBroadcastStateJob?.cancel()
132125
voiceBroadcastStateJob = null
133126

134-
// In case of live broadcast, stop observing new chunks
135-
currentTimeline = null
136-
timelineListener = null
127+
// Do not fetch the playlist anymore
128+
fetchPlaylistJob?.cancel()
129+
fetchPlaylistJob = null
137130

138131
// Update state
139132
playingState = State.IDLE
140133

141134
// Clear playlist
142135
playlist = emptyList()
143136
currentSequence = null
137+
144138
currentRoomId = null
145139
currentVoiceBroadcastId = null
146140
}
147141

148-
/**
149-
* Add a [Listener] to the given voice broadcast id.
150-
*/
151142
override fun addListener(voiceBroadcastId: String, listener: Listener) {
152143
listeners[voiceBroadcastId]?.add(listener) ?: run {
153144
listeners[voiceBroadcastId] = CopyOnWriteArrayList<Listener>().apply { add(listener) }
154145
}
155146
if (voiceBroadcastId == currentVoiceBroadcastId) listener.onStateChanged(playingState) else listener.onStateChanged(State.IDLE)
156147
}
157148

158-
/**
159-
* Remove a [Listener] from the given voice broadcast id.
160-
*/
161149
override fun removeListener(voiceBroadcastId: String, listener: Listener) {
162150
listeners[voiceBroadcastId]?.remove(listener)
163151
}
164152

165153
private fun startPlayback(roomId: String, eventId: String) {
166-
val room = session.getRoom(roomId) ?: error("Unknown roomId: $roomId")
167154
// Stop listening previous voice broadcast if any
168155
if (playingState != State.IDLE) stop()
169156

@@ -173,16 +160,11 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
173160
playingState = State.BUFFERING
174161

175162
val voiceBroadcastState = getVoiceBroadcastUseCase.execute(roomId, eventId)?.content?.voiceBroadcastState
176-
if (voiceBroadcastState == VoiceBroadcastState.STOPPED) {
177-
// Get static playlist
178-
updatePlaylist(getExistingChunks(room, eventId))
179-
startPlayback(false)
180-
} else {
181-
playLiveVoiceBroadcast(room, eventId)
182-
}
163+
isLive = voiceBroadcastState != null && voiceBroadcastState != VoiceBroadcastState.STOPPED
164+
observeIncomingEvents(roomId, eventId)
183165
}
184166

185-
private fun startPlayback(isLive: Boolean) {
167+
private fun startPlayback() {
186168
val event = if (isLive) playlist.lastOrNull() else playlist.firstOrNull()
187169
val content = event?.content ?: run { Timber.w("## VoiceBroadcastPlayer: No content to play"); return }
188170
val sequence = event.getVoiceBroadcastChunk()?.sequence
@@ -201,24 +183,10 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
201183
}
202184
}
203185

204-
private fun playLiveVoiceBroadcast(room: Room, eventId: String) {
205-
room.timelineService().getTimelineEvent(eventId)?.root?.asVoiceBroadcastEvent() ?: error("Cannot retrieve voice broadcast $eventId")
206-
updatePlaylist(getExistingChunks(room, eventId))
207-
startPlayback(true)
208-
observeIncomingEvents(room, eventId)
209-
}
210-
211-
private fun getExistingChunks(room: Room, eventId: String): List<MessageAudioEvent> {
212-
return room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, eventId)
213-
.mapNotNull { it.root.asMessageAudioEvent() }
214-
.filter { it.isVoiceBroadcast() }
215-
}
216-
217-
private fun observeIncomingEvents(room: Room, eventId: String) {
218-
currentTimeline = room.timelineService().createTimeline(null, TimelineSettings(5)).also { timeline ->
219-
timelineListener = TimelineListener(eventId).also { timeline.addListener(it) }
220-
timeline.start()
221-
}
186+
private fun observeIncomingEvents(roomId: String, voiceBroadcastId: String) {
187+
fetchPlaylistJob = getLiveVoiceBroadcastChunksUseCase.execute(roomId, voiceBroadcastId)
188+
.onEach(this::updatePlaylist)
189+
.launchIn(coroutineScope)
222190
}
223191

224192
private fun resumePlayback() {
@@ -229,11 +197,32 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
229197

230198
private fun updatePlaylist(playlist: List<MessageAudioEvent>) {
231199
this.playlist = playlist.sortedBy { it.getVoiceBroadcastChunk()?.sequence?.toLong() ?: it.root.originServerTs }
200+
onPlaylistUpdated()
201+
}
202+
203+
private fun onPlaylistUpdated() {
204+
when (playingState) {
205+
State.PLAYING -> {
206+
if (nextMediaPlayer == null) {
207+
coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
208+
}
209+
}
210+
State.PAUSED -> {
211+
if (nextMediaPlayer == null) {
212+
coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
213+
}
214+
}
215+
State.BUFFERING -> {
216+
val newMediaContent = getNextAudioContent()
217+
if (newMediaContent != null) startPlayback()
218+
}
219+
State.IDLE -> startPlayback()
220+
}
232221
}
233222

234223
private fun getNextAudioContent(): MessageAudioContent? {
235224
val nextSequence = currentSequence?.plus(1)
236-
?: timelineListener?.let { playlist.lastOrNull()?.sequence }
225+
?: playlist.lastOrNull()?.sequence
237226
?: 1
238227
return playlist.find { it.getVoiceBroadcastChunk()?.sequence == nextSequence }?.content
239228
}
@@ -279,37 +268,6 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
279268
}
280269
}
281270

282-
private inner class TimelineListener(private val voiceBroadcastId: String) : Timeline.Listener {
283-
override fun onTimelineUpdated(snapshot: List<TimelineEvent>) {
284-
val currentSequences = playlist.map { it.sequence }
285-
val newChunks = snapshot
286-
.mapNotNull { timelineEvent ->
287-
timelineEvent.root.asMessageAudioEvent()
288-
?.takeIf { it.isVoiceBroadcast() && it.getVoiceBroadcastEventId() == voiceBroadcastId && it.sequence !in currentSequences }
289-
}
290-
if (newChunks.isEmpty()) return
291-
updatePlaylist(playlist + newChunks)
292-
293-
when (playingState) {
294-
State.PLAYING -> {
295-
if (nextMediaPlayer == null) {
296-
coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
297-
}
298-
}
299-
State.PAUSED -> {
300-
if (nextMediaPlayer == null) {
301-
coroutineScope.launch { nextMediaPlayer = prepareNextMediaPlayer() }
302-
}
303-
}
304-
State.BUFFERING -> {
305-
val newMediaContent = getNextAudioContent()
306-
if (newMediaContent != null) startPlayback(true)
307-
}
308-
State.IDLE -> startPlayback(true)
309-
}
310-
}
311-
}
312-
313271
private inner class MediaPlayerListener : MediaPlayer.OnInfoListener, MediaPlayer.OnCompletionListener, MediaPlayer.OnErrorListener {
314272

315273
override fun onInfo(mp: MediaPlayer, what: Int, extra: Int): Boolean {
@@ -329,7 +287,7 @@ class VoiceBroadcastPlayerImpl @Inject constructor(
329287
val roomId = currentRoomId ?: return
330288
val voiceBroadcastId = currentVoiceBroadcastId ?: return
331289
val voiceBroadcastEventContent = getVoiceBroadcastUseCase.execute(roomId, voiceBroadcastId)?.content ?: return
332-
val isLive = voiceBroadcastEventContent.voiceBroadcastState != null && voiceBroadcastEventContent.voiceBroadcastState != VoiceBroadcastState.STOPPED
290+
isLive = voiceBroadcastEventContent.voiceBroadcastState != null && voiceBroadcastEventContent.voiceBroadcastState != VoiceBroadcastState.STOPPED
333291

334292
if (!isLive && voiceBroadcastEventContent.lastChunkSequence == currentSequence) {
335293
// We'll not receive new chunks anymore so we can stop the live listening
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/*
2+
* Copyright (c) 2022 New Vector Ltd
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package im.vector.app.features.voicebroadcast.listening.usecase
18+
19+
import im.vector.app.core.di.ActiveSessionHolder
20+
import im.vector.app.features.voicebroadcast.getVoiceBroadcastEventId
21+
import im.vector.app.features.voicebroadcast.isVoiceBroadcast
22+
import im.vector.app.features.voicebroadcast.model.VoiceBroadcastEvent
23+
import im.vector.app.features.voicebroadcast.model.VoiceBroadcastState
24+
import im.vector.app.features.voicebroadcast.model.asVoiceBroadcastEvent
25+
import im.vector.app.features.voicebroadcast.sequence
26+
import im.vector.app.features.voicebroadcast.usecase.GetVoiceBroadcastUseCase
27+
import kotlinx.coroutines.channels.awaitClose
28+
import kotlinx.coroutines.flow.Flow
29+
import kotlinx.coroutines.flow.callbackFlow
30+
import kotlinx.coroutines.flow.emptyFlow
31+
import kotlinx.coroutines.flow.flowOf
32+
import kotlinx.coroutines.flow.runningReduce
33+
import org.matrix.android.sdk.api.session.events.model.RelationType
34+
import org.matrix.android.sdk.api.session.room.model.message.MessageAudioEvent
35+
import org.matrix.android.sdk.api.session.room.model.message.asMessageAudioEvent
36+
import org.matrix.android.sdk.api.session.room.timeline.Timeline
37+
import org.matrix.android.sdk.api.session.room.timeline.TimelineEvent
38+
import org.matrix.android.sdk.api.session.room.timeline.TimelineSettings
39+
import javax.inject.Inject
40+
41+
/**
42+
* Get a [Flow] of [MessageAudioEvent]s related to the given voice broadcast.
43+
*/
44+
class GetLiveVoiceBroadcastChunksUseCase @Inject constructor(
45+
private val activeSessionHolder: ActiveSessionHolder,
46+
private val getVoiceBroadcastUseCase: GetVoiceBroadcastUseCase,
47+
) {
48+
49+
fun execute(roomId: String, voiceBroadcastId: String): Flow<List<MessageAudioEvent>> {
50+
val session = activeSessionHolder.getSafeActiveSession() ?: return emptyFlow()
51+
val room = session.roomService().getRoom(roomId) ?: return emptyFlow()
52+
val timeline = room.timelineService().createTimeline(null, TimelineSettings(5))
53+
54+
// Get initial chunks
55+
val existingChunks = room.timelineService().getTimelineEventsRelatedTo(RelationType.REFERENCE, voiceBroadcastId)
56+
.mapNotNull { timelineEvent -> timelineEvent.root.asMessageAudioEvent().takeIf { it.isVoiceBroadcast() } }
57+
58+
val voiceBroadcastEvent = getVoiceBroadcastUseCase.execute(roomId, voiceBroadcastId)
59+
val voiceBroadcastState = voiceBroadcastEvent?.content?.voiceBroadcastState
60+
61+
return if (voiceBroadcastState == null || voiceBroadcastState == VoiceBroadcastState.STOPPED) {
62+
// Just send the existing chunks if voice broadcast is stopped
63+
flowOf(existingChunks)
64+
} else {
65+
// Observe new timeline events if voice broadcast is ongoing
66+
callbackFlow {
67+
// Init with existing chunks
68+
send(existingChunks)
69+
70+
// Observe new timeline events
71+
val listener = object : Timeline.Listener {
72+
private var lastEventId: String? = null
73+
private var lastSequence: Int? = null
74+
75+
override fun onTimelineUpdated(snapshot: List<TimelineEvent>) {
76+
val newEvents = lastEventId?.let { eventId -> snapshot.subList(0, snapshot.indexOfFirst { it.eventId == eventId }) } ?: snapshot
77+
78+
// Detect a potential stopped voice broadcast state event
79+
val stopEvent = newEvents.findStopEvent()
80+
if (stopEvent != null) {
81+
lastSequence = stopEvent.content?.lastChunkSequence
82+
}
83+
84+
val newChunks = newEvents.mapToChunkEvents(voiceBroadcastId, voiceBroadcastEvent.root.senderId)
85+
86+
// Notify about new chunks
87+
if (newChunks.isNotEmpty()) {
88+
trySend(newChunks)
89+
}
90+
91+
// Automatically stop observing the timeline if the last chunk has been received
92+
if (lastSequence != null && newChunks.any { it.sequence == lastSequence }) {
93+
timeline.removeListener(this)
94+
timeline.dispose()
95+
}
96+
97+
lastEventId = snapshot.firstOrNull()?.eventId
98+
}
99+
}
100+
101+
timeline.addListener(listener)
102+
timeline.start()
103+
awaitClose {
104+
timeline.removeListener(listener)
105+
timeline.dispose()
106+
}
107+
}
108+
.runningReduce { accumulator: List<MessageAudioEvent>, value: List<MessageAudioEvent> -> accumulator.plus(value) }
109+
}
110+
}
111+
112+
/**
113+
* Find a [VoiceBroadcastEvent] with a [VoiceBroadcastState.STOPPED] state.
114+
*/
115+
private fun List<TimelineEvent>.findStopEvent(): VoiceBroadcastEvent? =
116+
this.mapNotNull { it.root.asVoiceBroadcastEvent() }
117+
.find { it.content?.voiceBroadcastState == VoiceBroadcastState.STOPPED }
118+
119+
/**
120+
* Transform the list of [TimelineEvent] to a mapped list of [MessageAudioEvent] related to a given voice broadcast.
121+
*/
122+
private fun List<TimelineEvent>.mapToChunkEvents(voiceBroadcastId: String, senderId: String?): List<MessageAudioEvent> =
123+
this.mapNotNull { timelineEvent ->
124+
timelineEvent.root.asMessageAudioEvent()
125+
?.takeIf {
126+
it.isVoiceBroadcast() && it.getVoiceBroadcastEventId() == voiceBroadcastId &&
127+
it.root.senderId == senderId
128+
}
129+
}
130+
}

0 commit comments

Comments
 (0)