Handle voice broadcast last_chunk_sequence (#9812)

src/voice-broadcast/models/VoiceBroadcastPlayback.ts

@@ -31,7 +31,12 @@ import { PlaybackManager } from "../../audio/PlaybackManager";
 import { UPDATE_EVENT } from "../../stores/AsyncStore";
 import { MediaEventHelper } from "../../utils/MediaEventHelper";
 import { IDestroyable } from "../../utils/IDestroyable";
-import { VoiceBroadcastLiveness, VoiceBroadcastInfoEventType, VoiceBroadcastInfoState } from "..";
+import {
+    VoiceBroadcastLiveness,
+    VoiceBroadcastInfoEventType,
+    VoiceBroadcastInfoState,
+    VoiceBroadcastInfoEventContent,
+} from "..";
 import { RelationsHelper, RelationsHelperEvent } from "../../events/RelationsHelper";
 import { VoiceBroadcastChunkEvents } from "../utils/VoiceBroadcastChunkEvents";
 import { determineVoiceBroadcastLiveness } from "../utils/determineVoiceBroadcastLiveness";
@@ -151,12 +156,20 @@ export class VoiceBroadcastPlayback
         this.setDuration(this.chunkEvents.getLength());
 
         if (this.getState() === VoiceBroadcastPlaybackState.Buffering) {
-            await this.start();
+            await this.startOrPlayNext();
         }
 
         return true;
     };
 
+    private startOrPlayNext = async (): Promise<void> => {
+        if (this.currentlyPlaying) {
+            return this.playNext();
+        }
+
+        return await this.start();
+    };
+
     private addInfoEvent = (event: MatrixEvent): void => {
         if (this.lastInfoEvent && this.lastInfoEvent.getTs() >= event.getTs()) {
             // Only handle newer events
@@ -263,7 +276,10 @@ export class VoiceBroadcastPlayback
             return this.playEvent(next);
         }
 
-        if (this.getInfoState() === VoiceBroadcastInfoState.Stopped) {
+        if (
+            this.getInfoState() === VoiceBroadcastInfoState.Stopped &&
+            this.chunkEvents.getSequenceForEvent(this.currentlyPlaying) === this.lastChunkSequence
+        ) {
             this.stop();
         } else {
             // No more chunks available, although the broadcast is not finished → enter buffering state.
@@ -271,6 +287,17 @@ export class VoiceBroadcastPlayback
         }
     }
 
+    /**
+     * @returns {number} The last chunk sequence from the latest info event.
+     *                   Falls back to the length of received chunks if the info event does not provide the number.
+     */
+    private get lastChunkSequence(): number {
+        return (
+            this.lastInfoEvent.getContent<VoiceBroadcastInfoEventContent>()?.last_chunk_sequence ||
+            this.chunkEvents.getNumberOfEvents()
+        );
+    }
+
     private async playEvent(event: MatrixEvent): Promise<void> {
         this.setState(VoiceBroadcastPlaybackState.Playing);
         this.currentlyPlaying = event;

src/voice-broadcast/utils/VoiceBroadcastChunkEvents.ts

@@ -97,6 +97,19 @@ export class VoiceBroadcastChunkEvents {
         return this.events.indexOf(event) >= this.events.length - 1;
     }
 
+    public getSequenceForEvent(event: MatrixEvent): number | null {
+        const sequence = parseInt(event.getContent()?.[VoiceBroadcastChunkEventType]?.sequence, 10);
+        if (!isNaN(sequence)) return sequence;
+
+        if (this.events.includes(event)) return this.events.indexOf(event) + 1;
+
+        return null;
+    }
+
+    public getNumberOfEvents(): number {
+        return this.events.length;
+    }
+
     private calculateChunkLength(event: MatrixEvent): number {
         return event.getContent()?.["org.matrix.msc1767.audio"]?.duration || event.getContent()?.info?.duration || 0;
     }

test/voice-broadcast/models/VoiceBroadcastPlayback-test.ts

@@ -279,26 +279,62 @@ describe("VoiceBroadcastPlayback", () => {
                 expect(chunk1Playback.play).not.toHaveBeenCalled();
             });
 
-            describe("and the playback of the last chunk ended", () => {
+            describe(
-                beforeEach(() => {
+                "and receiving a stop info event with last_chunk_sequence = 2 and " +
-                    chunk2Playback.emit(PlaybackState.Stopped);
+                    "the playback of the last available chunk ends",
-                });
+                () => {
-
-                itShouldSetTheStateTo(VoiceBroadcastPlaybackState.Buffering);
-
-                describe("and the next chunk arrived", () => {
                     beforeEach(() => {
-                        room.addLiveEvents([chunk3Event]);
+                        const stoppedEvent = mkVoiceBroadcastInfoStateEvent(
-                        room.relations.aggregateChildEvent(chunk3Event);
+                            roomId,
+                            VoiceBroadcastInfoState.Stopped,
+                            client.getSafeUserId(),
+                            client.deviceId!,
+                            infoEvent,
+                            2,
+                        );
+                        room.addLiveEvents([stoppedEvent]);
+                        room.relations.aggregateChildEvent(stoppedEvent);
+                        chunk2Playback.emit(PlaybackState.Stopped);
                     });
 
-                    itShouldSetTheStateTo(VoiceBroadcastPlaybackState.Playing);
+                    itShouldSetTheStateTo(VoiceBroadcastPlaybackState.Stopped);
+                },
+            );
 
-                    it("should play the next chunk", () => {
+            describe(
-                        expect(chunk3Playback.play).toHaveBeenCalled();
+                "and receiving a stop info event with last_chunk_sequence = 3 and " +
+                    "the playback of the last available chunk ends",
+                () => {
+                    beforeEach(() => {
+                        const stoppedEvent = mkVoiceBroadcastInfoStateEvent(
+                            roomId,
+                            VoiceBroadcastInfoState.Stopped,
+                            client.getSafeUserId(),
+                            client.deviceId!,
+                            infoEvent,
+                            3,
+                        );
+                        room.addLiveEvents([stoppedEvent]);
+                        room.relations.aggregateChildEvent(stoppedEvent);
+                        chunk2Playback.emit(PlaybackState.Stopped);
                     });
-                });
+
-            });
+                    itShouldSetTheStateTo(VoiceBroadcastPlaybackState.Buffering);
+
+                    describe("and the next chunk arrives", () => {
+                        beforeEach(() => {
+                            room.addLiveEvents([chunk3Event]);
+                            room.relations.aggregateChildEvent(chunk3Event);
+                        });
+
+                        itShouldSetTheStateTo(VoiceBroadcastPlaybackState.Playing);
+
+                        it("should play the next chunk", () => {
+                            expect(chunk3Playback.play).toHaveBeenCalled();
+                        });
+                    });
+                },
+            );
 
             describe("and the info event is deleted", () => {
                 beforeEach(() => {

test/voice-broadcast/utils/VoiceBroadcastChunkEvents-test.ts

@@ -62,6 +62,10 @@ describe("VoiceBroadcastChunkEvents", () => {
             ]);
         });
 
+        it("getNumberOfEvents should return 4", () => {
+            expect(chunkEvents.getNumberOfEvents()).toBe(4);
+        });
+
         it("getLength should return the total length of all chunks", () => {
             expect(chunkEvents.getLength()).toBe(3259);
         });
@@ -110,6 +114,7 @@ describe("VoiceBroadcastChunkEvents", () => {
                     eventSeq3Time2T,
                     eventSeq4Time1,
                 ]);
+                expect(chunkEvents.getNumberOfEvents()).toBe(4);
             });
         });
     });
@@ -129,6 +134,17 @@ describe("VoiceBroadcastChunkEvents", () => {
                 eventSeqUTime3,
                 eventSeq2Time4Dup,
             ]);
+            expect(chunkEvents.getNumberOfEvents()).toBe(5);
+        });
+
+        describe("getSequenceForEvent", () => {
+            it("should return the sequence if provided by the event", () => {
+                expect(chunkEvents.getSequenceForEvent(eventSeq3Time2)).toBe(3);
+            });
+
+            it("should return the index if no sequence provided by event", () => {
+                expect(chunkEvents.getSequenceForEvent(eventSeqUTime3)).toBe(4);
+            });
         });
     });
 });

test/voice-broadcast/utils/test-utils.ts

@@ -24,12 +24,16 @@ import {
 } from "../../../src/voice-broadcast";
 import { mkEvent } from "../../test-utils";
 
+// timestamp incremented on each call to prevent duplicate timestamp
+let timestamp = new Date().getTime();
+
 export const mkVoiceBroadcastInfoStateEvent = (
     roomId: Optional<string>,
     state: Optional<VoiceBroadcastInfoState>,
     senderId: Optional<string>,
     senderDeviceId: Optional<string>,
     startedInfoEvent?: MatrixEvent,
+    lastChunkSequence?: number,
 ): MatrixEvent => {
     const relationContent = {};
 
@@ -40,6 +44,8 @@ export const mkVoiceBroadcastInfoStateEvent = (
         };
     }
 
+    const lastChunkSequenceContent = lastChunkSequence ? { last_chunk_sequence: lastChunkSequence } : {};
+
     return mkEvent({
         event: true,
         // @ts-ignore allow everything here for edge test cases
@@ -53,7 +59,9 @@ export const mkVoiceBroadcastInfoStateEvent = (
             state,
             device_id: senderDeviceId,
             ...relationContent,
+            ...lastChunkSequenceContent,
         },
+        ts: timestamp++,
     });
 };