From: Marcel van der Veldt Date: Fri, 20 Feb 2026 14:14:05 +0000 (+0100) Subject: Small fixes to grouped airplay playback and late joining X-Git-Url: https://git.kitaultman.com/?a=commitdiff_plain;h=53df33761f9ec3b180ccc4e2ccd83124a0bb5034;p=music-assistant-server.git Small fixes to grouped airplay playback and late joining --- diff --git a/music_assistant/providers/airplay/stream_session.py b/music_assistant/providers/airplay/stream_session.py index c5a2f99b..0ac9625d 100644 --- a/music_assistant/providers/airplay/stream_session.py +++ b/music_assistant/providers/airplay/stream_session.py @@ -153,8 +153,19 @@ class AirPlayStreamSession: return async with self._lock: - # Get all buffered chunks to send - buffered_chunks = list(self._chunk_buffer) + # Get buffered chunks to send, but limit to ~5 seconds to avoid + # blocking real-time streaming to other players (causes packet loss) + max_late_join_buffer_seconds = 5.0 + all_buffered = list(self._chunk_buffer) + + # Filter to only include chunks within the time limit + if all_buffered: + min_position = self.seconds_streamed - max_late_join_buffer_seconds + buffered_chunks = [ + (chunk, pos) for chunk, pos in all_buffered if pos >= min_position + ] + else: + buffered_chunks = [] if buffered_chunks: # Calculate how much buffer we're sending @@ -209,7 +220,10 @@ class AirPlayStreamSession: if not self.sync_clients: break - await self._write_chunk_to_all_players(chunk) + has_running_clients = await self._write_chunk_to_all_players(chunk) + if not has_running_clients: + self.prov.logger.debug("No running clients remaining, stopping audio streamer") + break self.seconds_streamed += len(chunk) / pcm_sample_size finally: if not watchdog_task.done(): @@ -237,7 +251,9 @@ class AirPlayStreamSession: silence_duration = 0.1 silence_bytes = int(pcm_sample_size * silence_duration) silence_chunk = bytes(silence_bytes) - await self._write_chunk_to_all_players(silence_chunk) + has_running_clients = await self._write_chunk_to_all_players(silence_chunk) + if not has_running_clients: + break self.seconds_streamed += silence_duration silence_inserted += silence_duration await asyncio.sleep(0.05) @@ -248,12 +264,15 @@ class AirPlayStreamSession: silence_inserted, ) - async def _write_chunk_to_all_players(self, chunk: bytes) -> None: - """Write a chunk to all connected players.""" + async def _write_chunk_to_all_players(self, chunk: bytes) -> bool: + """Write a chunk to all connected players. + + :return: True if there are still running clients, False otherwise. + """ async with self._lock: sync_clients = [x for x in self.sync_clients if x.stream and x.stream.running] if not sync_clients: - return + return False # Add chunk to ring buffer for late joiners (before seconds_streamed is updated) chunk_position = self.seconds_streamed @@ -287,6 +306,10 @@ class AirPlayStreamSession: for player in players_to_remove: self.mass.create_task(self.remove_client(player)) + # Return False if all clients were removed (or scheduled for removal) + remaining_clients = len(sync_clients) - len(players_to_remove) + return remaining_clients > 0 + async def _write_chunk_to_player(self, airplay_player: AirPlayPlayer, chunk: bytes) -> None: """Write audio chunk to a player's ffmpeg process.""" player_id = airplay_player.player_id