From ad2ebf9fa544550dacee57d4e157328aaeb963fc Mon Sep 17 00:00:00 2001
From: Lukasz Klich <lukasz.klich@pubnub.com>
Date: Thu, 22 Jun 2023 11:50:42 +0200
Subject: [PATCH] Is this it?

---
 examples/native_threads/whatever.py | 80 +++++++++++++++++++++++++++++
 pubnub/endpoints/pubsub/__init__.py |  2 +
 pubnub/pubnub_asyncio.py            |  8 +--
 3 files changed, 86 insertions(+), 4 deletions(-)
 create mode 100644 examples/native_threads/whatever.py

diff --git a/examples/native_threads/whatever.py b/examples/native_threads/whatever.py
new file mode 100644
index 00000000..e690af01
--- /dev/null
+++ b/examples/native_threads/whatever.py
@@ -0,0 +1,80 @@
+import asyncio
+import os
+import sys
+import random
+
+from pubnub.endpoints.pubsub import Subscribe
+from pubnub.pnconfiguration import PNConfiguration
+from pubnub.pubnub_asyncio import PubNubAsyncio
+
+d = os.path.dirname
+PUBNUB_ROOT = d(d(os.path.dirname(os.path.abspath(__file__))))
+sys.path.append(PUBNUB_ROOT)
+
+class Sub:
+    def __init__(self, loop, pubnub):
+        self.task = None
+        self._loop = loop
+        self._pubnub = pubnub
+        self._id = random.randint(1, 99999)
+        self._cancelled = False
+
+    def run(self):
+        return self._loop.create_task(self.async_run())
+
+    async def async_run(self):
+        try:
+            print(f"aaa {self._id}")
+            if self._cancelled:
+                return
+            self.req = Subscribe(self._pubnub).channels(["ch1"]).future()
+            res = await self.req
+            if res.status.is_error():
+                print("error")
+            print(f"{res.result} {self._id}")
+            return res
+        except:
+            pass
+
+    def cancel(self):
+        print(f"cancelling {self._id}")
+        self._cancelled = True
+        self.req.cancel()
+
+async def run():
+    await asyncio.sleep(2)
+    print("after")
+
+async def run_and_cancel(loop, pubnub):
+    try:
+        sub = Sub(loop, pubnub)
+        sub.run()
+        print("after")
+        sub.cancel()
+    except:
+        pass
+    await asyncio.sleep(1)
+
+async def call_after_time(loop, pubnub):
+    await asyncio.sleep(3)
+    sub = Sub(loop, pubnub)
+    sub.run()
+    print("after")
+    await asyncio.sleep(1)
+
+
+def main():
+    pnconfig = PNConfiguration()
+    pnconfig.publish_key = "demo"
+    pnconfig.subscribe_key = "demo"
+    pnconfig.uuid = "UUID-PUB"
+    loop = asyncio.new_event_loop()
+    pubnub = PubNubAsyncio(pnconfig, custom_event_loop=loop)
+    sub = Sub(loop, pubnub)
+    task3 = sub.run()
+    task1 = loop.create_task(run())
+    task2 = loop.create_task(run_and_cancel(loop, pubnub))
+    task4 = loop.create_task(call_after_time(loop, pubnub))
+    loop.run_until_complete(asyncio.gather(task1, task2, task4))
+
+main()
\ No newline at end of file
diff --git a/pubnub/endpoints/pubsub/__init__.py b/pubnub/endpoints/pubsub/__init__.py
index e69de29b..8a7fa13e 100644
--- a/pubnub/endpoints/pubsub/__init__.py
+++ b/pubnub/endpoints/pubsub/__init__.py
@@ -0,0 +1,2 @@
+from .subscribe import Subscribe
+__all__ = [Subscribe]
\ No newline at end of file
diff --git a/pubnub/pubnub_asyncio.py b/pubnub/pubnub_asyncio.py
index 2f532686..840c371f 100644
--- a/pubnub/pubnub_asyncio.py
+++ b/pubnub/pubnub_asyncio.py
@@ -41,7 +41,7 @@ def __init__(self, config, custom_event_loop=None):
         self._connector = None
         self._session = None
 
-        self._connector = aiohttp.TCPConnector(verify_ssl=True)
+        self._connector = aiohttp.TCPConnector(verify_ssl=True, loop=self.event_loop)
         self._session = aiohttp.ClientSession(
             loop=self.event_loop,
             timeout=aiohttp.ClientTimeout(connect=self.config.connect_timeout),
@@ -49,7 +49,7 @@ def __init__(self, config, custom_event_loop=None):
         )
 
         if self.config.enable_subscribe:
-            self._subscription_manager = AsyncioSubscriptionManager(self)
+            self._subscription_manager = AsyncioSubscriptionManager(self, self.event_loop)
 
         self._publish_sequence_manager = AsyncioPublishSequenceManager(self.event_loop, PubNubCore.MAX_SEQUENCE)
 
@@ -340,11 +340,11 @@ async def get_next_sequence(self):
 
 
 class AsyncioSubscriptionManager(SubscriptionManager):
-    def __init__(self, pubnub_instance):
+    def __init__(self, pubnub_instance, loop):
         subscription_manager = self
 
         self._message_worker = None
-        self._message_queue = Queue()
+        self._message_queue = Queue(loop=loop)
         self._subscription_lock = Semaphore(1)
         self._subscribe_loop_task = None
         self._heartbeat_periodic_callback = None