prevent event loop polling on closed redis transports (and causing leak)

kombu/transport/redis.py

@@ -1237,6 +1237,12 @@ def register_with_event_loop(self, connection, loop):
         def _on_disconnect(connection):
             if connection._sock:
                 loop.remove(connection._sock)
+
+            # stop polling in the event loop
+            try:
+                loop.on_tick.remove(on_poll_start)
+            except KeyError:
+                pass
         cycle._on_connection_disconnect = _on_disconnect
 
         def on_poll_start():

t/unit/transport/test_redis.py

@@ -854,6 +854,21 @@ def test_register_with_event_loop(self):
             call(13, transport.on_readable, 13),
         ])
 
+    def test_register_with_event_loop__on_disconnect__loop_cleanup(self):
+        """Ensure event loop polling stops on disconnect."""
+        transport = self.connection.transport
+        self.connection._sock = None
+        transport.cycle = Mock(name='cycle')
+        transport.cycle.fds = {12: 'LISTEN', 13: 'BRPOP'}
+        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)
+        assert len(loop.on_tick) == 1
+        transport.cycle._on_connection_disconnect(self.connection)
+        assert loop.on_tick == set()
+
     def test_configurable_health_check(self):
         transport = self.connection.transport
         transport.cycle = Mock(name='cycle')