more fixes

music_assistant/server/controllers/streams.py

@@ -64,11 +64,12 @@
 DEFAULT_STREAM_HEADERS = {
     "transferMode.dlna.org": "Streaming",
     "contentFeatures.dlna.org": "DLNA.ORG_OP=00;DLNA.ORG_CI=0;DLNA.ORG_FLAGS=0d500000000000000000000000000000",  # noqa: E501
-    "Cache-Control": "no-cache",
+    "Cache-Control": "no-cache,must-revalidate",
+    "Pragma": "no-cache",
     "Connection": "close",
     "Accept-Ranges": "none",
-    "icy-name": "Music Assistant",
-    "icy-pub": "0",
+    "Icy-Name": "Music Assistant",
+    "Icy-Url": "https://music-assistant.io",
 }
 FLOW_DEFAULT_SAMPLE_RATE = 48000
 FLOW_DEFAULT_BIT_DEPTH = 24
@@ -122,7 +123,7 @@ def finished(self) -> bool:
     @property
     def pending(self) -> bool:
         """Return if this Job is pending start."""
-        return not self.finished and not self._audio_task
+        return not self.finished and not self.running
 
     @property
     def running(self) -> bool:
@@ -137,9 +138,7 @@ def start(self) -> None:
 
     def stop(self) -> None:
         """Stop running this job."""
-        if self._audio_task and self._audio_task.done():
-            return
-        if self._audio_task:
+        if self._audio_task and not self._audio_task.done():
             self._audio_task.cancel()
         self._finished = True
 
@@ -239,7 +238,7 @@ async def subscribe(
             self._subscribed_players.pop(player_id, None)
             self.logger.debug("Unsubscribed client %s", player_id)
             # check if this was the last subscriber and we should cancel
-            await asyncio.sleep(2)
+            await asyncio.sleep(5)
             if len(self._subscribed_players) == 0 and not self.finished:
                 self.logger.debug("Cleaning up, all clients disappeared...")
                 self.stop()
@@ -250,7 +249,7 @@ async def _stream_job_runner(self) -> None:
         retries = 50
         while retries:
             retries -= 1
-            await asyncio.sleep(0.2)
+            await asyncio.sleep(0.1)
             if len(self._subscribed_players) != len(self.expected_players):
                 continue
             await asyncio.sleep(0.2)

music_assistant/server/providers/sonos/__init__.py

@@ -36,6 +36,7 @@
 from music_assistant.constants import CONF_CROSSFADE, VERBOSE_LOG_LEVEL
 from music_assistant.server.helpers.didl_lite import create_didl_metadata
 from music_assistant.server.models.player_provider import PlayerProvider
+from music_assistant.server.providers.ugp import UGP_PREFIX
 
 from .player import SonosPlayer
 
@@ -341,11 +342,6 @@ async def play_media(
         queue_item: QueueItem,
     ) -> None:
         """Handle PLAY MEDIA on given player."""
-        url = self.mass.streams.resolve_stream_url(
-            player_id,
-            queue_item=queue_item,
-            output_codec=ContentType.FLAC,
-        )
         sonos_player = self.sonosplayers[player_id]
         mass_player = self.mass.players.get(player_id)
         if sonos_player.sync_coordinator:
@@ -355,10 +351,17 @@ async def play_media(
                 "accept play_media command, it is synced to another player."
             )
             raise PlayerCommandFailed(msg)
-        await self.mass.create_task(
+
+        is_flow_stream = queue_item.queue_item_id == "flow" or queue_item.queue_id.startswith(
+            UGP_PREFIX
+        )
+        url = self.mass.streams.resolve_stream_url(
+            player_id, queue_item=queue_item, output_codec=ContentType.FLAC
+        )
+        self.mass.create_task(
             sonos_player.soco.play_uri,
             url,
-            meta=create_didl_metadata(self.mass, url, queue_item),
+            meta=create_didl_metadata(self.mass, url, None if is_flow_stream else queue_item),
         )
 
     async def enqueue_next_queue_item(self, player_id: str, queue_item: QueueItem) -> None:

music_assistant/server/providers/ugp/__init__.py

@@ -20,14 +20,20 @@
 )
 from music_assistant.common.models.enums import (
     ConfigEntryType,
+    ContentType,
     PlayerFeature,
     PlayerState,
     PlayerType,
     ProviderFeature,
 )
+from music_assistant.common.models.media_items import AudioFormat
 from music_assistant.common.models.player import DeviceInfo, Player
 from music_assistant.common.models.queue_item import QueueItem
 from music_assistant.constants import CONF_CROSSFADE, CONF_GROUP_MEMBERS, SYNCGROUP_PREFIX
+from music_assistant.server.controllers.streams import (
+    FLOW_DEFAULT_BIT_DEPTH,
+    FLOW_DEFAULT_SAMPLE_RATE,
+)
 from music_assistant.server.models.player_provider import PlayerProvider
 
 if TYPE_CHECKING:
@@ -138,6 +144,8 @@ async def cmd_stop(self, player_id: str) -> None:
                 if member.state == PlayerState.IDLE:
                     continue
                 tg.create_task(self.mass.players.cmd_stop(member.player_id))
+        if existing := self.mass.streams.stream_jobs.pop(player_id, None):
+            existing.stop()
 
     async def cmd_play(self, player_id: str) -> None:
         """Send PLAY command to given player."""
@@ -166,11 +174,22 @@ async def play_media(
         await self.cmd_power(player_id, True)
         group_player = self.mass.players.get(player_id)
 
+        await self.cmd_stop(player_id)
+
         # create a multi-client stream job - all (direct) child's of this UGP group
         # will subscribe to this multi client queue stream
+        pcm_format = AudioFormat(
+            content_type=ContentType.from_bit_depth(FLOW_DEFAULT_BIT_DEPTH),
+            sample_rate=FLOW_DEFAULT_SAMPLE_RATE,
+            bit_depth=FLOW_DEFAULT_BIT_DEPTH,
+        )
+        queue = self.mass.player_queues.get(player_id)
         stream_job = self.mass.streams.create_stream_job(
-            player_id,
-            start_queue_item=queue_item,
+            queue.queue_id,
+            pcm_audio_source=self.mass.streams.get_flow_stream(
+                queue=queue, start_queue_item=queue_item, pcm_format=pcm_format
+            ),
+            pcm_format=pcm_format,
         )
         # create a fake queue item to forward to downstream play_media commands
         ugp_queue_item = QueueItem(