up version

Author: Zephor5

.flake8

@@ -0,0 +1,5 @@
+[flake8]
+ignore = E203, E266, E501, W503, F403, F401
+max-line-length = 88
+max-complexity = 18
+select = B,C,E,F,W,T4,B9

.gitignore

@@ -56,4 +56,5 @@ docs/_build/
 # PyBuilder
 target/
 
-.idea/
+.idea/
+.env/

.pre-commit-config.yaml

@@ -0,0 +1,26 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v2.5.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-docstring-first
+      - id: check-added-large-files
+        args: ['--maxkb=4096']
+      - id: check-yaml
+      - id: debug-statements
+      - id: name-tests-test
+  - repo: https://github.com/asottile/reorder_python_imports
+    rev: v1.9.0
+    hooks:
+      -  id: reorder-python-imports
+         language_version: python3
+  - repo: https://github.com/psf/black
+    rev: stable
+    hooks:
+      - id: black
+        language_version: python3
+  - repo: https://gitlab.com/pycqa/flake8
+    rev: 3.8.2
+    hooks:
+      - id: flake8

.travis.yml

@@ -14,4 +14,3 @@ deploy:
     secure: TNTQ/ZdP7ygaJTKdNop+KOqGXRghchl/8UUH5tM5P+fnFxrGBitVvWYlurhhHphjRWFXify0k4uCU3D14OIRTyPAKtC5wBqWCwV+hWx2C0T332ou8lowczvN/uEWLNnoJKiiS2127QBLUHEoFR2/LKdchCvVY7n4gnYzjkCnXin45/CdOV88065xAriO4/UG11Q2ohMTmP5Vm6z51102aPySTqwPQJB5ywg3Q7uUs2CZjQIa7IUV7ZXe/lGe0bs5jF24A1i6Vrdt5PykRuTdIsjsvvcnD4ghHpB5XOlev5OFWaXyCqAvjfZySjYaD6Tv1EJbGDZb/sTxps/7CRixCkJX7CDY6a2krMRxdq38zGnFjRN5/R9Fb5fbjQnrGgJusi0oLbNuzm1tSb37rdddYqK79iWBEUSt+nnu5LZL3FlPz6cxld8kBQId7juZ+aFY/Y6bThSB7WIVgouY8gW7t9NKDGisGrc0xYk38dH6EblYKbOQ/J4eYMqmrmgcGm71S2b4q5jOA+PJF/7NLUP7wmYyqi/6WficFsV4DDzd1wPHs7yBMwPXpNvgpPJ74iNiFIdI+HzL6LblMmTBm0EB+8Q6iYwK3GVJulrQwW5H5xaIABJNnOgMU5fQcGT5N76jbP2xHu79ZgBFzhE4fiNAlnSKnCN+qfX4TmEz09A8P1g=
   on:
     tags: true
-

README.rst

@@ -28,7 +28,7 @@ sample:
     # with connection
     from pika import URLParameters
     from pooled_pika import PooledConn
-    
+
     AMQP_PARAM = URLParameters('amqp://user:pwd@amqpserver')
     pooled_conn = PooledConn(AMQP_PARAM)
     d = pooled_conn.acquire()
@@ -43,7 +43,7 @@ or:
     # with channel
     from pika import URLParameters
     from pooled_pika import PooledConn
-    
+
     AMQP_PARAM = URLParameters('amqp://user:pwd@amqpserver')
     pooled_conn = PooledConn(AMQP_PARAM)
     d = pooled_conn.acquire(channel=True)

pooled_pika.py

@@ -1,17 +1,24 @@
 # coding=utf-8
 import logging
 import threading
-
 from contextlib import contextmanager
-from pika.adapters import TwistedProtocolConnection
+
 from pika.adapters.twisted_connection import TwistedChannel
+from pika.adapters.twisted_connection import TwistedProtocolConnection
 from pika.connection import Parameters
-from twisted.internet import reactor, defer
+from twisted.internet import defer
+from twisted.internet import reactor
 from twisted.internet.protocol import ClientCreator
 
+try:
+    itervalues = dict.itervalues
+except AttributeError:
+    itervalues = dict.values
+
 __all__ = ["VERSION", "PooledConn"]
 
-VERSION = "0.1.6"
+
+VERSION = "0.2.0"
 
 logger = logging.getLogger(__name__)
 
@@ -47,7 +54,7 @@ class PooledConn(object):
 
     loop = reactor
 
-    def __new__(cls, params, timeout_conn=None, max_size=None):
+    def __new__(cls, params, timeout_conn=0, max_size=None):
         """
         :param cls:
         :param params: connection params
@@ -67,18 +74,22 @@ def __new__(cls, params, timeout_conn=None, max_size=None):
                 cls._my_pools[1][_id] = {}
                 cls._my_channels[_id] = {}
                 cls._max_size[_id] = max_size if max_size else cls.max_size
-                cls._timeout[_id] = timeout_conn if timeout_conn > 0 else cls.timeout_conn
+                cls._timeout[_id] = (
+                    timeout_conn if timeout_conn > 0 else cls.timeout_conn
+                )
                 cls._waiting[_id] = []
                 # only works when first created
             instance.__params = cls._my_params[_id]
             instance.__max_size = cls._max_size[_id]
             instance.timeout_conn = cls._timeout[_id]
-            instance.__idle_pool, instance.__using_pool = (cls._my_pools[i][_id] for i in (0, 1))
+            instance.__idle_pool, instance.__using_pool = (
+                cls._my_pools[i][_id] for i in (0, 1)
+            )
             instance.__channel_pool = cls._my_channels[_id]
             instance.waiting = cls._waiting[_id]
             return instance
         else:
-            raise TypeError('only accept pika Parameters type')
+            raise TypeError("only accept pika Parameters type")
 
     def __init__(self, *args, **kwargs):
         """
@@ -97,25 +108,40 @@ def __connect(self, retrying=False):
         params = self.__params
         cc = ClientCreator(self.loop, TwistedProtocolConnection, params)
         _d = cc.connectTCP(params.host, params.port, timeout=self.timeout_conn)
-        _d.addCallback(lambda p: p.ready)
-        _d.addCallbacks(self._in_pool,
-                        lambda err: err if retrying or not self.retry else self.__connect(True))  # retry once when err
+
+        def conn_ready(c):
+            c.ready.addCallback(lambda _: c)
+            return c.ready
+
+        _d.addCallback(conn_ready)
+        _d.addCallbacks(
+            self._in_pool,
+            lambda err: err if retrying or not self.retry else self.__connect(True),
+        )  # retry once when err
         return _d
 
     def _in_pool(self, conn):
-        assert isinstance(conn, TwistedProtocolConnection), 'conn must be TwistedProtocolConnection'
-        logger.debug('in pool : %s' % conn)
+        assert isinstance(
+            conn, TwistedProtocolConnection
+        ), "conn must be TwistedProtocolConnection"
+        logger.debug("in pool : %s" % conn)
 
         _id = id(conn)
 
         if self.size < self.__max_size:
             # add hook to clear the bad connection object in the pool
             conn.ready = defer.Deferred()
-            conn.ready.addErrback(self._clear, self.__idle_pool, self.__using_pool, self.__channel_pool, _id)
+            conn.ready.addErrback(
+                self._clear,
+                self.__idle_pool,
+                self.__using_pool,
+                self.__channel_pool,
+                _id,
+            )
             # add new conn in using pool
             self.__using_pool[_id] = conn
         else:
-            raise RuntimeError('_in_pool, unexpected reach')
+            raise RuntimeError("_in_pool, unexpected reach")
 
         return conn
 
@@ -133,10 +159,10 @@ def _clear(reason, idle_pool, using_pool, channel_pool, conn_id):
         with _lock():
             try:
                 idle_pool.pop(conn_id)
-                logger.info('a connection lost when not using')
+                logger.info("a connection lost when not using")
             except KeyError:
                 if using_pool.pop(conn_id, None):
-                    logger.warn('connection lost when using, should be handled later')
+                    logger.warn("connection lost when using, should be handled later")
                     return reason
             finally:
                 channel_pool.pop(conn_id, None)
@@ -152,7 +178,9 @@ def _get_channel(self, conn):
                 d.callback(p[_id])
         if d is None:
             d = conn.channel()
-            d.addCallback(lambda ch: p.update({_id: ch}) or setattr(ch, 'pool_id_', _id) or ch)
+            d.addCallback(
+                lambda ch: p.update({_id: ch}) or setattr(ch, "pool_id_", _id) or ch
+            )
 
         def _h_err(ch, _conn):
             _conn.ready.addErrback(lambda _, _c: p.pop(id(_c), None), _conn)
@@ -203,5 +231,5 @@ def size(self):
 
     def clear(self):
         with _lock():
-            for c in self.__idle_pool.itervalues():
+            for c in itervalues(self.__idle_pool):
                 c.close()

setup.py

@@ -1,11 +1,12 @@
-from setuptools import setup, find_packages
+from setuptools import find_packages
+from setuptools import setup
 
 
 setup(
     name="pooled-pika",
-    version="0.1.6",
+    version="0.2.0",
     py_modules=["pooled_pika"],
-    install_requires=["twisted>=13.2.0", "pika>=0.10.0"],
+    install_requires=["twisted~=20.3.0", "pika~=1.1.0"],
     description="a connection pool wrapper about async pika",
     long_description="""
 Notice
@@ -21,7 +22,7 @@
     # with connection
     from pika import URLParameters
     from pooled_pika import PooledConn
-    
+
     AMQP_PARAM = URLParameters('amqp://user:pwd@amqpserver')
     pooled_conn = PooledConn(AMQP_PARAM)
     d = pooled_conn.acquire()
@@ -36,7 +37,7 @@
     # with channel
     from pika import URLParameters
     from pooled_pika import PooledConn
-    
+
     AMQP_PARAM = URLParameters('amqp://user:pwd@amqpserver')
     pooled_conn = PooledConn(AMQP_PARAM)
     d = pooled_conn.acquire(channel=True)
@@ -48,6 +49,5 @@
     author_email="zephor@qq.com",
     url="https://github.com/Zephor5/pooled-pika",
     license="BSD 3-clause",
-    packages=find_packages()
+    packages=find_packages(),
 )
-

test.py

@@ -0,0 +1,50 @@
+# coding=utf-8
+import logging
+
+from pika import URLParameters
+from twisted.internet import defer
+from twisted.internet import reactor
+from twisted.trial.unittest import TestCase
+
+from pooled_pika import PooledConn
+
+__author__ = "zephor"
+
+try:
+    range = xrange
+except NameError:
+    pass
+
+AMQP_PARAM = URLParameters("amqp://guest:guest@127.0.0.1:5672/spider")
+
+
+def sleep(_, seconds):
+    d = defer.Deferred()
+    reactor.callLater(seconds, d.callback, seconds)
+    return d
+
+
+class ConnTest(TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.pooled_conn = PooledConn(AMQP_PARAM,)
+
+    def test_get_conn(self):
+        self.n = 0
+        test_d = []
+
+        def plus(_c):
+            self.n += 1
+            return _c
+
+        for i in range(100):
+            d = self.pooled_conn.acquire(channel=True)
+            d.addCallback(plus)
+            d.addCallback(self.pooled_conn.release)
+            d.addErrback(lambda err: logging.error(err))
+            test_d.append(d)
+        res_d = defer.DeferredList(test_d)
+        res_d.addCallback(lambda _: self.assertEqual(self.n, 100))
+        res_d.addCallback(lambda _: self.pooled_conn.clear())
+        res_d.addCallback(sleep, 1)
+        return res_d

tox.ini

@@ -4,7 +4,7 @@
 # and then run "tox" from this directory.
 
 [tox]
-envlist = py27
+envlist = py27,py36,py37,py38
 
 [testenv]
 commands = nosetests