Py38

.circleci/config.yml

@@ -73,7 +73,7 @@ jobs:
     binaries:
         machine: true
         environment:
-          - PYTHON_VERSIONS: 3.5 3.6 3.7
+          - PYTHON_VERSIONS: 3.5 3.6 3.7 3.8
           - PYMODULE: pulsar
         steps:
           - checkout

.travis.yml

@@ -22,7 +22,10 @@ matrix:
           env: PYTHON_VERSION=3.6.3
 
         - os: osx
-          env: PYTHON_VERSION=3.7.0
+          env: PYTHON_VERSION=3.7.4
+
+        - os: osx
+          env: PYTHON_VERSION=3.8.0
 
 install:
     - ci/mac-pre-install.sh

pulsar/__init__.py

@@ -3,7 +3,7 @@
 from .utils.version import get_version
 
 
-VERSION = (2, 0, 2, 'final', 0)
+VERSION = (2, 0, 3, 'final', 0)
 
 
 __version__ = version = get_version(VERSION)

pulsar/asynclib/clients.py

@@ -37,7 +37,7 @@ def __init__(self, creator, pool_size=10, loop=None, timeout=None, **kw):
         self._creator = creator
         self._closed = False
         self._timeout = timeout
-        self._queue = asyncio.Queue(maxsize=pool_size, loop=loop)
+        self._queue = asyncio.Queue(maxsize=pool_size)
         self._connecting = 0
         self._loop = self._queue._loop
         self._logger = logger

pulsar/asynclib/mixins.py

@@ -178,7 +178,7 @@ class ResponsePipeline:
 
     def __init__(self, connection):
         self.connection = connection
-        self.queue = Queue(loop=connection._loop)
+        self.queue = Queue()
         self.debug = connection._loop.get_debug()
         self.worker = self.queue._loop.create_task(self._process())
         self.put = self.queue.put_nowait

pulsar/asynclib/protocols.py

@@ -228,10 +228,14 @@ async def start_serving(self, address=None, sockets=None,
                                           sock=sock,
                                           backlog=backlog,
                                           ssl=sslcontext)
-                if server:
-                    server.sockets.extend(srv.sockets)
-                else:
-                    server = srv
+                # scailer: I'm don't know why this code was written
+                # if server:
+                #     server.sockets.extend(srv.sockets)
+                # else:
+                #     server = srv
+
+                server = srv
+
         elif isinstance(address, tuple):
             server = await create_server(self.create_protocol,
                                          host=address[0],

pulsar/asynclib/timeout.py

@@ -1,4 +1,4 @@
-from asyncio import Task, CancelledError, TimeoutError
+from asyncio import CancelledError, TimeoutError, current_task
 
 
 class timeout:
@@ -16,7 +16,7 @@ def __init__(self, loop, timeout):
     def __enter__(self):
         if self._timeout is None:
             return self
-        self._task = Task.current_task(self._loop)
+        self._task = current_task()
         tm = self._loop.time() + self._timeout
         self._cancel_handler = self._loop.call_at(tm, self._cancel_task)
         return self

pulsar/utils/httpurl.py

@@ -7,7 +7,7 @@
 from io import BytesIO
 from urllib import request as urllibr
 from http import client as httpclient
-from urllib.parse import quote, splitport
+from urllib.parse import quote
 from http.cookiejar import CookieJar, Cookie
 from http.cookies import SimpleCookie
 
@@ -20,6 +20,8 @@
 HTTPError = urllibr.HTTPError
 URLError = urllibr.URLError
 parse_http_list = urllibr.parse_http_list
+re_port = re.compile('(.*):([0-9]*)$', re.DOTALL)
+
 
 tls_schemes = ('https', 'wss')
 
@@ -123,6 +125,15 @@ def iri_to_uri(iri, kwargs=None):
     return urlquote(unquote_unreserved(iri))
 
 
+def splitport(host):
+    match = re_port.match(host)
+    if match:
+        host, port = match.groups()
+        if port:
+            return host, port
+    return host, None
+
+
 def host_and_port(host):
     host, port = splitport(host)
     return host, int(port) if port else None

tests/utils/test_tools.py

@@ -148,7 +148,7 @@ def test_py_file(self):
 
     def __test_import_modules(self):
         self.assertEqual(import_modules(['gggggggggggg']), [])
-        mods = import_modules(['pulsar.async.*'])
+        mods = import_modules(['pulsar.asynclib.*'])
         self.assertTrue(mods)
 
     def test_date2timestamp(self):