celery-kombu-2492

conftest.py
docs
examples
kombu
kombu.egg-info
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
./Changelog.rst:4674:- ``Transport.on_poll_start()``
./Changelog.rst:4680:- ``Transport.on_poll_start(poller)``
./kombu/connection.py:302:    def register_with_event_loop(self, loop):
./kombu/transport/base.py:213:    def register_with_event_loop(self, connection, loop):
./kombu/transport/librabbitmq.py:163:    def register_with_event_loop(self, connection, loop):
./kombu/transport/pyamqp.py:222:    def register_with_event_loop(self, connection, loop):
./kombu/transport/qpid.py:1542:    def register_with_event_loop(self, connection, loop):
./kombu/transport/redis.py:511:    def _on_connection_disconnect(self, connection):
./kombu/transport/redis.py:556:    def on_poll_start(self):
…[1494 chars omitted]…
est_redis.py:1243:        transport.cycle.on_poll_start.assert_called_with()
./t/unit/transport/test_redis.py:1718:    def test_on_poll_start(self):
./t/unit/transport/test_redis.py:1721:        p.on_poll_start()
./t/unit/transport/test_redis.py:1729:        p.on_poll_start()
./t/unit/transport/test_redis.py:1735:        p.on_poll_start()
./t/unit/transport/test_redis.py:1741:        p.on_poll_start()

t/unit/transport/test_redis
…[423 chars omitted]…

t/unit/transport/test_redis.py:1905:        p._fd_to_chan = Mock()
t/unit/transport/test_redis.py:1913:        p._fd_to_chan.clear.assert_called_with()
t/unit/transport/test_redis.py:1926:        assert p._fd_to_chan[10] == (channel, type)
t/unit/transport/test_redis.py:2043:        p._fd_to_chan[1] = (channel, 'BRPOP')
t/unit/transport/test_redis.py:2053:        p._fd_to_chan[1] = (channel, 'BRPOP')

    def on_readable(self, fileno):
        chan, type = self._fd_to_chan[fileno]
        if chan.qos.can_consume():
            chan.handlers[type]()

    def handle_event(self, fileno, event):
        if event & READ:
            return self.on_readable(fileno), self
        elif event & ERR:
            chan, type = self._fd_to_chan[fileno]
…[8629 chars omitted]…
:        # Ensure 'subclient' is not in __dict__ (not yet accessed/cached)
1825:        channel.__dict__.pop('subclient', None)
1828:        p.maybe_check_subclient_health()  # must not raise
1830:    def test_maybe_check_subclient_health_swallows_connection_error(self):
1844:        client = Mock(name='subclient')
1846:        channel.__dict__['subclient'] = client
650:class Channel(virtual.Channel):

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[489 chars omitted]…
   +    where {<function Transport.register_with_event_loop.<locals>.on_poll_start at 0x79ad07914040>} = <Mock name='loop' id='133784062960576'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]

+def _connection_socket(connection):
+    """Return a connection's socket without allowing stale cleanup to fail."""
+    try:
+        return connection._sock
+    except Exception:
+        return None
+
+
+def _connection_matches_channel(channel, connection):
+    """Whether either Redis client owned by *channel* uses *connection*."""
+    for name in ('client', 'subclient'):
+        try:
+            client = getattr(chann
…[3582 chars omitted]…
 It is deliberately a
+            # per-tick callback: removing it while an old connection is
+            # being cleaned up can race with registration of its replacement
+            # and leave the worker alive but no longer consuming messages.
+            _remove_connection_fds(cycle, connection, loop=loop)
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', transport_options={})
        loop = Mock(name='loop')
        loop.on_tick = set()
        redis.
…[1782 chars omitted]…
:227: in <module>
    from botocore.client import BaseClient, Config
E   ModuleNotFoundError: No module named 'botocore'
=========================== short test summary info ============================
ERROR t/unit/transport/SQS - ModuleNotFoundError: No module named 'botocore'
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 error in 0.72s
 M kombu/transport/redis.py

docs
examples
kombu
kombu.egg-info
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
./Changelog.rst:4674:- ``Transport.on_poll_start()``
./Changelog.rst:4680:- ``Transport.on_poll_start(poller)``
./kombu/connection.py:302:    def register_with_event_loop(self, loop):
./kombu/transport/base.py:213:    def register_with_event_loop(self, connection, loop):
./kombu/transport/librabbitmq.py:163:    def register_with_event_loop(self, connection, loop):
./kombu/transport/pyamqp.py:222:    def register_with_event_loop(self, connection, loop):
./kombu/transport/qpid.py:1542:    def register_with_event_loop(self, connection, loop):
./kombu/transport/redis.py:487:        self._fd_to_chan = {}
./kombu/transport/redis.py:502:        self._fd_to_chan.clear()
./kombu/transport/redis.
…[2562 chars omitted]…
transport/test_redis.py:1905:        p._fd_to_chan = Mock()
./t/unit/transport/test_redis.py:1913:        p._fd_to_chan.clear.assert_called_with()
./t/unit/transport/test_redis.py:1926:        assert p._fd_to_chan[10] == (channel, type)
./t/unit/transport/test_redis.py:2043:        p._fd_to_chan[1] = (channel, 'BRPOP')
./t/unit/transport/test_redis.py:2053:        p._fd_to_chan[1] = (channel, 'BRPOP')

                R.VERSION = Rv

    def test_get_prefixed_client(self):
        from kombu.transport.redis import PrefixedStrictRedis
        self.channel.global_keyprefix = "test_"
        PrefixedRedis = redis.Channel._get_client(self.channel)
        assert isinstance(PrefixedRedis(), PrefixedStrictRedis)

    def test_get_response_error(self):
        from redis.exceptions import ResponseError
        assert redis.Channel._get_response_error(self.channel) is ResponseError


…[8629 chars omitted]…
      with pytest.raises(redis.Empty):
            p.get(Mock())

        channel._poll_error.assert_called_with('BRPOP')


class test_Mutex:

    def test_mutex(self, lock_id='xxx'):
        client = Mock(name='client')
        lock = client.lock.return_value = Mock(name='lock')

        # Won
        lock.acquire.return_value = True
        held = False
        with redis.Mutex(client, 'foo1', 100):

                else:
                    fun()

    @property
    def fds(self):
        return self._fd_to_chan


class Channel(virtual.Channel):
    """Redis Channel."""

    QoS = QoS

    _client = None
    _subclient = None
    _closing = False
    supports_fanout = True
    keyprefix_queue = '_kombu.binding.%s'
    keyprefix_fanout = '/{db}.'
    sep = '\x06\x16'
    _in_poll = False
    _in_listen = False
    _fanout_queues = {}
    ack_emulation = True

    d
…[5972 chars omitted]…
ass SentinelManagedSSLConnection(
            sentinel.SentinelManagedConnection,
            redis.SSLConnection):
        """Connect to a Redis server using Sentinel + TLS.

        Use Sentinel to identify which Redis server is the current master
        to connect to and when connecting to the Master server, use an
        SSL Connection.
        """

        pass


class SentinelChannel(Channel):

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[489 chars omitted]…
   +    where {<function Transport.register_with_event_loop.<locals>.on_poll_start at 0x7f43c1a2b920>} = <Mock name='loop' id='139928985297664'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]

    return fd if isinstance(fd, int) and fd >= 0 else None


def _channel_uses_connection(channel, connection):
    for attr in ('client', 'subclient'):
        try:
            client = getattr(channel, attr)
            if getattr(client, 'connection', None) is connection:
                return True
        except (AttributeError, TypeError):
            pass
    return False


def _remove_connection_mapping(cycle, connection, sock):
    """Remove the fd for a socket, or stale entries for a socketless client."""
    try:
        mapping = cycle._fd_to_chan
    except Exception:
        return

    if sock is not None:
        fd = _s
…[4977 chars omitted]…
on)
+            if sock is not None:
                 try:
-                    loop.on_tick.remove(on_poll_start)
-                except KeyError:
+                    loop.remove(sock)
+                except Exception:
                     pass
+            _remove_connection_mapping(cycle, connection, sock)
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', transport_options={})
        loop = Mock(name='loop')
        loop.on_tick = set()
        redis.Transport.register_with_event_loop(transport, conn, loop)
   
…[457 chars omitted]…
='131401462056000'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 1 passed, 114 deselected in 0.57s

+    except Exception:
+        return None
+
+
+def _socket_fileno(sock):
+    if isinstance(sock, int):
+        return sock if sock >= 0 else None
+    try:
+        fd = sock.fileno()
+    except Exception:
+        return None
+    return fd if isinstance(fd, int) and fd >= 0 else None
+
+
+def _channel_uses_connection(channel, connection):
+    for attr in ('client', 'subclient'):
+        try:
+            client = getattr(channel, attr)
+            if getattr(client, 'connection', 
…[2966 chars omitted]…
on)
+            if sock is not None:
                 try:
-                    loop.on_tick.remove(on_poll_start)
-                except KeyError:
+                    loop.remove(sock)
+                except Exception:
                     pass
+            _remove_connection_mapping(cycle, connection, sock)
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

docs
examples
kombu
kombu.egg-info
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
bash: line 6: rg: command not found

        assert chan.ResponseError
        assert conn.transport.connection_errors
        assert conn.transport.channel_errors

    def test_brpop_timeout_propagates_from_transport_options(self):
        # Set either polling_interval or brpop_timeout to 2
        conn = Connection("redis://localhost/0", transport_options={"polling_interval": 2})

        # Avoid network I/O during Channel.__init__()
        with patch("kombu.transport.redis.redis.Redis.ping", ret
…[8628 chars omitted]…
      with pytest.raises(redis.Empty):
            p.get(Mock())

        channel._poll_error.assert_called_with('BRPOP')


class test_Mutex:

    def test_mutex(self, lock_id='xxx'):
        client = Mock(name='client')
        lock = client.lock.return_value = Mock(name='lock')

        # Won
        lock.acquire.return_value = True
        held = False
        with redis.Mutex(client, 'foo1', 100):

537:                (channel, client, cmd) in self._chan_to_sock)
550:        if not self._client_registered(channel, channel.subclient, 'LISTEN'):
552:            self._re
…[2269 chars omitted]…
       sentinel://0.0.0.0:26379;sentinel://0.0.0.0:26380/...

    where each sentinel is separated by a `;`.

t/unit/transport/test_redis.py:427:            def _on_connection_disconnect(self, connection):
t/unit/transport/test_redis.py:486:            def _on_connection_disconnect(self, connection):
t/unit/transport/test_redis.py:1217:        transport.cycle._on_connection_disconnect(self.connection)

        conn.transport.cycle.add.assert_called_once()
        assert len(conn.transport.channels) == 1
        # the channel was flagged as registered into poller
        assert chan._registered

    def test_redis_on_disconnect_channel_only_if_was_registered(self):
        """Test should check if the _on_disconnect method is called only
           if the channel was registered into the poller."""
        # given: moc
…[5905 chars omitted]…
r_tcp_connparams(self, socket_keepalive=None,
                               socket_keepalive_options=None, **params):
        return params

    def _process_credential_provider(self, credential_provider, connparams):
        if credential_provider:
            if isinstance(credential_provider, str):
kombu/transport/redis.py:822:            self.connection.cycle._on_connection_disconnect(connection)

Connection String
=================
Connection string has the following format:

.. code-block::

    redis://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]
    rediss://[USER:PASSWORD@]REDIS_ADDRESS[:PORT][/VIRTUALHOST]

To use sentinel for dynamic Redis discovery,
the connection string has following format:

.. code-block::

    sentinel://[USER:PASSWORD@]SENTINEL_ADDRESS[:PORT]

Transport Options
=================
* ``sep``
* ``ack_emulation``: (bool) If set to True transport will
  simulate Acknowledge of AMQP protocol.
* ``unacked_key``
* ``unacked_index_key``
* ``unacked_mutex_key``
* ``unacked_mutex_expire``
* ``visibility_timeout``
* ``unacked_restore_limit`
…[1854 chars omitted]…
.compat import register_after_fork
84:from kombu.utils.encoding import bytes_to_str
85:from kombu.utils.eventio import ERR, READ, poll
86:from kombu.utils.functional import accepts_argument
87:from kombu.utils.json import dumps, loads
88:from kombu.utils.objects import cached_property
89:from kombu.utils.scheduling import cycle_by_name
90:from kombu.utils.url import _parse_url
92:from . import virtual

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[489 chars omitted]…
   +    where {<function Transport.register_with_event_loop.<locals>.on_poll_start at 0x78b035f480e0>} = <Mock name='loop' id='132698214179776'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]

E           AssertionError: assert 1 == 0
E            +  where 1 = len({<function Transport.register_with_event_loop.<locals>.on_poll_start at 0x75711108bec0>})
E            +    where {<function Transport.register_with_event_loop.<locals>.on_poll_start at 0x75711108bec0>} = <Mock name='loop' id='129128476156384'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]
1 failed, 142 passed in 4.93s

        assert len(loop.on_tick) == 1
        transport.cycle._on_connection_disconnect(self.connection)
        if fds:
>           assert len(loop.on_tick) == 0
E           AssertionError: assert 1 == 0
E            +  where 1 = len({<function Transport.register_with_event_loo
…[5999 chars omitted]…
                    loop.remove(sock)
+                except Exception:
                     pass
+            try:
+                _remove_connection_poll_entries(cycle, connection)
+            except Exception:
+                # Disconnect cleanup must never prevent the next poll tick.
+                pass
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

            cycle.poller.unregister(sock)
        except Exception:
            # Disconnect is cleanup and must never mask reconnect handling.
            pass

        fd = None
        if isinstance(sock, numbers.Integral):
            fd = int(sock)
        else:
            try:
                candidate = sock.fileno()
                if candidate >= 0:
            
…[553 chars omitted]…
t cleanup must never prevent the next poll tick.
                pass
        cycle._on_connection_disconnect = _on_disconnect

        def on_poll_start():
            cycle_poll_start()
            [add_reader(fd, on_readable, fd) for fd in cycle.fds]
        loop.on_tick.add(on_poll_start)
=================================== FAILURES ===================================
1 failed, 142 passed in 5.18s

                candidate = sock.fileno()
                if candidate >= 0:
                    fd = candidate
            except Exception:
                pass

        if fd is not None:
            try:
                entry = cycle._fd_to_chan.get(fd)
            except Exception:
                entry = None
            if entry is not None:
                try:
                    belongs = _channel_uses_connection(entry[0], connection)
                except (IndexError, KeyError, TypeError):
                    belongs = None
         
…[749 chars omitted]…
red_sock is sock or client_connection is connection:
                try:
                    del cycle._chan_to_sock[key]
                except (KeyError, TypeError):
                    pass
    except (AttributeError, TypeError):
        pass

    if sock is None:
        try:
            entries = list(cycle._fd_to_chan.items())
        except (AttributeError, TypeError):
            entries = ()

docs
examples
kombu
kombu.egg-info
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
bash: line 6: rg: command not found

    if redis:
        connection_errors, channel_errors = get_redis_error_classes()

    def __init__(self, *args, **kwargs):
        if redis is None:
            raise ImportError('Missing redis library (pip install redis)')
        super().__init__(*args, **kwargs)

        # All channels share the same poller.
        self.cycle = MultiChannelPoller()
        # Use polling_interval to set brpop_timeout if provided, but do not modify polling_interval itself.
        if self.polling_interval is not None:
            self.brpop_timeout 
…[4686 chars omitted]…
    assert conn1.disconnected
1857:        p._fd_to_chan[13] = chan, 'BRPOP'
1875:        p._fd_to_chan = {1: 2}
1876:        assert p.fds == p._fd_to_chan
1905:        p._fd_to_chan = Mock()
1913:        p._fd_to_chan.clear.assert_called_with()
1926:        assert p._fd_to_chan[10] == (channel, type)
2043:        p._fd_to_chan[1] = (channel, 'BRPOP')
2053:        p._fd_to_chan[1] = (channel, 'BRPOP')

        assert redis.Channel._get_response_error(self.channel) is ResponseError

    def test_avail_client(self):
        self.channel._pool = Mock()
        cc = self.channel._create_client = Mock()
        with self.channel.conn_or_acquire():
            pass
        cc.assert_called_with()

    def test_register_with_event_loop(self):
        transport = self.connection.transport
        transport.cycle = Mock(name='cycle')
        transport
…[8629 chars omitted]…
 the current master
        to connect to and when connecting to the Master server, use an
        SSL Connection.
        """

        pass


class SentinelChannel(Channel):
    """Channel with explicit Redis Sentinel knowledge.

    Broker url is supposed to look like:

    .. code-block::

        sentinel://0.0.0.0:26379;sentinel://0.0.0.0:26380/...

    where each sentinel is separated by a `;`.

+    removal, but its descriptor must not be used to alter the channel map.
+    """
+    if isinstance(sock, int):
+        return sock
+    try:
+        fd = sock.fileno()
+    except Exception:
+        return None
+    return fd if fd >= 0 else None
+
+
+def _remove_connection_fd(fd_to_chan, sock):
+    fd = _connection_fd(sock)
+    if fd is not None:
+        try:
+            fd_to_chan.pop(fd, None)
+        except Exception:
+            # Disconnect is best eff
…[2584 chars omitted]…
remove(on_poll_start)
-                except KeyError:
+                    loop.remove(sock)
+                except Exception:
                     pass
+                _remove_connection_fd(cycle._fd_to_chan, sock)
+            else:
+                _remove_connection_channels(cycle._fd_to_chan, connection)
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

setup.py:129:        'Programming Language :: Python :: Implementation :: CPython',

        return None


def _remove_connection_fd(fd_to_chan, sock):
    fd = _connection_fd(sock)
    if fd is not None:
        try:
            fd_to_chan.pop(fd, None)
        except Exception:
            # Disconnect is best effort; a concurrently cleaned-up entry is
            # not an error.
            pass


def _channel_uses_connection(channel, connection):
    for attr in ('client', 'subclient'):
        try:
            client = getattr(channel, attr)
            if getattr(client, 'co
…[3480 chars omitted]…
nels(cycle._fd_to_chan, connection)
        cycle._on_connection_disconnect = _on_disconnect

        def on_poll_start():
            cycle_poll_start()
            [add_reader(fd, on_readable, fd) for fd in cycle.fds]
        loop.on_tick.add(on_poll_start)

        # Cancel stale timer entries from a previous connection before
        # registering new ones. Without this, each reconnect accumulates

+
+    A socket which has been closed can still be passed to the event loop for
+    removal, but its descriptor must not be used to alter the channel map.
+    """
+    if isinstance(sock, int):
+        return sock
+    try:
+        fd = sock.fileno()
+        return fd if fd >= 0 else None
+    except Exception:
+        return None
+
+
+def _remove_connection_fd(fd_to_chan, sock):
+    fd = _connection_fd(sock)
+    
…[2815 chars omitted]…
remove(on_poll_start)
-                except KeyError:
+                    loop.remove(sock)
+                except Exception:
                     pass
+                _remove_connection_fd(cycle._fd_to_chan, sock)
+            else:
+                _remove_connection_channels(cycle._fd_to_chan, connection)
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

    ???
/usr/local/lib/python3.12/site-packages/_pytest/assertion/rewrite.py:186: in exec_module
    exec(co, module.__dict__)
t/unit/transport/SQS/conftest.py:15: in <module>
    from kombu.transport.SQS.SNS import SNS, _SnsSubscription
kombu/transport/SQS/__init__.py:227: in <module>
    from botocore.client import BaseClient, Config
E   ModuleNotFoundError: No module named 'botocore'
=========================== short test summary info ============================
ERROR t/unit/transport/SQS - ModuleNotFoundError: No module named 'botocore'
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 error in 0.96s

        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', transport_options={})
        loop = Mock(name='loop')
        loop.on_tick = set()
        redis.
…[531 chars omitted]…
5440'>.on_tick

t/unit/transport/test_redis.py:1219: AssertionError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 62 passed in 1.04s
 M kombu/transport/redis.py

docs
examples
kombu
kombu.egg-info
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini
SLMQ.py
SQS
__init__.py
azureservicebus.py
azurestoragequeues.py
base.py
confluentkafka.py
consul.py
etcd.py
filesystem.py
gcpubsub.py
librabbitmq.py
memory.py
mongodb.py
native_delayed_delivery.py
pyamqp.py
pyro.py
qpid.py
redis.py
sqlalchemy
virtual
zookeeper.py

1440:    def register_with_event_loop(self, connection, loop):
1443:        cycle_poll_start = cycle.on_poll_start
1455:                    loop.on_tick.remove(on_poll_start)
1458:        cycle._on_connection_disconnect = _on_disconnect
1460:        def on_poll_start():
1463:        loop.on_tick.add(on_poll_start)

        # active channels
        self._channels = set()
        # file descriptor -> channel map.
        self._fd_to_chan = {}
        # channel -> socket map
        self._chan_to_sock = {}
        # poll implementation (epoll/kqueue/select)
        self.poller = poll()
        # one-shot callbacks called after reading from socket.
        self.after_read = set()

    def close(self):
        for fd in self._chan_to_sock.values():
            try:
                self.poller.unregister(fd)
            except (KeyError, ValueError):
                pass
        self._channels.cl
…[3641 chars omitted]…
     # Cancel stale timer entries from a previous connection before
        # registering new ones. Without this, each reconnect accumulates
        # an extra entry in hub.timer._queue; they all fire against the
        # same cycle and can crash the event loop during reconnect.
        for attr in ('_restore_messages_tref', '_subclient_health_tref'):
            old_tref = getattr(cycle, attr, None)

        if async_pool is not None:
            async_pool.disconnect()

    def _on_connection_disconnect(self, connection):
        if self._in_poll is connection:
            self._in_poll = None
        if self._in_listen is connection:
            self._in_listen = None
        if self.connection and self.connection.cycle:
            self.connection.cycle._on_connection_disconnect(connection)

    def _do_restore_message(self, payload, exchange, routing_key,
                            pipe, leftmost=False):
        try:
            try:
                payload['headers']['redelivered'] = True
                payload['
…[351 chars omitted]…
_called_with()
1217:        transport.cycle._on_connection_disconnect(self.connection)
1240:        on_poll_start = loop.on_tick.add.call_args[0][0]
1242:        on_poll_start()
1243:        transport.cycle.on_poll_start.assert_called_with()
1718:    def test_on_poll_start(self):
1721:        p.on_poll_start()
1729:        p.on_poll_start()
1735:        p.on_poll_start()
1741:        p.on_poll_start()

        assert redis.Channel._get_response_error(self.channel) is ResponseError

    def test_avail_client(self):
        self.channel._pool = Mock()
        cc = self.channel._create_client = Mock()
        with self.channel.conn_or_acquire():
            pass
        cc.assert_called_with()

    def test_register_with_event_loop(self):
        transport = self.connection.transport
        transport.cycle = Mock(name='cycle')
        transport
…[2938 chars omitted]…
out storing the trefs, stale timer entries accumulate in
        hub.timer._queue across reconnects (each reconnect adds a new entry
        without removing the old one).
        """
        transport = self.connection.transport
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = {}
        conn = Mock(name='conn')
        conn.client = Mock(name='client', transport_options={})

!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 62 passed in 1.22s

+    try:
+        fd = sock.fileno()
+    except (AttributeError, ValueError, OSError):
+        return None
+    if fd is None or fd < 0:
+        return None
+    return fd
+
+
+def _prune_stale_channels(fd_to_chan, connection):
+    """Remove stale ``fd_to_chan`` entries belonging to *connection*."""
+    try:
+        items = list(fd_to_chan.items())
+    except (AttributeError, TypeError):
+        return
+    for fd, entry in items:
+        try:
+
…[2481 chars omitted]…
     # NOTE: ``on_poll_start`` must stay registered on
+            # ``loop.on_tick`` so that polling resumes when the
+            # replacement connection is registered.
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():
ERROR t/unit/utils/test_json.py
!!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!!
9 skipped, 7 errors in 5.39s