Add AQ options support

Author: GoodManWEN

cx_Oracle_async/AQ.py

@@ -1,11 +1,14 @@
 import cx_Oracle
+from asyncio import Lock as aioLock
+
 
 class AsyncQueueWrapper:
 
     def __init__(self , queue , loop , thread_pool , ):
         self._queue = queue
         self._loop = loop
         self._thread_pool = thread_pool 
+        self._deqlock = aioLock()
 
     async def enqOne(self , *args , **kwargs):
         return await self._loop.run_in_executor(self._thread_pool , self._queue.enqOne , *args , **kwargs)
@@ -14,19 +17,30 @@ async def enqMany(self , *args , **kwargs):
         return await self._loop.run_in_executor(self._thread_pool , self._queue.enqMany , *args , **kwargs)
 
     async def deqOne(self , *args , **kwargs):
-        return await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne , *args , **kwargs)
+        async with self._deqlock:
+            return await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne , *args , **kwargs)
 
     def deqMany(self , maxMessages):
-        return DeqManyWrapper(self._loop , self._thread_pool , self._queue , maxMessages)
+        return DeqManyWrapper(self._loop , self._thread_pool , self._queue , self._deqlock , maxMessages)
+
+    @property
+    def enqOptions(self):
+        return self._queue.enqOptions
+    
+    @property
+    def deqOptions(self):
+        return self._queue.deqOptions
+
 
 class DeqManyWrapper:
 
-    def __init__(self , loop , thread_pool , queue , maxMessages):
+    def __init__(self , loop , thread_pool , queue , deqlock ,maxMessages):
         self._loop = loop
         self._thread_pool = thread_pool
         self._queue = queue
         self._count = 0
         self._max = maxMessages
+        self._deqlock = deqlock
 
     def __aiter__(self):
         return self
@@ -35,9 +49,10 @@ async def __anext__(self):
         self._count += 1
         if self._count <= self._max:
             _tmp = self._queue.deqOptions.wait
-            self._queue.deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
-            data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne)
-            self._queue.deqOptions.wait = _tmp
+            async with self._deqlock:
+                self._queue.deqOptions.wait = cx_Oracle.DEQ_NO_WAIT
+                data = await self._loop.run_in_executor(self._thread_pool , self._queue.deqOne)
+                self._queue.deqOptions.wait = _tmp
             if data:
                 return data
             else:

docs/temporary_document_of_AQ.md

@@ -97,5 +97,95 @@ async def main():
     async with cx_Oracle_async.create_pool(user = '' , password = '' , dsn = dsn) as oracle_pool:
         await features(oracle_pool)
 
+asyncio.run(main())
+```
+
+It is noteworthy that since we were not implement this library asynchronous in a very basic level ,yet it's just a wrapper of synchronous functions via threads , that makes it not gonna work if you are doing two different things in a single connection at a time. For example in the following situation the code will **NOT** work:
+
+```Python
+import cx_Oracle_async
+import asyncio
+
+async def coro_to_get_from_queue(conn , queue , oracle_pool):
+    print(f"coroutine start fetching")
+    ret = (await queue.deqOne()).payload.decode(conn.encoding)
+    print(f"coroutine returned , {ret=}")
+    await conn.commit()
+
+async def main():
+    loop = asyncio.get_running_loop()
+    dsn = cx_Oracle_async.makedsn(
+        host = 'localhost',
+        port = '1521',
+        service_name='orcl'
+    )
+    async with cx_Oracle_async.create_pool(user = 'C##SCOTT' , password = '123456' , dsn = dsn) as oracle_pool:
+        async with oracle_pool.acquire() as conn:
+            queue = await conn.queue("DEMO_RAW_QUEUE")
+            loop.create_task(coro_to_get_from_queue(conn , queue , oracle_pool))
+
+            await asyncio.sleep(1)
+            
+            data = 'Hello World'
+            print(f"mainthread put some thing in queue ,{data=}")
+            await queue.enqOne(conn.msgproperties(payload=data))
+            await conn.commit()
+            print(f"mainthread put some thing done")
+
+    await asyncio.sleep(1)
+    print('Process terminated.')
+
+asyncio.run(main())
+```
+
+As we planned , there should be a fetching thread(coroutine) start fetcing , this action will block since the queue is empty , and will return until there's something put into the queue. Then after one second sleep , the main thread will put 'Hello World' into AQ and that will trigger the blocked fetching thread , and then the whole program terminated.
+
+However we will find the program blocking forever in real practice. That's because since `queue.deqOptions.wait` equals to `cx_Oracle.DEQ_WAIT_FOREVER` thus while there's nothing in the queue , the query will block **AND** this will take over the control of connection thread , which makes it impossible for the following code to put anything into the queue using the same thread, thus makes it a deadlock.
+
+If you would like to achieve the same result , you should do that in **ANOTHER** connection thread. Simply modify the code as follow:
+```Python
+import cx_Oracle_async
+import asyncio
+from async_timeout import timeout
+
+async def coro_to_get_from_queue(conn , queue , oracle_pool):
+    try:
+        async with timeout(2):
+            print(f"coroutine start fetching")
+            ret = (await queue.deqOne()).payload.decode(conn.encoding)
+            print(f"coroutine returned , {ret=}")
+            await conn.commit()
+    except asyncio.TimeoutError:
+        print('two seconds passed , timeout triggered.')
+        async with oracle_pool.acquire() as conn2:
+            queue2 = await conn2.queue("DEMO_RAW_QUEUE")
+            data = 'Hello World'
+            print(f"another connection put some thing in queue ,{data=}")
+            await queue2.enqOne(conn2.msgproperties(payload=data))
+            await conn2.commit()
+            print(f"another connection put some thing done")
+
+async def main():
+    loop = asyncio.get_running_loop()
+    dsn = cx_Oracle_async.makedsn(
+        host = 'localhost',
+        port = '1521',
+        service_name='orcl'
+    )
+    async with cx_Oracle_async.create_pool(user = 'C##SCOTT' , password = '123456' , dsn = dsn) as oracle_pool:
+        async with oracle_pool.acquire() as conn:
+            queue = await conn.queue("DEMO_RAW_QUEUE")
+            loop.create_task(coro_to_get_from_queue(conn , queue , oracle_pool))
+
+            await asyncio.sleep(1)
+
+            cursor = await conn.cursor()
+            await cursor.execute(f"SELECT COUNT(*) FROM DEPT")
+            fetch_result = await cursor.fetchall()
+            print(f"main thread continue , {fetch_result=}")
+
+    await asyncio.sleep(1)
+    print('Process terminated.')
+
 asyncio.run(main())
 ```

tests/test_aq.py

@@ -4,11 +4,43 @@
 import asyncio
 import time
 from cx_Oracle_async import *
+import cx_Oracle
+
+async def modify_deqopts(queue , val):
+    queue.deqOptions.wait = val
+
+async def fetch_from_queue_no_wait(oracle_pool , loop):
+    async with oracle_pool.acquire() as conn:
+        queue = await conn.queue("DEMO_RAW_QUEUE")
+        loop.create_task(modify_deqopts(queue , cx_Oracle.DEQ_NO_WAIT))
+        ret = await queue.deqOne()
+        if ret:
+            ret = ret.payload.decode(conn.encoding)
+        await conn.commit()
+        return ret
+
+async def fetch_from_queue_wait_forever(oracle_pool , loop):
+    async with oracle_pool.acquire() as conn:
+        queue = await conn.queue("DEMO_RAW_QUEUE")
+        loop.create_task(modify_deqopts(queue , cx_Oracle.DEQ_WAIT_FOREVER))
+        ret = await queue.deqOne()
+        if ret:
+            ret = ret.payload.decode(conn.encoding)
+        return ret
+
+async def put_into_queue(oracle_pool , loop):
+    await asyncio.sleep(2)
+    async with oracle_pool.acquire() as conn:
+        queue = await conn.queue("DEMO_RAW_QUEUE")
+        await queue.enqOne(conn.msgproperties(payload='Hello World'))
+        await conn.commit()
 
 @pytest.mark.asyncio
 async def test_multiquery():
+    loop = asyncio.get_running_loop()
     dsn  = makedsn('localhost','1521',sid='xe')
-    async with create_pool(user='system',password='oracle',dsn=dsn) as oracle_pool:
+    INAQ = 0.5
+    async with create_pool(user='system',password='oracle',dsn=dsn,max=4) as oracle_pool:
         async with oracle_pool.acquire() as conn:
             async with conn.cursor() as cursor:
                 try:
@@ -123,4 +155,21 @@ async def test_multiquery():
                 async for m in queue.deqMany(maxMessages=5):
                     res.append(m.payload.decode(conn.encoding))
                 ed_time = time.time()
-                assert (ed_time - st_time) <= 0.5
+                assert (ed_time - st_time) <= 0.5
+
+        # test aq options
+        st_time = time.time()
+        _task = loop.create_task(fetch_from_queue_no_wait(oracle_pool , loop))
+        result = await _task
+        ed_time = time.time()
+        assert result == None
+        assert (ed_time - st_time) <= INAQ
+
+        #
+        st_time = time.time()
+        _task = loop.create_task(fetch_from_queue_wait_forever(oracle_pool , loop))
+        loop.create_task(put_into_queue(oracle_pool , loop))
+        result = await _task
+        ed_time = time.time()
+        assert result == "Hello World"
+        assert (2 - INAQ) <= (ed_time - st_time) <= (2 + INAQ)