fix: correctly unsubscribe to the ffi queue (#72)

Author: theomonnom

examples/publish_wave.py

@@ -1,4 +1,5 @@
 import asyncio
+import time
 import logging
 from signal import SIGINT, SIGTERM
 
@@ -12,30 +13,22 @@
 SAMPLE_RATE = 48000
 NUM_CHANNELS = 1
 
-
 async def publish_frames(source: livekit.AudioSource, frequency: int):
     amplitude = 32767  # for 16-bit audio
     samples_per_channel = 480  # 10ms at 48kHz
     time = np.arange(samples_per_channel) / SAMPLE_RATE
     total_samples = 0
-
     audio_frame = livekit.AudioFrame.create(
         SAMPLE_RATE, NUM_CHANNELS, samples_per_channel)
-
     audio_data = np.ctypeslib.as_array(audio_frame.data)
-
     while True:
         time = (total_samples + np.arange(samples_per_channel)) / SAMPLE_RATE
-
         sine_wave = (amplitude * np.sin(2 * np.pi *
                      frequency * time)).astype(np.int16)
         np.copyto(audio_data, sine_wave)
-
         await source.capture_frame(audio_frame)
-
         total_samples += samples_per_channel
 
-
 async def main(room: livekit.Room) -> None:
 
     @room.on("participant_disconnected")
@@ -45,11 +38,9 @@ def on_participant_disconnect(participant: livekit.Participant, *_):
     logging.info("connecting to %s", URL)
     try:
         e2ee_options = livekit.E2EEOptions()
-        e2ee_options.key_provider_options.shared_key = b"password"
 
         await room.connect(URL, TOKEN, options=livekit.RoomOptions(
             auto_subscribe=True,
-            e2ee=e2ee_options
         ))
         logging.info("connected to room %s", room.name)
     except livekit.ConnectError as e:

livekit/audio_stream.py

@@ -44,6 +44,9 @@ def __init__(self, track: Track,
 
         self._task = self._loop.create_task(self._run())
 
+    def __del__(self) -> None:
+        ffi_client.queue.unsubscribe(self._ffi_queue)
+
     async def _run(self):
         while True:
             event = await self._ffi_queue.wait_for(self._is_event)
@@ -57,6 +60,7 @@ async def _run(self):
                 break
 
     async def close(self):
+        ffi_client.queue.unsubscribe(self._ffi_queue)
         del self._ffi_handle
         await self._task
 

livekit/room.py

@@ -61,13 +61,16 @@ def __init__(self, loop: Optional[asyncio.AbstractEventLoop] = None) -> None:
 
         self._ffi_handle: Optional[FfiHandle] = None
         self._loop = loop or asyncio.get_event_loop()
-        self._ffi_queue = ffi_client.queue.subscribe(self._loop)
         self._room_queue = BroadcastQueue[proto_ffi.FfiEvent]()
         self._info = proto_room.RoomInfo()
 
         self.participants: dict[str, RemoteParticipant] = {}
         self.connection_state = ConnectionState.CONN_DISCONNECTED
 
+    def __del__(self) -> None:
+        if self._ffi_handle is not None:
+            ffi_client.queue.unsubscribe(self._ffi_queue)
+
     @property
     def sid(self) -> str:
         return self._info.sid
@@ -120,15 +123,19 @@ async def connect(self,
             req.connect.options.rtc_config.ice_servers.extend(
                 options.rtc_config.ice_servers)
 
+        # subscribe before connecting so we don't miss any events
+        self._ffi_queue = ffi_client.queue.subscribe(self._loop)
+
         try:
-            queue = ffi_client.queue.subscribe(self._loop)
+            queue = ffi_client.queue.subscribe()
             resp = ffi_client.request(req)
             cb = await queue.wait_for(lambda e: e.connect.async_id ==
                                       resp.connect.async_id)
         finally:
             ffi_client.queue.unsubscribe(queue)
 
         if cb.connect.error:
+            ffi_client.queue.unsubscribe(self._ffi_queue)
             raise ConnectError(cb.connect.error)
 
         self._ffi_handle = FfiHandle(cb.connect.room.handle.id)
@@ -169,6 +176,7 @@ async def disconnect(self) -> None:
             ffi_client.queue.unsubscribe(queue)
 
         await self._task
+        ffi_client.queue.unsubscribe(self._ffi_queue)
 
     async def _listen_task(self) -> None:
         # listen to incoming room events

livekit/video_stream.py

@@ -44,6 +44,9 @@ def __init__(self, track: Track,
 
         self._task = self._loop.create_task(self._run())
 
+    def __del__(self) -> None:
+        ffi_client.queue.unsubscribe(self._ffi_queue)
+
     async def _run(self):
         while True:
             event = await self._ffi_queue.wait_for(self._is_event)
@@ -60,6 +63,7 @@ async def _run(self):
                 break
 
     async def close(self):
+        ffi_client.queue.unsubscribe(self._ffi_queue)
         del self._ffi_handle
         await self._task