MOD: Add task for subscribe

Author: nmacholl

databento/live/session.py

@@ -366,12 +366,15 @@ def subscribe(
                     loop=self._loop,
                 )
 
-            self._protocol.subscribe(
+        asyncio.run_coroutine_threadsafe(
+            self._subscribe_task(
                 schema=schema,
                 symbols=symbols,
                 stype_in=stype_in,
                 start=start,
-            )
+            ),
+            loop=self._loop,
+        ).result()
 
     def resume_reading(self) -> None:
         """
@@ -496,3 +499,21 @@ async def _connect_task(
         )
 
         return transport, protocol
+
+    async def _subscribe_task(
+        self,
+        schema: Schema | str,
+        symbols: Iterable[str] | Iterable[Number] | str | Number = ALL_SYMBOLS,
+        stype_in: SType | str = SType.RAW_SYMBOL,
+        start: str | int | None = None,
+    ) -> None:
+        with self._lock:
+            if self._protocol is None:
+                return
+
+            self._protocol.subscribe(
+                schema=schema,
+                symbols=symbols,
+                stype_in=stype_in,
+                start=start,
+            )