Merge pull request #9 from GoodManWEN/dev

Dev

Author: GoodManWEN

cx_Oracle_async/AQ.py

@@ -1,11 +1,22 @@
-import cx_Oracle
+from cx_Oracle import MessageProperties , DEQ_NO_WAIT 
+from ThreadPoolExecutorPlus import ThreadPoolExecutor
+from asyncio import Lock as aioLock
+from collections.abc import Iterable
+from typing import Union , TYPE_CHECKING
+if TYPE_CHECKING:
+    from .connections import AsyncConnectionWrapper
+    from cx_Oracle import Queue
+    from asyncio.windows_events import ProactorEventLoop
+
 
 class AsyncQueueWrapper:
 
-    def __init__(self , queue , loop , thread_pool , ):
+    def __init__(self , queue: 'Queue', loop: 'ProactorEventLoop' , thread_pool: ThreadPoolExecutor , conn : 'AsyncConnectionWrapper'):
         self._queue = queue
         self._loop = loop
         self._thread_pool = thread_pool 
+        self._conn = conn
+        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 +25,54 @@ 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: int = -1):
+        return DeqManyWrapper(self._loop , self._thread_pool , self._queue , self._deqlock , maxMessages)
+
+    def _decode(self , _object: MessageProperties):
+        return _object.payload.decode(self._conn.encoding)
+
+    def unpack(self , _object: Union[MessageProperties , list]):
+        if isinstance(_object , Iterable):
+            return list(map(self._decode , _object))
+        else:
+            return self._decode(_object)
+
+    @property
+    def pack(self):
+        return self._conn.msgproperties
+
+    @property
+    def enqOptions(self):
+        return self._queue.enqOptions
+    
+    @property
+    def deqOptions(self):
+        return self._queue.deqOptions
 
-    def deqMany(self , maxMessages):
-        return DeqManyWrapper(self._loop , self._thread_pool , self._queue , maxMessages)
 
 class DeqManyWrapper:
 
-    def __init__(self , loop , thread_pool , queue , maxMessages):
+    def __init__(self , loop : 'ProactorEventLoop' , thread_pool : ThreadPoolExecutor, queue : 'Queue' , deqlock: aioLock , maxMessages : int):
         self._loop = loop
         self._thread_pool = thread_pool
         self._queue = queue
         self._count = 0
-        self._max = maxMessages
+        self._max = maxMessages if maxMessages > -1 else (1 << 16 - 1) 
+        self._max = self._max if self._max <= (1 << 16 - 1) else (1 << 16 - 1) 
+        self._deqlock = deqlock
+
+    def __await__(self):
+        yield from self._deqlock.acquire().__await__()
+        try:
+            ret = yield from self._loop.run_in_executor(self._thread_pool , self._queue.deqMany , self._max).__await__()
+        except Exception as exc:
+            raise exc
+        finally:
+            self._deqlock.release()
+        return ret
 
     def __aiter__(self):
         return self
@@ -35,9 +81,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 = 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:

cx_Oracle_async/__init__.py

@@ -1,8 +1,10 @@
 __version__ = ''
 
-from .utils import create_pool , makedsn
+from .utils import create_pool , makedsn , DEQ_NO_WAIT , DEQ_WAIT_FOREVER
 
 __all__ = (
 	'create_pool',
-	'makedsn'
+	'makedsn',
+    'DEQ_NO_WAIT',
+    'DEQ_WAIT_FOREVER'
 )

cx_Oracle_async/connections.py

@@ -1,7 +1,11 @@
 from .context import AbstractContextManager as BaseManager
 from .cursors import AsyncCursorWrapper , AsyncCursorWrapper_context
 from .AQ import AsyncQueueWrapper
-
+from cx_Oracle import Connection , SessionPool
+from ThreadPoolExecutorPlus import ThreadPoolExecutor
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from asyncio.windows_events import ProactorEventLoop
 
 class AsyncConnectionWrapper_context(BaseManager):
 
@@ -15,12 +19,11 @@ async def __aexit__(self, exc_type, exc, tb):
 
 class AsyncConnectionWrapper:
 
-    def __init__(self , conn , loop , thread_pool , pool):
+    def __init__(self , conn: Connection, loop: 'ProactorEventLoop', thread_pool: ThreadPoolExecutor, pool: SessionPool):
         self._conn = conn  
         self._loop = loop
         self._pool = pool
         self._thread_pool = thread_pool
-        self.encoding = self._conn.encoding
 
     def cursor(self):
         coro = self._loop.run_in_executor(self._thread_pool , self._cursor)
@@ -32,8 +35,12 @@ def _cursor(self):
     def msgproperties(self , *args , **kwargs):
         return self._conn.msgproperties(*args , **kwargs)
 
+    @property
+    def encoding(self):
+        return self._conn.encoding
+
     async def queue(self , *args , **kwargs):
-        return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool)
+        return AsyncQueueWrapper(self._conn.queue(*args , **kwargs) , self._loop , self._thread_pool , self)
 
     async def gettype(self , *args , **kwargs):
         '''

cx_Oracle_async/context.py

@@ -1,11 +1,13 @@
+from types import CoroutineType
+
 class AbstractContextManager:
 
-    def __init__(self , coro):
+    def __init__(self , coro : CoroutineType):
         self._coro = coro
         self._obj = None
 
-    def __next__(self):
-        return self.send(None)
+    # def __next__(self):
+    #     return self.send(None)
 
     def __iter__(self):
         return self._coro.__await__()

cx_Oracle_async/cursors.py

@@ -1,15 +1,21 @@
 from .context import AbstractContextManager as BaseManager
+from ThreadPoolExecutorPlus import ThreadPoolExecutor
+from types import CoroutineType
+from cx_Oracle import Cursor
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from asyncio.windows_events import ProactorEventLoop
 
 
 class AsyncCursorWrapper_context(BaseManager):
 
-    def __init__(self , coro):
+    def __init__(self , coro : CoroutineType):
         super().__init__(coro)
 
 
 class AsyncCursorWrapper:
 
-    def __init__(self , cursor , loop , thread_pool):
+    def __init__(self , cursor : Cursor, loop : 'ProactorEventLoop' , thread_pool : ThreadPoolExecutor):
         self._cursor = cursor
         self._loop = loop
         self._thread_pool = thread_pool

cx_Oracle_async/pools.py

@@ -1,9 +1,14 @@
 from .context import AbstractContextManager as BaseManager
 from .connections import AsyncConnectionWrapper , AsyncConnectionWrapper_context
 from ThreadPoolExecutorPlus import ThreadPoolExecutor
+from cx_Oracle import SessionPool
+from types import CoroutineType
 import asyncio
 import platform
 import os
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from asyncio.windows_events import ProactorEventLoop
 
 pltfm = platform.system()
 if pltfm == 'Windows':
@@ -16,7 +21,7 @@
 
 class AsyncPoolWrapper_context(BaseManager):
 
-    def __init__(self , coro):
+    def __init__(self , coro : CoroutineType):
         super().__init__(coro)
 
     async def __aexit__(self, exc_type, exc, tb):
@@ -26,8 +31,7 @@ async def __aexit__(self, exc_type, exc, tb):
 
 class AsyncPoolWrapper:
 
-    def __init__(self , pool , loop = None):
-        
+    def __init__(self , pool : SessionPool, loop : 'ProactorEventLoop' = None):
         if loop == None:
             loop = asyncio.get_running_loop()
         self._thread_pool = ThreadPoolExecutor(max_workers = max(DEFAULT_MAXIMUM_WORKER_NUM , pool.max << DEFAULT_MAXIMUM_WORKER_TIMES)) 

cx_Oracle_async/utils.py

@@ -1,17 +1,23 @@
+from .pools import AsyncPoolWrapper , AsyncPoolWrapper_context
 from ThreadPoolExecutorPlus import ThreadPoolExecutor
 import cx_Oracle as cxor
 import asyncio
-from .pools import AsyncPoolWrapper , AsyncPoolWrapper_context
+from typing import TYPE_CHECKING
+if TYPE_CHECKING:
+    from asyncio.windows_events import ProactorEventLoop
+
 
 makedsn = cxor.makedsn
+DEQ_NO_WAIT = cxor.DEQ_NO_WAIT
+DEQ_WAIT_FOREVER = cxor.DEQ_WAIT_FOREVER
 
 async def _create_pool(
-        host=None,
-        port=None,
-        service_name=None,
-        sid=None,
-        loop=None,
-        dsn=None,
+        host: str =None,
+        port: str =None,
+        service_name: str =None,
+        sid: str =None,
+        loop: 'ProactorEventLoop' =None,
+        dsn: str =None,
         **kwargs
     ):
     if loop == None:
@@ -27,11 +33,11 @@ async def _create_pool(
     return pool
 
 def create_pool(
-        user=None, 
-        password=None, 
-        dsn=None, 
-        min=2, 
-        max=4, 
+        user: str =None, 
+        password: str =None, 
+        dsn: str =None, 
+        min: int =2, 
+        max: int =4, 
         increment=1, 
         connectiontype=cxor.Connection, 
         threaded=True, 
@@ -46,11 +52,11 @@ def create_pool(
         maxLifetimeSession=0, 
         sessionCallback=None, 
         maxSessionsPerShard=0,
-        host=None,
-        port=None,
-        service_name=None,
-        sid=None,
-        loop=None,
+        host: str =None,
+        port: str =None,
+        service_name: str =None,
+        sid: str =None,
+        loop: 'ProactorEventLoop' =None,
     ):
     coro = _create_pool(
         user=user,