try to fix things

DA-344 · DA-344 · commit fe8e6028cb79 · 2025-08-30T19:38:53.000+02:00
diff --git a/discord/opus.py b/discord/opus.py
@@ -42,6 +42,8 @@
 from .sinks import RawData
 
 if TYPE_CHECKING:
+    from discord.voice.client import VoiceClient
+
     T = TypeVar("T")
     APPLICATION_CTL = Literal["audio", "voip", "lowdelay"]
     BAND_CTL = Literal["narrow", "medium", "wide", "superwide", "full"]
@@ -346,9 +348,9 @@ class OpusError(DiscordException):
         The error code returned.
     """
 
-    def __init__(self, code: int):
+    def __init__(self, code: int = 0, message: str | None = None):
         self.code: int = code
-        msg = _lib.opus_strerror(self.code).decode("utf-8")
+        msg = message or _lib.opus_strerror(self.code).decode("utf-8")
         _log.info('"%s" has happened', msg)
         super().__init__(msg)
 
@@ -523,7 +525,7 @@ def _get_last_packet_duration(self):
 
     def decode(self, data, *, fec=False):
         if data is None and fec:
-            raise OpusError("Invalid arguments: FEC cannot be used with null data")
+            raise OpusError(message="Invalid arguments: FEC cannot be used with null data")
 
         if data is None:
             frame_size = self._get_last_packet_duration() or self.SAMPLES_PER_FRAME
@@ -545,62 +547,3 @@ def decode(self, data, *, fec=False):
         )
 
         return array.array("h", pcm[: ret * channel_count]).tobytes()
-
-
-class DecodeManager(threading.Thread, _OpusStruct):
-    def __init__(self, client: VoiceRecorderClient):
-        super().__init__(daemon=True, name="DecodeManager")
-
-        self.client: VoiceRecorderClient = client
-        self.decode_queue: list[RawData] = []
-
-        self.decoder: dict[int, Decoder] = {}
-
-        self._end_thread = threading.Event()
-
-    def decode(self, opus_frame: RawData):
-        if not isinstance(opus_frame, RawData):
-            raise TypeError("opus_frame should be a RawData object.")
-        self.decode_queue.append(opus_frame)
-
-    def run(self):
-        while not self._end_thread.is_set():
-            try:
-                data = self.decode_queue.pop(0)
-            except IndexError:
-                time.sleep(0.001)
-                continue
-
-            try:
-                if data.decrypted_data is None:
-                    continue
-                else:
-                    data.decoded_data = self.get_decoder(data.ssrc).decode(
-                        data.decrypted_data
-                    )
-            except OpusError:
-                _log.exception(
-                    "Error occurred while decoding opus frame.", exc_info=True
-                )
-                continue
-
-            self.client.receive_audio(data)
-
-    def stop(self) -> None:
-        while self.decoding:
-            time.sleep(0.1)
-            self.decoder = {}
-            gc.collect()
-            _log.debug("Decoder Process Killed")
-        self._end_thread.set()
-
-    def get_decoder(self, ssrc: int) -> Decoder:
-        d = self.decoder.get(ssrc)
-        if d is not None:
-            return d
-        self.decoder[ssrc] = Decoder()
-        return self.decoder[ssrc]
-
-    @property
-    def decoding(self) -> bool:
-        return bool(self.decode_queue)
diff --git a/discord/sinks/errors.py b/discord/sinks/errors.py
@@ -104,7 +104,7 @@ class FFmpegNotFound(SinkException):
     """
 
 
-class NoUserAdio(SinkException):
+class NoUserAudio(SinkException):
     """Exception thrown when you try to format the audio of a user not saved in a sink.
 
     .. versionadded:: 2.7
diff --git a/discord/sinks/m4a.py b/discord/sinks/m4a.py
@@ -38,7 +38,7 @@
 
 from .core import CREATE_NO_WINDOW, RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, M4ASinkError, MaxProcessesCountReached, NoUserAdio
+from .errors import FFmpegNotFound, M4ASinkError, MaxProcessesCountReached, NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -188,7 +188,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         temp_path = f"{user_id}-{time.time()}-recording.m4a.tmp"
         args = [
diff --git a/discord/sinks/mka.py b/discord/sinks/mka.py
@@ -36,7 +36,7 @@
 
 from .core import CREATE_NO_WINDOW, RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, MaxProcessesCountReached, MKASinkError, NoUserAdio
+from .errors import FFmpegNotFound, MaxProcessesCountReached, MKASinkError, NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -186,7 +186,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         args = [
             executable,
diff --git a/discord/sinks/mkv.py b/discord/sinks/mkv.py
@@ -36,7 +36,7 @@
 
 from .core import RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, MaxProcessesCountReached, MKVSinkError, NoUserAdio
+from .errors import FFmpegNotFound, MaxProcessesCountReached, MKVSinkError, NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -186,7 +186,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         args = [
             executable,
diff --git a/discord/sinks/mp3.py b/discord/sinks/mp3.py
@@ -36,7 +36,7 @@
 
 from .core import CREATE_NO_WINDOW, RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, MaxProcessesCountReached, MP3SinkError, NoUserAdio
+from .errors import FFmpegNotFound, MaxProcessesCountReached, MP3SinkError, NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -186,7 +186,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         args = [
             executable,
diff --git a/discord/sinks/mp4.py b/discord/sinks/mp4.py
@@ -38,7 +38,7 @@
 
 from .core import CREATE_NO_WINDOW, RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, MaxProcessesCountReached, MP4SinkError, NoUserAdio
+from .errors import FFmpegNotFound, MaxProcessesCountReached, MP4SinkError, NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -188,7 +188,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         temp_path = f"{user_id}-{time.time()}-recording.mp4.tmp"
         args = [
diff --git a/discord/sinks/ogg.py b/discord/sinks/ogg.py
@@ -36,7 +36,7 @@
 
 from .core import CREATE_NO_WINDOW, RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import FFmpegNotFound, MaxProcessesCountReached, NoUserAdio, OGGSinkError
+from .errors import FFmpegNotFound, MaxProcessesCountReached, NoUserAudio, OGGSinkError
 
 if TYPE_CHECKING:
     from discord import abc
@@ -186,7 +186,7 @@ def format_user_audio(
             data = self.__audio_data.pop(user_id)
         except KeyError:
             _log.info("There is no audio data for %s, ignoring.", user_id)
-            raise NoUserAdio
+            raise NoUserAudio
 
         args = [
             executable,
diff --git a/discord/sinks/pcm.py b/discord/sinks/pcm.py
@@ -32,7 +32,7 @@
 
 from .core import RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import NoUserAdio
+from .errors import NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -157,7 +157,7 @@ def format_user_audio(
         try:
             data = self.__audio_data.pop(user_id)
         except KeyError:
-            raise NoUserAdio
+            raise NoUserAudio
 
         data.seek(0)
 
diff --git a/discord/sinks/wave.py b/discord/sinks/wave.py
@@ -33,7 +33,7 @@
 
 from .core import RawData, Sink, SinkFilter, SinkHandler
 from .enums import SinkFilteringMode
-from .errors import NoUserAdio
+from .errors import NoUserAudio
 
 if TYPE_CHECKING:
     from discord import abc
@@ -166,7 +166,7 @@ def format_user_audio(
         try:
             data = self.__audio_data.pop(user_id)
         except KeyError:
-            raise NoUserAdio
+            raise NoUserAudio
 
         decoder = self.client.decoder
 
diff --git a/discord/voice/state.py b/discord/voice/state.py
@@ -27,7 +27,6 @@
 
 import asyncio
 import logging
-from os import altsep
 import select
 import socket
 import struct
@@ -149,6 +148,7 @@ def run(self) -> None:
             )
         finally:
             self.stop()
+            self._started.clear()
             self._running.clear()
             self._callbacks.clear()
 
@@ -251,13 +251,18 @@ def __init__(
 
         self._end: threading.Event = threading.Event()
 
-    def decode(self, frame: RawData) -> None:
+    def decode(self, frame: RawData, *, left: bool = False) -> None:
         if not isinstance(frame, RawData):
             raise TypeError(
                 f"expected a RawData object, got {frame.__class__.__name__}"
             )
-        self.decode_queue.append(frame)
-        _log.debug('Added frame %s to decode queue', frame)
+
+        if left:
+            self.decode_queue.appendleft(frame)
+        else:
+            self.decode_queue.append(frame)
+        self.resume()
+        _recv_log.debug('Added frame %s to decode queue', frame)
 
     def is_running(self) -> bool:
         return self._started.is_set()
@@ -266,6 +271,11 @@ def pause(self) -> None:
         self._idle_paused = False
         self._running.clear()
 
+    def is_paused(self) -> bool:
+        return self._idle_paused or (
+            not self._running.is_set() and not self._end.is_set()
+        )
+
     def resume(self, *, force: bool = False) -> None:
         if self._running.is_set():
             return
@@ -299,6 +309,7 @@ def run(self) -> None:
             )
         finally:
             self.stop()
+            self._started.clear()
             self._running.clear()
             self.decode_queue.clear()
 
@@ -311,12 +322,11 @@ def get_decoder(self, ssrc: int) -> opus.Decoder:
 
     def _do_run(self) -> None:
         while not self._end.is_set():
-            if not self.decode_queue:
-                if not self._warned_queue:
+            if not self._running.is_set():
+                if self._warned_queue:
                     _recv_log.warning('No decode queue found, waiting')
                     self._warned_queue = True
-
-                time.sleep(0.01)
+                self._running.wait()
                 continue
 
             if self._warned_queue:
@@ -326,6 +336,8 @@ def _do_run(self) -> None:
             try:
                 data = self.decode_queue.popleft()
             except IndexError:
+                _recv_log.warning('No more voice packets found, idle pausing.')
+                self.pause()
                 continue
 
             _recv_log.debug('Popped %s from the decode queue', data)
@@ -344,6 +356,14 @@ def _do_run(self) -> None:
                     exc_info=True,
                 )
 
+
+            _recv_log.info('Decoded frame %s to %s', data, data.decoded_data)
+
+            if data.decoded_data is MISSING:
+                _recv_log.warning('Decoded data %s is still MISSING after decode, appending it to left to decode', data)
+                self.decode(data, left=True)
+                continue
+
             self.state.dispatch_packet_sinks(data)
 
 
@@ -396,6 +416,7 @@ def __init__(
         self._voice_recv_socket = SocketVoiceRecvReader(self)
         self._voice_recv_socket.register(self.handle_voice_recv_packet)
         self._decoder_thread = DecoderThread(self)
+        self.start_record_socket()
         self.user_ssrc_map: dict[int, SSRC] = {}
         self.user_voice_timestamps: dict[int, tuple[int, float]] = {}
         self.sync_recording_start: bool = False
@@ -407,23 +428,19 @@ def __init__(
         self.__sink_dispatch_task_set: set[asyncio.Task[Any]] = set()
 
     def start_record_socket(self) -> None:
-        if self._voice_recv_socket.is_paused():
+        try:
+            self._voice_recv_socket.start()
+        except RuntimeError:
             self._voice_recv_socket.resume()
-            return
-        if self._voice_recv_socket.is_running():
-            return
 
-        if not self._decoder_thread.is_running():
+        try:
             self._decoder_thread.start()
-
-        self._voice_recv_socket.start()
+        except RuntimeError:
+            self._decoder_thread.resume()
 
     def stop_record_socket(self) -> None:
-        if self._voice_recv_socket.is_running():
-            self._voice_recv_socket.stop()
-
-        if self._decoder_thread.is_running():
-            self._decoder_thread.stop()
+        self._voice_recv_socket.stop()
+        self._decoder_thread.stop()
 
         for cb, args in self.recording_done_callbacks:
             task = self.loop.create_task(cb(*args))
@@ -447,13 +464,13 @@ def handle_voice_recv_packet(self, packet: bytes) -> None:
             return
 
         if self.paused_recording():
-            _log.debug('Ignoring packet %s because recording is stopped', packet)
+            _recv_log.debug('Ignoring packet %s because recording is stopped', packet)
             return
 
         data = RawData(packet, self.client)
 
         if data.decrypted_data == opus.OPUS_SILENCE:
-            _log.debug('Ignoring packet %s because it is an opus silence frame', data)
+            _recv_log.debug('Ignoring packet %s because it is an opus silence frame', data)
             return
 
         self._decoder_thread.decode(data)
