celery-kombu-2498-2492

-rw-r--r-- 1 root root   5686 Nov  4 21:00 AUTHORS
-rw-r--r-- 1 root root 168725 Nov  4 21:00 Changelog.rst
-rw-r--r-- 1 root root    436 Nov  4 21:00 FAQ
-rw-r--r-- 1 root root    387 Nov  4 21:00 INSTALL
-rw-r--r-- 1 root root   1664 Nov  4 21:00 LICENSE
-rw-r--r-- 1 root root    5
…[952 chars omitted]…
s
./.git/refs
./.github
./.github/workflows
./docs
./docs/images
./docs/includes
./docs/reference
./docs/templates
./docs/userguide
./docs/_ext
./docs/_static
./docs/_templates
./examples
./examples/experimental
./examples/rpc-tut6
./examples/simple_task_queue
./kombu
./kombu/asynchronous
./kombu/transport
./kombu/utils
./kombu.egg-info
./requirements
./requirements/extras
./t
./t/integration
./t/unit

805:        if self.connection and self.connection.cycle:
806:            self.connection.cycle._on_connection_disconnect(connection)
851:        # Update fair cycle between queues.
853:      
…[894 chars omitted]…
tick.remove(on_poll_start)
1442:        cycle._on_connection_disconnect = _on_disconnect
1445:            cycle_poll_start()
1446:            [add_reader(fd, on_readable, fd) for fd in cycle.fds]
1447:        loop.on_tick.add(on_poll_start)
1448:        loop.call_repeatedly(10, cycle.maybe_restore_messages)
1455:            cycle.maybe_check_subclient_health
1460:        self.cycle.on_readable(fileno)

    @cached_property
    def unacked_mutex_expire(self):
        return self.channel.unacked_mutex_expire

    @cached_property
    def visibility_timeout(self):
        return self.channel.visibility_timeout


class MultiChannelPoller:
    """Async I/O poller for Redis transport."""

    eventflags = READ | ERR

    #: Set by :meth:`get` while reading from the socket.
    _in_protected_read = False

    #: Set of one-shot callbacks to call after reading from socket.
    after_read = None

    def __init__(self):
        # active channels
        self._channels = set()
        # file descriptor -> 
…[7344 chars omitted]…
ls.scheduling.round_robin_cycle`).
    #:
    #:    Make sure each queue has an equal opportunity to be consumed from.
    #:
    #: - ``sorted``
    #:   (:class:`~kombu.utils.scheduling.sorted_cycle`).
    #:
    #:    Consume from queues in alphabetical order.
    #:    If the first queue in the sorted list always contains messages,
    #:    then the rest of the queues will never be consumed from.

        self.ResponseError = self._get_response_error()
        self.active_fanout_queues = set()
        self.auto_delete_queues = set()
        self._fanout_to_queue = {}
        self.handlers = {'BRPOP': self._brpop_read, 'LISTEN': self._receive}
        self.brpop_timeout = self.connection.brpop_timeout

        if self.fanout_prefix:
            if isinstance(self.fanout_prefix, str):
                self.k
…[8628 chars omitted]…
is.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):
    """Channel with explicit Redis Sentinel knowledge.

    Broker url is supposed to look like:

port.register_with_event_loop(transport, conn, loop)
t/unit/transport/test_redis.py:1236:            call(10, transport.cycle.maybe_restore_messages),
t/unit/transport/test_redis.py:1237:            call(15, transport.cycle.maybe_check_subclient_health),
t/unit/transport/test_redis.py:1668:class test_MultiChannelPoller:
t/unit/transport/test_redis.py:1671:        self.Poller = redis.MultiChannelPoller

        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
…[8629 chars omitted]…
all_count == 3
        u_args = poller.unregister.call_args_list

        assert sorted(u_args) == [
            ((1,), {}),
            ((2,), {}),
            ((3,), {}),
        ]

    def test_close_when_unregister_raises_KeyError(self):
        p = self.Poller()
        p.poller = Mock()
        p._chan_to_sock.update({1: 1})
        p.poller.unregister.side_effect = KeyError(1)
        p.close()

                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


…[2834 chars omitted]…
ck.add.assert_called()
        on_poll_start = loop.on_tick.add.call_args[0][0]

        on_poll_start()
        transport.cycle.on_poll_start.assert_called_with()
        loop.add_reader.assert_has_calls([
            call(12, transport.on_readable, 12),
            call(13, transport.on_readable, 13),
        ])

    def test_transport_on_readable(self):
        transport = self.connection.transport

- Update msgpack to 1.1.0 (#2105)
- Update pytest to 8.3.3 (#2108)
- Prepare for release: v5.4.1 (#2109)

.. _version-5.4.0:

5.4.0
=====
:release-date: 6 August, 2024
:release-by: Tomer Nosrati

We want to add a special thanks to contribution `#2007 <https://github.com/celery/kombu/pull/2007>`_ by
@awmackowiak for fixing the Redis reconnection bug. Restoring Redis stability has been an essential improvement - thank you!

The rest of the changes are listed below.

Changes
-------
- fix: 
…[6131 chars omitted]…
fter reconnect - consumer starts consuming the tasks after crash. (#2007)
- Update flake8 to 7.1.0 (#2028)
- Add support for mongodb+srv scheme (#1976)

.. _version-5.3.7:

5.3.7
=====
:release-date: 11 April, 2024
:release-by: Tomer Nosrati

The release of v5.3.6 was missing the bumbversion commit so v5.3.7 is only released to sync it back.

.. _version-5.3.6:

5.3.6
=====
:release-date: 27 Mar, 2024

398:    def test_redis_on_disconnect_channel_only_if_was_registered(self):
399:        """Test should check if the _on_disconnect method is called only
409:        # for the possibility to run disconnect method
411:            def disconnect(self, *args):
427:            def _on_connection_disconnect(self, connection):
450:        # than disconnect the Redis connection
451:        r
…[3737 chars omitted]…
l, channel.client, 'BRPOP')] = True
1822:        channel.subclient.connection._sock = None
1827:        p._register.assert_called_with(channel, channel.subclient, 'LISTEN')
1832:        p._chan_to_sock[(channel, channel.subclient, 'LISTEN')] = 3
1833:        channel.subclient.connection._sock = Mock()
1904:        p._fd_to_chan[1] = (channel, 'BRPOP')
1914:        p._fd_to_chan[1] = (channel, 'BRPOP')

        assert cmd == type
        self.connection._sock.data = []
        if type == 'BRPOP':
            timeout = queues.pop()
            item = self.brpop(queues, timeout)
            if item:
                return item
            raise Empty()

    def brpop(self, keys, timeout=None):
        for key in keys:
            try:
                item = self.queues[key].get_nowait()
            except Empty:
                pass
            else:
                return key, item

    def rpop(self, key):
        try:
            return s
…[6688 chars omitted]…
ient.lock.return_value = Mock(name='lock')

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

        client.reset_mock()
        lock.reset_mock()

        # Did not win

/usr/local/lib/python312.zip []
/usr/local/lib/python3.12 []
/usr/local/lib/python3.12/lib-dynload []
/usr/local/lib/python3.12/site-packages []

    def _close_poller(self):
        if self._poller is not None:
            self._poller.close()
            self._poller = None
            self._register_fd = None
            self._unregister_fd = None

    def stop(self):
        self.call_soon(_raise_stop_error)

    def __repr__(self):
        return '<Hub@{:#x}: R:{} W:{}>'.format(
            id(self), len(self.readers), len(se
…[3335 chars omitted]…
f _unregister(self, fd):
        try:
            self.poller.unregister(fd)
        except (AttributeError, KeyError, OSError):
            pass

    def _pop_ready(self):
        with self._ready_lock:
            ready = self._ready
            self._ready = set()
            return ready

    def close(self, *args):
        [self._unregister(fd) for fd in self.readers]
        self.readers.clear()

import sys
from collections import namedtuple
from datetime import datetime
from functools import total_ordering
from time import monotonic
from time import time as _time
from typing import TYPE_CHECKING
from weakref import proxy as weakrefproxy

from vine.utils import wraps

from kombu.log import get_logger

if sys.version_info >= (3, 9):
    from zoneinfo import ZoneInfo
else:
    from backports.zoneinfo import ZoneInfo

if TYPE_CHECKING:
    from types import TracebackType

__all__ = ('Entry', 'Timer', 'to_timestamp')

logger = get_logger(__name__)

DEFAULT_MAX_INT
…[3090 chars omitted]…
cancel(self, tref):
        tref.cancel()

    def __len__(self):
        return len(self._queue)

    def __nonzero__(self):
        return True

    @property
    def queue(self, _pop=heapq.heappop):
        """Snapshot of underlying datastructure."""
        events = list(self._queue)
        return [_pop(v) for v in [events] * len(events)]

    @property
    def schedule(self):
        return self

/root/.cache/pip/http-v2/4/4/e/5/b/44e5b11a6caa92636d8ccfe658d420ba4ed8f67f7f4e835b214255aa
/root/.cache/pip/http-v2/4/4/e/5/b/44e5b11a6caa92636d8ccfe658d420ba4ed8f67f7f4e835b214255aa.body
/root/.cache/pip/http-v2/3/b/0/c/5/3b0c5b0aec5aaa79a8ac084ba31087d7fff942582228659a50dd43af.body
/root/.cache/pip/http-v2/3/b/0/c/5/3b0c5b0aec5aaa79a8ac084ba31087d7fff942582228659a50dd43af
pip

            # credential provider as query string
            credential_provider = query.pop("credential_provider", None)
            self._process_credential_provider(credential_provider, connparams)

            connparams.pop('host', None)
            connparams.pop('port', None)
        connparams['db'] = self._prepare_virtual_host(
            
…[1378 chars omitted]…
onnection_disconnect(self, connection):
800:    def _on_connection_disconnect(self, connection):
806:            self.connection.cycle._on_connection_disconnect(connection)
1308:            class Connection(connection_cls):
1309:                def disconnect(self, *args):
1314:                        channel._on_connection_disconnect(self)
1442:        cycle._on_connection_disconnect = _on_disconnect

import pytest

from kombu import Connection, Consumer, Exchange, Producer, Queue
from kombu.exceptions import VersionMismatch
from kombu.transport import virtual
from kombu.utils import eventio  # patch poll
from kombu.utils.json import dumps

if TYPE_CHECKING:
    from types import TracebackType


def _redis_modules():

    class ConnectionError(Exception):
        pass

    class AuthenticationError(Exception):
        pass

    class InvalidData(Exception):
        pass

    class InvalidResponse(Exception):
        pass

    class ResponseError(Exception):
        pass

    exceptions = types
…[7179 chars omitted]…
      ) as conn:
            params = conn.default_channel._connparams()
            assert params['client_name'] == 'kombu-worker'

    def test_rotate_cycle_ValueError(self):
        cycle = self.channel._queue_cycle
        cycle.update(['kramer', 'jerry'])
        cycle.rotate('kramer')
        assert cycle.items, ['jerry' == 'kramer']
        cycle.rotate('elaine')

    def test_get_client(self):

            assert self.channel._receive_one(self.channel.subclient) is None

    def test_receive_connection_has_gone(self):
        def _receive_one(c):
            c.connection = None
            _receive_one.called = True
            return True

        _receive_one.called = False
        self.channel._receive_one = _receive_one

        assert self.channel._receive()
        assert _receive_one.called

    def test_brpop_read_raises(self):

…[1530 chars omitted]…

        msg1 = {'properties': {'priority': 3}}

        self.channel._put('george', msg1)
        client().lpush.assert_called_with(
            self.channel._q_for_pri('george', 3), dumps(msg1),
        )

        msg2 = {'properties': {'priority': 313}}
        self.channel._put('george', msg2)
        client().lpush.assert_called_with(
            self.channel._q_for_pri('george', 9), dumps(msg2),

    def test_maybe_update_queues_expire(self):
        with Connection(transport=Transport) as conn:
            channel = conn.channel()
            channel._expires = {'test_queue': 5000}

            client_mock = Mock()
            pipeline_mock = Mock()
            pipeline_mock.__enter__ = lambda self: pipeline_mock
            pipeline_mock.__exit__ = lambda self, *args: None
            client_mock.pipeline.return_value = pi
…[1063 chars omitted]…
    assert self.queue(channel).get() is None
        assert self.queue(channel).get() is None
        assert self.queue(channel).get() is None

    @pytest.mark.replace_module_value(redis.redis, 'VERSION', [2, 5, 10])
    def test_publish__get_redispyv2(self, replace_module_value):
        channel = self.connection.channel()
        producer = Producer(channel, self.exchange, routing_key='test_Redis')

        self.connection = self.create_connection()
        self.channel = self.connection.default_channel

    def create_connection(self, **kwargs):
        kwargs.setdefault('transport_options', {'fanout_patterns': True})
        return Connection(transport=Transport, **kwargs)

    def _get_one_delivery_tag(self, n='test_uniq_tag'):
        with self.create_connection() as conn1:
            chan = conn1.default_channel
            chan.exchange_declare(n)
            chan.queue_declare(n)
            chan.queue_bind(n, n, n)
            msg = chan.prepare_message('quick brown fox')
            chan.basic_publish(msg, n, n)
            payload = chan._get(n)
            a
…[268 chars omitted]…
t in seen
            seen.add(tag)
            with pytest.raises(ValueError):
                int(tag)
            assert len(tag) == 36

    def test_disable_ack_emulation(self):
        conn = Connection(transport=Transport, transport_options={
            'ack_emulation': False,
        })

        chan = conn.channel()
        assert not chan.ack_emulation
        assert chan.QoS == virtual.QoS

drwxr-xr-x 1 root root 4096 Nov  4 21:01 ..
-rw-r--r-- 1 root root    0 Nov  4 21:02 install.log
/repo/conftest.py
/repo/requirements/funtest.txt
/repo/requirements/test-ci.txt
/repo/requirements/test-integration.txt
/repo/requirements/test.txt

t/unit/transport/test_redis.py:1235:        loop.call_repeatedly.assert_has_calls([
t/unit/transport/test_redis.py:1239:        loop.on_tick.add.assert_called()
t/unit/transport/test_redis.py:1240:        on_poll_start = loop.on_tick.add.call_args[0][0]

        if os.getpid() == self.pid:
            try:
                conn_sock.shutdown(socket.SHUT_RDWR)
            except (OSError, TypeError):
                pass

        try:
            conn_sock.close()
        except OSError:
            pass

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`
…[1568 chars omitted]…
.utils.json import dumps, loads
from kombu.utils.objects import cached_property
from kombu.utils.scheduling import cycle_by_name
from kombu.utils.url import _parse_url

from . import virtual

try:
    import redis
    _REDIS_GET_CONNECTION_WITHOUT_ARGS = Version(version("redis")) >= Version("5.3.0")
except ImportError:  # pragma: no cover
    redis = None
    _REDIS_GET_CONNECTION_WITHOUT_ARGS = None

+    """Return the file descriptor for ``sock`` if it is still valid."""
+    if isinstance(sock, int):
+        return sock
+    try:
+        fd = sock.fileno()
+    except Exception:
+        return None
+    try:
+        return fd if fd >= 0 else None
+    except TypeError:
+        return None
+
+
+def _client_uses_connection(client, connection):
+    if client is connection:
+        return True
+    try:
+        return getattr(client, 'connection', None) is connection
+
…[7884 chars omitted]…
val',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
-            health_check_interval,
-            cycle.maybe_check_subclient_health
+        self._timer_handles.append(
+            loop.call_repeatedly(
+                health_check_interval,
+                cycle.maybe_check_subclient_health,
+            ),
         )
 
     def on_readable(self, fileno):

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x75e4ce080540>} = <Mock name='loop' id='129625569939664'>.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, 134 passed in 5.32s

258-    def close(self, *args):
259-        [self._unregister(fd) for fd in self.readers]
260-        self.readers.clear()
261-        [self._unregister(fd) for fd in self.writers]
--
281:    def _discard(self, fd):
282-        fd = fileno(fd)
283-        self.readers.pop(fd, None)
284-        self.writers.pop(fd, None)
285-        self.consolidate.discard(fd)
286-
287-    def on_callback_error(self, callback, exc):
288-        logger.error(
289-            'Callback %r raised exception: %r', callback, exc, exc_info=1,
290-        )
291-
292-    def create_loop(self,
293-                    generator=generator, sleep=sleep, min=min, next=next,
294-                    Empty=Empty, StopIteration=StopIteration,
295-                    KeyError=KeyError, READ=READ, WRITE=WRITE, ERR=ERR):
296-        readers, writers = self.readers, self.writers

    ???
/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 Bas
…[2024 chars omitted]…
n.py
ERROR t/unit/asynchronous/aws/test_aws.py
ERROR t/unit/asynchronous/aws/test_connection.py
ERROR t/unit/transport/SQS - ModuleNotFoundError: No module named 'botocore'
ERROR t/unit/transport/test_azurestoragequeues.py
ERROR t/unit/transport/test_gcpubsub.py
ERROR t/unit/utils/test_json.py
!!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!!
9 skipped, 7 errors in 3.86s

+    try:
+        fd = sock.fileno()
+    except Exception:
+        return None
+    try:
+        return fd if fd >= 0 else None
+    except TypeError:
+        return None
+
+
+def _client_uses_connection(client, connection):
+    if client is connection:
+        return True
+    try:
+        return getattr(client, 'connection', None) is connection
+    except Exception:
+        return False
+
+
+def _channel_uses_connection(channel, connection):
+    for attr in ('client', 'subclie
…[7895 chars omitted]…
val',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
-            health_check_interval,
-            cycle.maybe_check_subclient_health
+        self._timer_handles.append(
+            loop.call_repeatedly(
+                health_check_interval,
+                cycle.maybe_check_subclient_health,
+            ),
         )
 
     def on_readable(self, fileno):

        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
…[86 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x783a2e2885e0>} = <Mock name='loop' id='132191278037280'>.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, 134 passed in 4.84s

4
6
7
8
9
a

test_mongodb.py
test_native_delayed_delivery.py
test_pyamqp.py
test_pyro.py
test_qpid.py
test_redis.py
test_sqlalchemy.py
test_transport.py
test_zookeeper.py
virtual
            ['server1', 'server2'],
            transport=TimeoutingTransport,
            connect_timeout=1,
            transport_options={'connect_retries_timeout': 2,
                               'interval_start': 0, 'interval_step': 0},
        ) as conn:
            conn._establish_connection = Mock(
                side_effect=conn._establish_connection
            )
            with pytest.raises(OperationalError):
                conn.default_channel
>           assert conn._establish_connection.call_count == 2
E           
…[117 chars omitted]…
e <Mock id='132152835540272'> = <Connection: <class 't.mocks.TimeoutingTransport'>://:**@server1// at 0x78313ad014f0>._establish_connection

t/unit/test_connection.py:803: AssertionError
=========================== short test summary info ============================
FAILED t/unit/test_connection.py::test_Connection::test_connection_failover_with_total_timeout
1 failed, 157 passed, 5 skipped in 10.86s

def _safe_getattr(obj, name, default=None):
    try:
        return getattr(obj, name, default)
    except Exception:
        return default


def _socket_fd(sock):
    """Return the file descriptor for ``sock`` if it is still valid."""
    if isinstance(sock, int):
        return sock
    try:
        fd = sock.fileno()
    except Exception:
        return None
    try:
        return fd if fd >= 0 else None
    except TypeError:
        return None


def _client_uses_connection(client, connection):
    if client is connection:
        return True
    try:
        return getattr(client, 'connection
…[8629 chars omitted]…
 loop.call_repeatedly(10, cycle.maybe_restore_messages),
        )
        health_check_interval = connection.client.transport_options.get(
            'health_check_interval',
            DEFAULT_HEALTH_CHECK_INTERVAL
        )
        self._timer_handles.append(
            loop.call_repeatedly(
                health_check_interval,
                cycle.maybe_check_subclient_health,
            ),

        # 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.clear()
        self._fd_to_chan.clear()
        self._chan_to_sock.clear()

    def add(self, channel):
        self._channels.add(channel)

    def discard(self, channel):
        self._channels
…[856 chars omitted]…
dles.append(
            loop.call_repeatedly(
                health_check_interval,
                cycle.maybe_check_subclient_health,
            ),
        )

    def on_readable(self, fileno):
        """Handle AIO event for one of our file descriptors."""
        self.cycle.on_readable(fileno)


if sentinel:
    class SentinelManagedSSLConnection(
            sentinel.SentinelManagedConnection,

                pass
            fd = _socket_fd(sock)
            if fd is not None:
                try:
                    fd_to_chan.pop(fd, None)
                except Exception:
                    pass
        else:
            _discard_connection_fds(fd_to_chan, connection)
        _discard_connection_socks(
            getattr(self, '_chan_to_sock', None), connection, sock,
        )

    def _register(self, channel, client, type):
        if (channel, client, type) in self._chan_to_sock:
    
…[86 chars omitted]…
None:   # not connected yet.
            client.connection.connect()
        sock = client.connection._sock
        self._fd_to_chan[sock.fileno()] = (channel, type)
        self._chan_to_sock[(channel, client, type)] = sock
        self.poller.register(sock, self.eventflags)

    def _unregister(self, channel, client, type):
        self.poller.unregister(self._chan_to_sock[(channel, client, type)])

    lock = client.lock(name, timeout=expire)
           ^^^^^^^^^^^
AttributeError: 'Client' object has no attribute 'lock'

/redis.py:1579:        for timer in old_timers:
kombu/transport/redis.py:1584:        self._timer_handles = []
kombu/transport/redis.py:1629:        self._timer_handles.append(
kombu/transport/redis.py:1630:            loop.call_repeatedly(10, cycle.maybe_restore_messages),
kombu/transport/redis.py:1636:        self._timer_handles.append(
kombu/transport/redis.py:1637:            loop.call_repeatedly(

t/unit/asynchronous/aws/test_connection.py:11: in <module>
    from kombu.asynchronous.aws.connection import (AsyncAWSQueryConnection,
kombu/asynchronous/aws/__init__.py:5
…[171 chars omitted]…
import Serializer
E   ModuleNotFoundError: No module named 'botocore'
=========================== short test summary info ============================
ERROR t/unit/asynchronous/aws/sqs/test_connection.py
ERROR t/unit/asynchronous/aws/test_aws.py
ERROR t/unit/asynchronous/aws/test_connection.py
!!!!!!!!!!!!!!!!!!! Interrupted: 3 errors during collection !!!!!!!!!!!!!!!!!!!!
2 skipped, 3 errors in 0.96s

kombu/utils/eventio.py:66: OSError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0]
FAILED t/unit/transport/test_redis.py::test_Redis::test_publish__consume - OS...
2 failed, 143 passed in 4.97s

+        return default
+
+
+def _socket_fd(sock):
+    """Return the file descriptor for ``sock`` if it is still valid."""
+    if isinstance(sock, int):
+        return sock
+    try:
+        fd = sock.fileno()
+    except Exception:
+        return None
+    try:
+        return fd if fd >= 0 else None
+    except TypeError:
+        return None
+
+
+def _client_uses_connection(client, connection):
+    if client is connection:
+        return True
+    try:
+        return getattr(client, 'connection', None) i
…[8628 chars omitted]…
val',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
-            health_check_interval,
-            cycle.maybe_check_subclient_health
+        self._timer_handles.append(
+            loop.call_repeatedly(
+                health_check_interval,
+                cycle.maybe_check_subclient_health,
+            ),
         )
 
     def on_readable(self, fileno):

-rw-r--r-- 1 root root    616 Nov  2 05:36 .readthedocs.yaml
-rw-r--r-- 1 root root   5686 Nov  2 05:36 AUTHORS
-rw-r--r-- 1 root root 168725 Nov  2 05:36 Changelog.rst
-rw-r--r-- 1 root root    436 Nov  2 05:36 FAQ
-rw-r--r-- 1 root root    387 Nov  2 05:36 INSTALL
-rw-r--r-- 1 root root   1664 Nov  2 05:36 LICENSE
-rw-r--r-- 1 root roo
…[2163 chars omitted]…
-info/top_level.txt
./LICENSE
./Makefile
./MANIFEST.in
./pyproject.toml
./README.rst
./requirements/default.txt
./requirements/dev.txt
./requirements/docs.txt
./requirements/funtest.txt
./requirements/pkgutils.txt
./requirements/test-ci.txt
./requirements/test-integration.txt
./requirements/test.txt
./SECURITY.md
./setup.cfg
./setup.py
./t/mocks.py
./t/skip.py
./t/__init__.py
./THANKS
./TODO
./tox.ini

ort.cycle.maybe_check_subclient_health),
t/unit/transport/test_redis.py:1217:        transport.cycle._on_connection_disconnect(self.connection)
t/unit/transport/test_redis.py:1236:            call(10, transport.cycle.maybe_restore_messages),
t/unit/transport/test_redis.py:1237:            call(15, transport.cycle.maybe_check_subclient_health),
t/integration/test_redis.py
t/unit/transport/test_redis.py

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`
…[8630 chars omitted]…
nsport, 13)
        cycle.on_readable.assert_called_with(13)

    def test_transport_connection_errors(self):
        """Ensure connection_errors are populated."""
        assert redis.Transport.connection_errors

    def test_transport_channel_errors(self):
        """Ensure connection_errors are populated."""
        assert redis.Transport.channel_errors

    def test_transport_driver_version(self):

        with self.channel.conn_or_acquire(client) as client:
            for tag in self._delivered:
                self.restore_by_tag(tag, client=client)
        self._delivered.clear()

    def ack(self, delivery_tag):
        self._remove_from_indices(delivery_tag).execute()
        super().ack(delivery_tag)

    def reject(self, delivery_tag, requeue=False):
        if requeue:
            self.restore_by_tag(delivery_tag, leftmost=True)
  
…[8630 chars omitted]…
f call_repeatedly(self, delay, callback, *args):
kombu/asynchronous/hub.py:218:        return self.timer.call_repeatedly(delay, callback, args)
kombu/asynchronous/timer.py:126:    def call_repeatedly(self, secs, fun, args=(), kwargs=None, priority=0):
kombu/transport/redis.py:1448:        loop.call_repeatedly(10, cycle.maybe_restore_messages)
kombu/transport/redis.py:1453:        loop.call_repeatedly(

    def discard(self, channel):
        self._channels.discard(channel)

    def _on_connection_disconnect(self, connection):
        try:
            self.poller.unregister(connection._sock)
        except (AttributeError, TypeError):
            pass

    def _register(self, channel, client, type):
        if (channel, client, type) in self._chan_to_sock:
            self._unregister(channel, client, type)
        if client.connection._sock is None:   # not connected yet.
            client.connection.connect()
        sock = client
…[8629 chars omitted]…
ed.
        """
        if eta is None:
            eta = time()
        if isinstance(eta, datetime):
            try:
kombu/asynchronous/hub.py:83:        self.on_tick = set()
kombu/asynchronous/hub.py:226:    def remove_reader(self, fd):
kombu/asynchronous/hub.py:314:            for tick_callback in copy(self.on_tick):
kombu/asynchronous/hub.py:338:                            self.remove_reader(fd)

        try:
            self.poller.unregister(connection._sock)
        except (AttributeError, TypeError):
            pass

    def _register(self, channel, client, type):
        if (channel, client, type) in self._chan_to_sock:
            self._unregister(channel, client, type)
        if client.connection._sock is None:   # not connected yet.
            client.connection.connect()
        sock = client.connection._sock
        self._fd_to_chan[sock.fileno()] = (channel, type)
        self._chan_to_sock[(channel, client, type)] = sock
        self.poller.register(sock, s
…[8629 chars omitted]…
try:
                            cb, cbargs = writers[fd]
                        except KeyError:
                            self.remove_writer(fd)
                            continue
                    elif event & ERR:
                        general_error = True
                    else:
                        logger.info(W_UNKNOWN_EVENT, event, fd)
                        general_error = True

    _in_listen = False
    _fanout_queues = {}
    ack_emulation = True
    unacked_key = 'unacked'
    unacked_index_key = 'unacked_index'
    unacked_mutex_key = 'unacked_mutex'
    unacked_mutex_expire = 300  # 5 minutes
    unacked_restore_limit = None
    visibility_timeout = 3600   # 1 hour
    priority_steps = PRIORITY_STEPS
    socket_timeout = None
    socket_connect_timeout = None
    socket_keepalive = None
    socket_keepalive_options = None
    retry_on_timeout = None
    max_connections = 10
    health_check_interval = DEFAULT_HEALTH_CHECK_INTERVAL
    client_name = None
    #: Transport option to disable fanout keyprefix.
    #:
…[8628 chars omitted]…
 DEFAULT_HEALTH_CHECK_INTERVAL
        )
        loop.call_repeatedly(
            health_check_interval,
            cycle.maybe_check_subclient_health
        )

    def on_readable(self, fileno):
        """Handle AIO event for one of our file descriptors."""
        self.cycle.on_readable(fileno)


if sentinel:
    class SentinelManagedSSLConnection(
            sentinel.SentinelManagedConnection,

1668:class test_MultiChannelPoller:
1671:        self.Poller = redis.MultiChannelPoller

    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", return_value=True):
            chan = conn.channel()

        assert conn.transport.brpop_timeout == 2
        assert chan.brpop_timeout == 2
        assert
…[8629 chars omitted]…
nsport/redis.py:148:        (virtual.Transport.connection_errors + (
kombu/transport/redis.py:780:        self.connection_errors = self.connection.connection_errors
kombu/transport/redis.py:954:        except self.connection_errors:
kombu/transport/redis.py:997:            except self.connection_errors:
kombu/transport/redis.py:1408:        connection_errors, channel_errors = get_redis_error_classes()

    def _subscribe(self):
        keys = [self._get_subscribe_topic(queue)
                for queue in self.active_fanout_queues]
        if not keys:
            return
        c = self.subclient
        if c.connection._sock is None:
            c.connection.connect()
        self._in_listen = c.connection
        c.psubscribe(keys)

    def _unsubscribe_from(self, que
…[8628 chars omitted]…
 cancelled(self):
91:    def cancelled(self, value):
105:        self._queue = []
172:        push(self._queue, scheduled(eta, priority, entry))
187:    def stop(self):
199:        queue = self._queue
221:    def clear(self):
222:        self._queue[:] = []  # atomic, without creating a new list.
224:    def cancel(self, tref):
228:        return len(self._queue)
236:        events = list(self._queue)

kombu/transport/redis.py:1455:        
…[8629 chars omitted]…
      return self.transport.implements.heartbeats

    @property
    def is_evented(self):
        return self.transport.implements.asynchronous


BrokerConnection = Connection


class PooledConnection(Connection):
    """Wraps :class:`kombu.Connection`.

    This wrapper modifies :meth:`kombu.Connection.__exit__` to close the connection
    in case any exception occurred while the context was active.

                    fun()

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


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

    QoS = QoS

    _client = None
    _subclient = None

    def test_register_BRPOP(self):
        p = self.Poller()
        channel = Mock()
        channel.client.connection._sock = None
        p._register = Mock()

        channel._in_poll = False
        p._register_BRPOP(channel)
        assert channel._brpop_start.call_
…[2974 chars omitted]…
un_forever(self):
        self._running = True
        try:
            while 1:
                try:
                    self.run_once()
                except Stop:
                    break
        finally:
517:    def _register(self, channel, client, type):
527:    def _unregister(self, channel, client, type):
539:    def _register_BRPOP(self, channel):
548:    def _register_LISTEN(self, channel):

                # table does not exists since all queues bound to the exchange
                # were deleted. We need just return empty list.
                return []
            return [tuple(bytes_to_str(val).split(self.sep)) for val in values]

    def _purge(self, queue):
        with self.conn_or_acquire() as client:
            with client.pipeline() as pipe:
                for pri in self.pri
…[5301 chars omitted]…
Redis,
                global_keyprefix=self.global_keyprefix,
530:    def _client_registered(self, channel, client, cmd):
542:        if not self._client_registered(channel, channel.client, 'BRPOP'):
550:        if not self._client_registered(channel, channel.subclient, 'LISTEN'):
748:        self._registered = False
776:        self._registered = True
1313:                    if channel._registered:

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        if not self.ack_emulation:  # disable visibility timeout
            self.QoS = virtual.QoS
        self._registered = False
        self._expires = {}
        self._queue_cycle = cycle_by_name(self.queue_order_strategy)()
        self.Client = self._get_client()
        self.ResponseError = self._get_response_error()
        self.active_fanout_queues = set()
        self.auto_delete_queues = set()
…[1121 chars omitted]…
onnect_pools()

    def _disconnect_pools(self):
        pool = self._pool
        async_pool = self._async_pool
t/unit/transport/test_redis.py:1191:            call(10, transport.cycle.maybe_restore_messages),
t/unit/transport/test_redis.py:1236:            call(10, transport.cycle.maybe_restore_messages),
grep: t/unit/transport/__pycache__/test_redis.cpython-312-pytest-8.4.2.pyc: binary file matches

                if not self.handle_error(exc):
                    raise
                return
        return self._enter(eta, priority, entry)

    def enter_after(self, secs, entry, priority=0, time=monotonic):
        return self.enter_at(entry, time() + float(secs), priority)

    def _enter(self, eta, priority, entry, push=heapq.heappush):
        push(self._queue, scheduled(eta, priority, entry))
        return entry

    def apply_entry(self, entry):
        try:
            entry()
        excep
…[2636 chars omitted]…
ack, flags, args=(), consolidate=False):
        fd = fileno(fd)
        try:
            self.poller.register(fd, flags)
        except ValueError:
            self._remove_from_loop(fd)
            raise
        else:
            dest = self.readers if flags & READ else self.writers
            if consolidate:
                self.consolidate.add(fd)
                dest[fd] = None
            else:

596:            chan, type = self._fd_to_chan[fileno]
631:        return self._fd_to_chan

    )


def fileno(f):
    """Get fileno from file-like object."""
    if isinstance(f, numbers.Integral):
        return f
    return f.fileno()


def maybe_fileno(f):
    """Get object fileno, or :const:`None` if not defined."""
    try:
        return fileno(f)
    except FILENO_ERRORS:
        pass


@contextmanager
495:    def close(self):
1167:    def close(self):
1188:    def _close_clients(self):

        pool = Mock(name='pool')
        client = Mock(
            name='client',
            ping=Mock(return_value=True)
        )

        # create RedisConnectionMock class
        # for the possibility to run disconnect method
        class RedisConnectionMock:
            def disconnect(self, *args):
                pas
…[5333 chars omitted]…
ed_with()
1205:    def test_register_with_event_loop__on_disconnect__loop_cleanup(self, fds):
1206:        """Ensure event loop polling stops on disconnect (if started)."""
1217:        transport.cycle._on_connection_disconnect(self.connection)
1619:    def test_close_disconnects(self):
1624:        assert conn1.disconnected
1625:        assert conn2.disconnected
1633:        assert conn1.disconnected

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x75774f569940>} = <Mock name='loop' id='129155293445056'>.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, 134 passed in 4.90s

+    except (TypeError, AttributeError):
+        namespace = {}
+    if name in namespace:
+        return namespace[name]
+
+    # Channel.client and Channel.subclient are lazy cached properties.  Do not
+    # create clients merely to inspect which connection a poller entry uses.
+    descriptor = getattr(type(channel), name, None)
+    if descriptor is None or isinstance(descriptor, cached_property):
+        return None
+    try:
+        # In particular, this ha
…[7991 chars omitted]…
h_check_interval',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
+        timer_handles.append(loop.call_repeatedly(
             health_check_interval,
             cycle.maybe_check_subclient_health
-        )
+        ))
 
     def on_readable(self, fileno):
         """Handle AIO event for one of our file descriptors."""
periodic maintenance reproducer passed

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x7f9656e99940>} = <Mock name='loop' id='140283688404944'>.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, 134 passed in 4.88s

        namespace = {}
    if name in namespace:
        return namespace[name]

    # Channel.client and Channel.subclient are lazy cached properties.  Do not
    # create clients merely to inspect which connection a poller entry uses.
    descriptor = getattr(type(channel), name, None)
    if descriptor is None or isinstance(descriptor, cached_property):
        return None
    try:
        # In particular, this handles attributes stored in __slots__.
        return getattr(channel, name)
    except Exception:
        return Non
…[8629 chars omitted]…
on:
                        continue
                    try:
                        loop.remove(fd)
                    except Exception:
                        pass
                    try:
                        fds = cycle.fds
                    except Exception:
                        fds = None
                    _discard_descriptor(fds, fd)
                    if registrations is not fds:

                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)
        self._event_loop = loop
        self._event
…[3536 chars omitted]…
,)
        elif not isinstance(errors, tuple):
            try:
                errors = tuple(errors)
            except TypeError:
                errors = ()
        try:
            return isinstance(exc, errors)
        except TypeError:
            return False

    def maybe_restore_messages(self):
        for channel in self._channels:
            if channel.active_queues:
                try:

 kombu/transport/redis.py | 263 +++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 241 insertions(+), 22 deletions(-)
 M kombu/transport/redis.py

+    try:
+        namespace = vars(channel)
+    except (TypeError, AttributeError):
+        namespace = {}
+    if name in namespace:
+        return namespace[name]
+
+    # Channel.client and Channel.subclient are lazy cached properties.  Do not
+    # create clients merely to inspect which connection a poller entry uses.
+    descriptor = getattr(type(channel), name, None)
+ 
…[8629 chars omitted]…
nsport_options.get(
             'health_check_interval',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
+        timer_handles.append(loop.call_repeatedly(
             health_check_interval,
             cycle.maybe_check_subclient_health
-        )
+        ))
 
     def on_readable(self, fileno):
         """Handle AIO event for one of our file descriptors."""

                    try:
                        loop.remove(sock)
                    except Exception:
                        pass
                    fd = _descriptor_for_socket(sock)
                    if fd is not None:
                        try:
                            fds = cycle.fds
                        except Exception:
                            fds = None
  
…[2221 chars omitted]…
ked while the transport is already
                # recovering from a connection loss; cleanup must never make
                # that recovery fail as well.
                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)

                    except Exception:
                        pass
                    fd = _descriptor_for_socket(sock)
                    if fd is not None:
                        try:
                            fds = cycle.fds
                        except Exception:
                            fds = None
                        try:
                            registrations = cy
…[2035 chars omitted]…
sockets(
                        cycle, disconnected_connection, fd)
            except Exception:
                # This callback is invoked while the transport is already
                # recovering from a connection loss; cleanup must never make
                # that recovery fail as well.
                pass

        cycle._on_connection_disconnect = _on_disconnect

        def on_poll_start():

        namespace = {}
    if name in namespace:
        return namespace[name]

    # Channel.client and Channel.subclient are lazy cached properties.  Do not
    # create clients merely to inspect which connection a poller entry uses.
    descriptor = getattr(type(channel), name, None)
    if descriptor is None or isinstance(descriptor, cached_property):
        return None
    try:
        # In particular, this handles attributes stored in __slots__.
        return getattr(channel, name)
    except Exception:
        return Non
…[8628 chars omitted]…
es
        timer_handles.append(
            loop.call_repeatedly(10, cycle.maybe_restore_messages)
        )
        health_check_interval = connection.client.transport_options.get(
            'health_check_interval',
            DEFAULT_HEALTH_CHECK_INTERVAL
        )
        timer_handles.append(loop.call_repeatedly(
            health_check_interval,
            cycle.maybe_check_subclient_health

                    except Exception:
                        pass
                    if fd is not None:
                        try:
                            fds = cycle.fds
                        except Exception:
                            fds = None
                        try:
                            registrations = cycle._fd_to_chan
                        except Excep
…[187 chars omitted]…
  _discard_descriptor(registrations, fd)
                        _discard_channel_sockets(
                            cycle, disconnected_connection, fd)
                    return

                # Redis may clear _sock before notifying the channel.  In that
                # case use the channel/client association to find only this
                # connection's remaining descriptor registrations.

        namespace = {}
    if name in namespace:
        return namespace[name]

    descriptor = getattr(type(channel), name, None)
    if descriptor is None or isinstance(descriptor, cached_property):
        return None
    try:
        # Slots-only channel objects use descriptors rather than __dict__.
        return getattr(channel, name)
    except Exception:
        return None


def _client_uses_connection(client, connection):
    if client is connection:
        return True
    try:
        return getattr(client, 'connecti
…[5900 chars omitted]…
dy
                # recovering from a connection loss; cleanup must never make
                # that recovery fail as well.
                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)
        self._event_loop = loop

+    except (TypeError, AttributeError):
+        namespace = {}
+    if name in namespace:
+        return namespace[name]
+
+    descriptor = getattr(type(channel), name, None)
+    if descriptor is None or isinstance(descriptor, cached_property):
+        return None
+    try:
+        # Slots-only channel objects use descriptors rather than __dict__.
+        return getattr(channel, name)
+    except Exception:
+        return None
+
+
+def _client_uses_connectio
…[8629 chars omitted]…
nsport_options.get(
             'health_check_interval',
             DEFAULT_HEALTH_CHECK_INTERVAL
         )
-        loop.call_repeatedly(
+        timer_handles.append(loop.call_repeatedly(
             health_check_interval,
             cycle.maybe_check_subclient_health
-        )
+        ))
 
     def on_readable(self, fileno):
         """Handle AIO event for one of our file descriptors."""

    def client(self):
        """Client used to publish messages, BRPOP etc."""
        return self._create_client(asynchronous=True)

    @cached_property
    def subclient(self):
        """Pub/Sub connection used to consume fanout queues."""
        client = self._create_client(asynchronous=True)
        return client.pubsub()

    def _update_queue_cycle(self):
        self._queue_cycle.update(self.active_queues)

    def _get_response_error(self):
        from redis import exceptions
        return exceptions.ResponseError

    @property
    def active_queues(self):
   
…[5100 chars omitted]…
ked while the transport is already
                # recovering from a connection loss; cleanup must never make
                # that recovery fail as well.
                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)

            )

    @staticmethod
    def _is_connection_error(channel, exc):
        try:
            errors = getattr(channel, 'connection_errors', ())
        except Exception:
            return False
        if isinstance(errors, type):
            errors = (errors,)
        elif not isinstance(errors, tuple):
            try:
                errors = tuple(errors)
            ex
…[816 chars omitted]…
allable(check_health):
                    check_health()
            except Exception as exc:
                if not self._is_connection_error(channel, exc):
                    raise

    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:

        if callable(discard):
            discard(fd)
        else:
            registrations.pop(fd, None)
    except Exception:
        # The event loop may already have forgotten this descriptor.
        pass


def _discard_channel_sockets(cycle, connection, sock=None):
    """Drop concrete poller socket references for a disconnected client."""
    if not isinstance(cycle, MultiChannelPoller):
        return
    try:
        registrations = cycle._chan_to_sock
        entries = list(registrations.items())
    except Exception:
        return
    for key, registered_sock in entries:
        try:
            _, client, _ = key
        except (TypeError, ValueError):
            continue
        if sock is None:

    try:
        fds = cycle.fds
    except Exception:
        fds = None
    try:
        registrations = cycle._fd_to_chan
    except Exception:
        registrations = None
    _discard_descriptor(fds, fd)

        fds = None
    try:
        registrations = cycle._fd_to_chan
    except Exception:
        registrations = None
    _discard_descriptor(fds, fd)
    if registrations is not fds:
        _discard_descriptor(registrations, fd)


class MultiChannelPoller:
    """Async I/O poller for Redis transport."""

    eventflags = READ | ERR

    #: Set by :meth:`get` while reading from the socket.
    _in_protected_read = False

    #: Set of one-shot callbacks to call after reading from socket.
    after_read = None

    def __init__(self):
        # active channels
    
…[255 chars omitted]…
ne-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.clear()
        self._fd_to_chan.clear()
        self._chan_to_sock.clear()

    def add(self, channel):

                    try:
                        loop.remove(fd)
                    except Exception:
                        pass
                    _discard_cycle_descriptor(cycle, fd)
                _discard_channel_sockets(cycle, disconnected_connection)
            except Exception:
                # This callback is invoked whil
…[4262 chars omitted]…
wargs,
        ).connection_pool

    def _get_pool(self, asynchronous=False):
        params = self._connparams(asynchronous=asynchronous)
        self.keyprefix_fanout = self.keyprefix_fanout.format(db=params['db'])
        return self._sentinel_managed_pool(asynchronous)


class SentinelTransport(Transport):
    """Redis Sentinel Transport."""

    default_port = 26379
    Channel = SentinelChannel

        return
    try:
        registrations = cycle._chan_to_sock
        entries = list(registrations.items())
    except Exception:
        return
    for key, registered_sock in entries:
        try:
            _, client, _ = key
        except (TypeError, ValueError):
            continue
        registered_fd = _descriptor_for_socket(registered_sock)
        if registered_fd is None:
            continue
        if fd is not None:
            matches = registered_fd == fd
        else:

…[1066 chars omitted]…
tion:
                    return
                for fd, channel_registration in entries:
                    if not isinstance(fd, numbers.Integral) or fd < 0:
                        continue
                    try:
                        uses_connection = _registration_uses_connection(
                            channel_registration, disconnected_connection)
                    except Exception:

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x7172952b5da0>} = <Mock name='loop' id='124736943471744'>.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, 134 passed in 4.89s

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x72f5becfde40>} = <Mock name='loop' id='126399797979056'>.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, 134 passed in 4.88s

        asynchronous=True,
        exchange_type=frozenset(['direct', 'topic', 'fanout'])
    )

    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 se
…[6782 chars omitted]…
ection`).

    You must provide at least one option in Transport options:
     * `master_name` - name of the redis group to poll

    Example:
    -------
    .. code-block:: python

        >>> import kombu
        >>> c = kombu.Connection(
             'sentinel://sentinel1:26379;sentinel://sentinel2:26379',
             transport_options={'master_name': 'mymaster'}
        )
        >>> c.connect()

    try:
        descriptor = getattr(type(channel), name, None)
    except Exception:
        descriptor = None
    if descriptor is None or isinstance(descriptor, cached_property):
        return None
    try:
        # Slots-only channel objects use descriptors rather than __dict__.
        return getattr(channel, name)
    except Exception:
        return None


def _client_uses_connection(client, connection):
    if client is connection:
        return True
    try:
        return ge
…[8629 chars omitted]…
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):
    """Channel with explicit Redis Sentinel knowledge.

                    cancel()
                except Exception:
                    pass
        self._event_loop_timer_handles = []

        # Re-registering after a disconnect should not re-add file descriptors
        # that the previous connection cleanup already removed from the loop.
        if getattr(self, '_event_loop', None) is not loop
…[1320 chars omitted]…
ce(fd, numbers.Integral) or fd < 0:
                        continue
                    try:
                        uses_connection = _registration_uses_connection(
                            channel_registration, disconnected_connection)
                    except Exception:
                        uses_connection = False
                    if not uses_connection:
                        continue

                # Even if the socket has already been closed, ask the event
                # loop to remove the object it was given.  Its remove
                # operation is idempotent for descriptors it no longer
                # tracks.
                if sock is not None:
                    fd = _descriptor_for_socket(sock)
                    try:
                        loop.remove(sock)
                    except Exception:
   
…[469 chars omitted]…
     # Redis may clear _sock before notifying the channel.  In that
                # case use the channel/client association to find only this
                # connection's remaining descriptor registrations.
                try:
                    registrations = cycle._fd_to_chan
                    entries = list(registrations.items())
                except Exception:
                    return

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x77ac41561e40>} = <Mock name='loop' id='131581714941712'>.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, 134 passed in 5.04s

        self.connection._sock = None
        transport.cycle = Mock(name='cycle')
        transport.cycle.fds = fds
        conn = Mock(name='conn')
        conn.client = Mock(name='client', tr
…[519 chars omitted]…
port.register_with_event_loop.<locals>.on_poll_start at 0x738bd32fde40>} = <Mock name='loop' id='127044381408432'>.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, 134 passed in 4.89s

        descriptor = None
    if descriptor is None or isinstance(descriptor, cached_property):
        return None
    try:
        # Slots-only channel objects use descriptors rather than __dict__.
        return getattr(channel, name)
    except Exception:
        return None


def _client_uses_connection(client, connection):
    if client is connection:
        return True
    try:
        return getattr(client, 'connection', None) is connection
    except Exception:
        return False


def _channel_uses_connection(channel, c
…[6993 chars omitted]…
nnected_connection)
                        except Exception:
                            uses_connection = False
                    if not uses_connection:
                        continue
                    try:
                        loop.remove(fd)
                    except Exception:
                        pass
                    _discard_cycle_descriptor(cycle, fd)
                    try:

        def on_poll_start():
            cycle_poll_start()
            current_fds = set(cycle.fds)
            registered_fds = self._event_loop_registered_fds
            for fd in registered_fds - current_fds:
                try:
                    loop.remove(fd)
                except Exception:
                    pass
            registered_fds.
…[520 chars omitted]…
.transport_options.get(
            'health_check_interval',
            DEFAULT_HEALTH_CHECK_INTERVAL
1582:        for timer in getattr(self, '_event_loop_timer_handles', ()):
1589:        self._event_loop_timer_handles = []
1603:                previous_loop.on_tick.remove(previous_on_poll_start)
1715:        loop.on_tick.add(on_poll_start)
1719:        timer_handles = self._event_loop_timer_handles