music-assistant-server

1"""
2Music Assistant Snapcast source stream.
3
4This module implements a Music Assistant-managed Snapcast stream that is exposed to the
5Snapcast server as a TCP source. The stream is produced by running an FFmpeg pipeline
6which pulls audio from Music Assistant and pushes it to the Snapcast source URI.
7
8Optionally, a Unix socket server can be started to provide a control channel for a
9Snapcast control script (used by the built-in Snapcast server integration).
10"""
11
12from __future__ import annotations
13
14import asyncio
15import random
16import time
17import urllib.parse
18from contextlib import suppress
19from typing import TYPE_CHECKING, cast
20
21from music_assistant_models.enums import ContentType
22from music_assistant_models.media_items import AudioFormat
23
24from music_assistant.controllers.streams.audio_processing import (
25    AudioOutputPlan,
26    get_media_session_id,
27)
28from music_assistant.helpers.ffmpeg import FFMpeg
29from music_assistant.providers.snapcast.socket_server import SnapcastSocketServer
30
31from .constants import (
32    CONTROL_SOCKET_PATH_TEMPLATE,
33    DEFAULT_SNAPCAST_FORMAT,
34)
35
36if TYPE_CHECKING:
37    from music_assistant_models.player import PlayerMedia
38
39    from music_assistant.helpers.dsp import ComplexFilter
40
41    from .provider import SnapCastProvider
42    from .snap_cntrl_proto import SnapstreamProto
43
44
45class SnapcastMAStream:
46    """
47    A Music Assistant-managed Snapcast stream.
48
49    The stream lifecycle is:
50    - setup: ensure required server resources exist (Snapcast source, optional socket server)
51    - start_stream: start the FFmpeg streaming task
52    - request_stop_stream / wait_for_stopped: stop streaming and await termination
53    - destroy: stop streaming, remove Snapcast source, and stop ancillary services
54
55    If `cntrl_queue_id` is provided, a Unix socket server is started to allow a Snapcast
56    control script to communicate with Music Assistant.
57    """
58
59    def __init__(
60        self,
61        provider: SnapCastProvider,
62        media: PlayerMedia,
63        stream_name: str,
64        source_id: str | None = None,
65        filter_settings_owner: str | None = None,
66        use_cntrl_script: bool = False,
67        destroy_on_stop: bool = False,
68    ) -> None:
69        """
70        Initialize the stream.
71
72        Args:
73            provider: The Snapcast provider instance.
74            media: The media item to stream.
75            stream_name: Name used to register the stream on the Snapcast server.
76            cntrl_queue_id: If set, enables the control socket server used by the control script.
77            filter_settings_owner: Player/entity id used to fetch DSP/filter parameters.
78            destroy_on_stop: If true, delete this MA stream once streaming stops.
79        """
80        self.media = media
81        self.stream_name = stream_name
82        self.snap_stream: SnapstreamProto | None = None
83
84        self._provider = provider
85        self._logger = provider.logger
86        self._mass = provider.mass
87        self._source_id = source_id
88        self._use_cntrl_script = use_cntrl_script
89        self._cntrl_queue_id = source_id if use_cntrl_script else None
90        self._filter_settings_owner = filter_settings_owner
91        self._destroy_on_stop = destroy_on_stop
92
93        self._lifecycle_lock = asyncio.Lock()
94        self._destroyed = False
95        self._setup_done = False
96        self._is_streaming = False
97        self._restart_requested: bool = False
98        self._stop_requested: bool = False
99        self._streaming_started_at: float | None = None
100        self._output_plan: AudioOutputPlan | None = None
101
102        self._socket_server: SnapcastSocketServer | None = None
103        self._socket_path: str | None = None
104        self._streamer_task: asyncio.Task[None] | None = None
105        self._stop_streamer_evt = asyncio.Event()
106        self._streamer_started_evt = asyncio.Event()
107        self._stop_timer: asyncio.Handle | None = None
108        self._stop_timer_started_at: float | None = None
109        self._filter_settings: list[str | ComplexFilter] | None = None
110
111    @property
112    def source_id(self) -> str | None:
113        """Return the source id this stream was created for."""
114        return self._source_id
115
116    @property
117    def stream_id(self) -> str | None:
118        """Return the Snapcast stream identifier, if registered."""
119        if self.snap_stream:
120            return self.snap_stream.identifier
121        return None
122
123    @property
124    def is_streaming(self) -> bool:
125        """Return True if the FFmpeg streaming task is currently running."""
126        return self._is_streaming
127
128    @property
129    def playback_started_at(self) -> float | None:
130        """
131        Return when the playback started at the clients.
132
133        return The (UTC) timestamp when the playback was started on the client
134        or None if not started yet or not streaming.
135        """
136        if self._streaming_started_at is None:
137            return None
138        if self._provider._use_builtin_server:
139            buffer_ms = self._provider._snapcast_server_buffer_size
140            if time.time() - self._streaming_started_at < buffer_ms / 1000.0:
141                return None
142            return self._streaming_started_at + buffer_ms / 1000.0
143        return self._streaming_started_at
144
145    async def setup(self) -> None:
146        """
147        Prepare the Snapcast stream resources.
148
149        Ensures a Snapcast source exists on the server. If `cntrl_queue_id` is set,
150        also starts the Unix socket server used by the control script.
151        """
152        async with self._lifecycle_lock:
153            if self._destroyed:
154                raise RuntimeError("Session is destroyed")
155            if self._setup_done:
156                return
157            if self._provider._snapserver is None:
158                raise RuntimeError("Snapserver needs to be setup first")
159
160            if self._cntrl_queue_id:
161                await self._start_socket_server()
162
163            await self._register_tcp_server_source()
164            self._setup_done = True
165
166    async def destroy(self) -> None:
167        """
168        Stop streaming and tear down all resources.
169
170        This stops the streamer task (if running), removes the Snapcast source,
171        and stops the optional control socket server.
172        """
173        async with self._lifecycle_lock:
174            if self._destroyed:
175                return
176            self._destroyed = True
177
178        self.request_stop_stream()
179        await self.wait_for_stopped()
180        await self._remove_snap_source()
181        await self._stop_socket_server()
182
183    async def start_stream(self, allow_restart: bool = False) -> None:
184        """
185        Start streaming the configured media to the Snapcast source.
186
187        Raises:
188            RuntimeError: If the streamer task is already running.
189        """
190        await self.setup()
191        async with self._lifecycle_lock:
192            if self._streamer_task and not self._streamer_task.done():
193                if not allow_restart:
194                    raise RuntimeError("streamer already running")
195                if self._stop_requested or self._stop_streamer_evt.is_set():
196                    # stop in flight; _on_streamer_done will start the fresh run
197                    self._restart_requested = True
198                else:
199                    self._restart_if_running()
200                return
201
202            self._stop_requested = False
203            self._restart_requested = False
204            self._stop_streamer_evt.clear()
205            self._streamer_started_evt.clear()
206            self._streamer_task = self._mass.create_task(self._streamer_task_impl())
207            self._streamer_task.add_done_callback(self._on_streamer_done)
208
209    async def wait_for_started(self, timeout_sec: float | None = None) -> None:
210        """
211        Wait until the streamer task signals it has started.
212
213        Args:
214            timeout_sec: Optional timeout in seconds.
215        """
216        try:
217            await asyncio.wait_for(self._streamer_started_evt.wait(), timeout_sec)
218        except TimeoutError:
219            self._logger.warning(
220                "Timeout waiting for stream %s to start; Canceling...",
221                self.stream_name,
222            )
223
224    def update_media(self, media: PlayerMedia) -> None:
225        """Update the media to play and restart the stream if required."""
226        if media != self.media:
227            self.media = media
228            self._restart_if_running()
229
230    def update_filter_settings(self, from_player: str | None = None) -> None:
231        """Update the filter setting."""
232        take_from = from_player or self._filter_settings_owner
233        if not take_from:
234            raise RuntimeError("No player provided to read filter settings from.")
235        output_format = self._get_transport_format()
236        self._output_plan = self._mass.streams.audio.get_player_output_plan(
237            take_from,
238            DEFAULT_SNAPCAST_FORMAT,
239            output_format,
240            handoff_format=DEFAULT_SNAPCAST_FORMAT,
241        )
242        self._register_output_plan()
243        new_settings = self._output_plan.filter_params
244        if from_player:
245            self._filter_settings_owner = from_player
246        if new_settings != self._filter_settings:
247            self._restart_if_running()
248
249    def request_stop_stream(self) -> None:
250        """
251        Request the streamer task to stop.
252
253        This is cooperative: the streamer task will stop when it observes the stop event.
254        Any pending inactivity stop timer is canceled.
255        """
256        self._stop_requested = True
257        self._restart_requested = False  # explicit stop cancels any pending restart
258        self._stop_streamer_evt.set()
259
260        self._stop_timer_started_at = None
261        if self._stop_timer:
262            self._stop_timer.cancel()
263
264    def set_in_use(self, in_use: bool) -> None:
265        """
266        Mark the stream as in-use or idle.
267
268        When marked idle, a delayed stop is scheduled. When marked in-use, any pending
269        delayed stop is canceled.
270        """
271        if in_use:
272            self._stop_timer_started_at = None
273            if self._stop_timer:
274                self._stop_timer.cancel()
275        elif self._stop_timer_started_at is None and not self._stop_requested:
276            self._stop_timer_started_at = self._mass.loop.time()
277            self._stop_timer = self._mass.loop.call_later(3.0, self.request_stop_stream)
278
279    async def wait_for_stopped(self, timeout_sec: float | None = None) -> None:
280        """
281        Wait for the streamer task to finish.
282
283        If the task does not finish within the timeout, it is canceled and awaited.
284
285        Args:
286            timeout_sec: Optional timeout in seconds.
287        """
288        curr_task = self._streamer_task
289        if not curr_task:
290            return
291        try:
292            await asyncio.wait_for(curr_task, timeout_sec)
293        except asyncio.CancelledError:
294            self._logger.warning("Streamer task got canceled")
295        except TimeoutError:
296            self._logger.warning(
297                "Timeout waiting for stream %s to finish; Canceling...",
298                self.stream_name,
299            )
300            curr_task.cancel()
301            await asyncio.gather(curr_task, return_exceptions=True)
302
303    async def _streamer_task_impl(self) -> None:
304        """
305        Streamer task implementation.
306
307        Runs FFmpeg to push audio to the Snapcast TCP source until FFmpeg exits or a stop
308        request is received. After exit, waits briefly for the Snapcast stream to report
309        an idle state.
310        """
311        stream_path = self._snap_get_stream_path()
312        if stream_path is None:
313            raise RuntimeError("The path to stream to is not set")
314
315        self._logger.debug("Start streaming to %s", stream_path)
316        self._stop_streamer_evt.clear()
317        self._streamer_started_evt.clear()
318        if self._filter_settings_owner:
319            output_format = self._get_transport_format()
320            self._output_plan = self._mass.streams.audio.get_player_output_plan(
321                self._filter_settings_owner,
322                DEFAULT_SNAPCAST_FORMAT,
323                output_format,
324                handoff_format=DEFAULT_SNAPCAST_FORMAT,
325            )
326            self._register_output_plan()
327            self._filter_settings = self._output_plan.filter_params
328        audio_source = self._mass.streams.get_stream(
329            self.media,
330            DEFAULT_SNAPCAST_FORMAT,
331            self._filter_settings_owner,
332            use_flow_stream_buffering=True,
333        )
334        try:
335            async with FFMpeg(
336                audio_input=audio_source,
337                input_format=DEFAULT_SNAPCAST_FORMAT,
338                output_format=DEFAULT_SNAPCAST_FORMAT,
339                filter_params=self._filter_settings or [],
340                audio_output=stream_path,
341                extra_input_args=["-y", "-re"],
342            ) as ffmpeg_proc:
343                wait_ffmpeg = self._mass.create_task(ffmpeg_proc.wait())
344                wait_stop = self._mass.create_task(self._stop_streamer_evt.wait())
345                self._streaming_started_at = time.time()
346                self._streamer_started_evt.set()
347                self._is_streaming = True
348
349                done, pending = await asyncio.wait(
350                    {wait_ffmpeg, wait_stop},
351                    return_when=asyncio.FIRST_COMPLETED,
352                )
353
354                if wait_stop in done and wait_ffmpeg not in done:
355                    self._logger.debug("Stopping stream %s requested.", self.stream_name)
356                    wait_ffmpeg.cancel()
357                    await asyncio.gather(wait_ffmpeg, return_exceptions=True)
358                    return
359
360                await wait_ffmpeg
361                for t in pending:
362                    t.cancel()
363                await asyncio.gather(*pending, return_exceptions=True)
364        except asyncio.CancelledError:
365            self._logger.debug("Snapcast stream %s cancelled", self.stream_name)
366            raise
367        except Exception as err:
368            self._logger.error("Snapcast stream %s error: %s", self.stream_name, err, exc_info=err)
369            raise
370        finally:
371            self._is_streaming = False
372            self._logger.debug("Finished streaming to %s", stream_path)
373            await self._wait_stream_idle()
374
375    async def _wait_stream_idle(self) -> None:
376        """Wait for the Snapcast stream to become idle after streaming ends."""
377        try:
378
379            async def wait_until_idle() -> None:
380                while True:
381                    stream_is_idle = False
382                    with suppress(KeyError):
383                        snap_stream = self._provider._snapserver.stream(self.stream_name)
384                        stream_is_idle = snap_stream.status == "idle"
385                    if self._mass.closing or stream_is_idle:
386                        break
387                    await asyncio.sleep(0.25)
388
389            await asyncio.wait_for(wait_until_idle(), timeout=10.0)
390        except TimeoutError:
391            self._logger.warning(
392                "Timeout waiting for stream %s to become idle",
393                self.stream_name,
394            )
395        finally:
396            self._streaming_started_at = None
397
398    def _on_streamer_done(self, t: asyncio.Task[None]) -> None:
399        """Handle streamer task completion and optional cleanup."""
400        restart = False
401        try:
402            t.result()
403        except asyncio.CancelledError:
404            self._logger.debug("Streamer task cancelled: %s", self.stream_name)
405        except Exception:
406            self._logger.exception("Streamer task failed")
407        finally:
408            restart = self._restart_requested and not self._destroyed
409
410            if self._streamer_task is t:
411                self._streamer_task = None
412
413            # reset per-run state
414            self._restart_requested = False
415            self._stop_requested = False
416            self._stop_streamer_evt.clear()
417            self._streamer_started_evt.clear()
418
419        if restart:
420            self._mass.create_task(self._restart_stream_locked())
421        elif self._destroy_on_stop:
422            self._mass.create_task(self._provider.delete_ma_stream(self.stream_name))
423
424    def _restart_if_running(self) -> None:
425        """Request a running stream to restart."""
426        t = self._streamer_task
427        if not t or t.done():
428            return
429
430        if self._stop_requested or self._stop_streamer_evt.is_set():
431            return
432
433        self._restart_requested = True
434        self._stop_requested = True
435        self._stop_streamer_evt.set()
436
437        self._stop_timer_started_at = None
438        if self._stop_timer:
439            self._stop_timer.cancel()
440
441    async def _restart_stream_locked(self) -> None:
442        """Restart the streamer under the lifecycle lock."""
443        async with self._lifecycle_lock:
444            if self._destroyed:
445                return
446            if self._streamer_task and not self._streamer_task.done():
447                return
448
449            # reset state and start a fresh run
450            self._stop_requested = False
451            self._restart_requested = False
452            self._stop_streamer_evt.clear()
453            self._streamer_started_evt.clear()
454
455            self._streamer_task = self._mass.create_task(self._streamer_task_impl())
456            self._streamer_task.add_done_callback(self._on_streamer_done)
457
458    def _get_transport_format(self) -> AudioFormat:
459        """Return the format Snapserver sends to its clients."""
460        if self._provider._use_builtin_server:
461            codec_name = str(self._provider._snapcast_server_transport_codec)
462        else:
463            stream_data = self.snap_stream._stream if self.snap_stream else {}
464            uri_data = stream_data.get("uri", {})
465            query_data = uri_data.get("query", {}) if isinstance(uri_data, dict) else {}
466            codec_name = str(query_data.get("codec", "") if isinstance(query_data, dict) else "")
467        codec_name = codec_name.partition(":")[0].lower()
468        content_type, codec_type = {
469            "flac": (ContentType.FLAC, ContentType.FLAC),
470            "ogg": (ContentType.OGG, ContentType.VORBIS),
471            "opus": (ContentType.OPUS, ContentType.OPUS),
472            "pcm": (ContentType.PCM_S16LE, ContentType.PCM_S16LE),
473        }.get(codec_name, (ContentType.UNKNOWN, ContentType.UNKNOWN))
474        return AudioFormat(
475            content_type=content_type,
476            codec_type=codec_type,
477            sample_rate=DEFAULT_SNAPCAST_FORMAT.sample_rate,
478            bit_depth=DEFAULT_SNAPCAST_FORMAT.bit_depth,
479            channels=DEFAULT_SNAPCAST_FORMAT.channels,
480        )
481
482    def _register_output_plan(self) -> None:
483        """Register the shared Snapcast path for every connected group member."""
484        queue_id = self.media.source_id
485        session_id = get_media_session_id(self.media)
486        if self._output_plan is None or queue_id is None or session_id is None:
487            return
488        player_ids = set(self._output_plan.output_details.player_ids)
489        if self.snap_stream:
490            for group in self._provider._snapserver.groups:
491                if group.stream != self.snap_stream.identifier:
492                    continue
493                for client_id in group.clients:
494                    if player_id := self._provider._get_ma_id(client_id):
495                        player_ids.add(player_id)
496        for player_id in player_ids:
497            self._mass.streams.audio_processing.update_output(
498                player_id,
499                self._output_plan,
500                queue_id=queue_id,
501                session_id=session_id,
502            )
503
504    def _find_local_stream_by_name(self, name: str) -> SnapstreamProto | None:
505        """
506        Look up a snapserver stream by its name (not id) in the local cache.
507
508        :param name: Stream name to look up.
509        :return: The matching SnapstreamProto, or None if not found.
510        """
511        for s in self._provider._snapserver.streams:
512            if getattr(s, "name", None) == name:
513                return s
514        return None
515
516    @staticmethod
517    def _is_name_collision_error(result: object) -> bool:
518        """
519        Detect snapserver's 'Stream with name X already exists' error.
520
521        :param result: Result returned by snapserver.stream_add_stream.
522        :return: True if the result indicates a stream-name collision.
523        """
524        if not isinstance(result, dict):
525            return False
526        data = result.get("data", "")
527        return (
528            isinstance(data, str)
529            and data.startswith("Stream with name")
530            and "already exists" in data
531        )
532
533    def _pick_port_avoiding(self, used: set[int]) -> int | None:
534        """
535        Pick a random TCP source port within the unchanged upstream range.
536
537        Avoids ports already tried in the current retry loop.
538
539        :param used: Set of ports that should not be returned again.
540        :return: A port in [4953, 5153] not in `used`, or None if none could be
541            found within 20 attempts (extremely rare; the range has 201 values).
542        """
543        for _ in range(20):
544            port = random.randint(4953, 4953 + 200)
545            if port not in used:
546                return port
547        return None
548
549    async def _register_tcp_server_source(self) -> None:
550        """Create a Snapcast TCP source for this stream (or reuse an existing one)."""
551        # This runs under the per-stream `_lifecycle_lock` (acquired in setup()), not the
552        # provider-wide `_snapcast_ma_streams_lock`. Adoption is safe because each MA-managed
553        # stream has at most one outstanding setup() call at a time.
554
555        # prefer to reuse existing stream if possible
556        if self.snap_stream:
557            return
558
559        # The control script is used only for music streams in the builtin server
560        extra_args = ""
561        if (cntrl_queue_id := self._cntrl_queue_id) is not None:
562            # Create socket server for control script communication
563            socket_path = self._socket_path
564            if socket_path is None:
565                raise RuntimeError("socket_path needs to be set if cntrl_queue_id is set")
566            extra_args = (
567                f"&controlscript={urllib.parse.quote_plus('control.py')}"
568                f"&controlscriptparams=--queueid={urllib.parse.quote_plus(cntrl_queue_id)}%20"
569                f"--socket={urllib.parse.quote_plus(socket_path)}%20"
570                f"--streamserver-ip={self._mass.streams.publish_ip}%20"
571                f"--streamserver-port={self._mass.streams.publish_port}"
572            )
573
574        tried_ports: set[int] = set()
575        attempts = 50
576        loop_succeeded = False
577        try:
578            while attempts:
579                attempts -= 1
580                port = self._pick_port_avoiding(tried_ports)
581                if port is None:
582                    break
583                tried_ports.add(port)
584                result = await self._provider._snapserver.stream_add_stream(
585                    # NOTE: setting the sampleformat to something else
586                    # (like 24 bits bit depth) does not seem to work at all!
587                    f"tcp://0.0.0.0:{port}?sampleformat=48000:16:2"
588                    f"&idle_threshold={self._provider._snapcast_stream_idle_threshold}"
589                    f"{extra_args}&name={self.stream_name}"
590                )
591                if isinstance(result, dict) and "id" in result:
592                    self.snap_stream = self._provider._snapserver.stream(result["id"])
593                    self.snap_stream.set_callback(self._snap_on_stream_update)
594                    loop_succeeded = True
595                    return
596
597                if self._is_name_collision_error(result):
598                    adopted = self._find_local_stream_by_name(self.stream_name)
599                    if adopted is None:
600                        # Local cache may be stale (e.g. after an MA restart). Resync.
601                        status, _ = await self._provider._snapserver.status()
602                        if isinstance(status, dict):
603                            self._provider._snapserver.synchronize(status)
604                        adopted = self._find_local_stream_by_name(self.stream_name)
605                    if adopted is not None:
606                        self._logger.info(
607                            "Adopted orphaned snapserver stream %s (name=%s)",
608                            adopted.identifier,
609                            self.stream_name,
610                        )
611                        self.snap_stream = adopted
612                        self.snap_stream.set_callback(self._snap_on_stream_update)
613                        loop_succeeded = True
614                        return
615                elif isinstance(result, dict) and result.get("code") == -32603:
616                    # Forward-compat hint: snapserver may rephrase the name-collision
617                    # error in a future release; surface mismatches in debug logs.
618                    self._logger.debug(
619                        "snapserver returned internal error not matching "
620                        "name-collision pattern: %s",
621                        result,
622                    )
623
624                # if the port is already taken, the result will be an error
625                self._logger.warning("stream_add_stream failed: %s", result)
626                continue
627        finally:
628            # Invariant: if we leave _register_tcp_server_source without setting
629            # self.snap_stream (retries exhausted OR exception during a retry),
630            # any socket_server we started must be stopped.
631            if not loop_succeeded and self._socket_server:
632                await self._stop_socket_server()
633
634        msg = f"Unable to register snapserver stream {self.stream_name!r} after 50 attempts"
635        raise RuntimeError(msg)
636
637    async def _remove_snap_source(self) -> None:
638        """Remove the Snapcast source created for this stream and detach groups."""
639        if self._mass.closing or self.snap_stream is None:
640            return
641
642        for snap_group in self._provider._snapserver.groups:
643            if snap_group.stream != self.snap_stream.identifier:
644                continue
645            self._logger.debug(f"Set stream of group {snap_group.name} to default.")
646            await snap_group.set_stream("default")
647
648        with suppress(KeyError, AttributeError):
649            snap_stream = self._provider._snapserver.stream(self.stream_name)
650            await self._provider._snapserver.stream_remove_stream(snap_stream.identifier)
651
652        if self._socket_server:
653            await self._stop_socket_server()
654        self._snap_on_stream_update()
655
656        return
657
658    def _snap_get_stream_path(self) -> str | None:
659        """Return the Snapcast TCP URI to stream to."""
660        if self.snap_stream is None:
661            return None
662
663        uri = self.snap_stream._stream.get("uri", {})
664        uri_host = uri.get("host", "")
665        stream_path = self.snap_stream.path or f"tcp://{uri_host}"
666        return stream_path.replace("0.0.0.0", self._provider._snapcast_server_host)
667
668    def _snap_on_stream_update(self, stream: SnapstreamProto | None = None) -> None:
669        """Handle Snapcast stream updates and trigger group member refresh."""
670        if self.snap_stream is None:
671            return
672
673        for snap_group in self._provider._snapserver.groups:
674            if snap_group.stream != self.snap_stream.identifier:
675                continue
676            self._provider.poke_group_members(snap_group)
677        self._register_output_plan()
678
679    async def _start_socket_server(self) -> str:
680        """
681        Get or create a socket server for the given queue.
682
683        :return: The path to the Unix socket.
684        """
685        if self._socket_server:
686            return self._socket_server.socket_path
687
688        if self._cntrl_queue_id is None:
689            raise RuntimeError("Socket server require _cntrl_queue_id to be set")
690
691        socket_path = CONTROL_SOCKET_PATH_TEMPLATE.format(queue_id=self._cntrl_queue_id)
692        socket_server = SnapcastSocketServer(
693            mass=self._mass,
694            queue_id=self._cntrl_queue_id,
695            socket_path=socket_path,
696            streamserver_ip=str(self._mass.streams.publish_ip),
697            streamserver_port=cast("int", self._mass.streams.publish_port),
698        )
699        await socket_server.start()
700        self._socket_server = socket_server
701        self._socket_path = socket_path
702        self._logger.debug(
703            "Created socket server for queue %s at %s", self._cntrl_queue_id, socket_path
704        )
705        return socket_path
706
707    async def _stop_socket_server(self) -> None:
708        """Stop and remove the socket server for the given queue."""
709        if not self._socket_server:
710            return
711
712        await self._socket_server.stop()
713        self._socket_server = None
714        self._logger.debug("Stopped socket server for queue %s", self._cntrl_queue_id)
715