enhance: allow uses to disable broker heartbeats by not providing a timeout (#1997,#1998)

kombu/mixins.py

@@ -195,10 +195,11 @@ def consume(self, limit=None, timeout=None, safety_interval=1, **kwargs):
                 try:
                     conn.drain_events(timeout=safety_interval)
                 except socket.timeout:
-                    conn.heartbeat_check()
-                    elapsed += safety_interval
-                    if timeout and elapsed >= timeout:
-                        raise
+                    if timeout:
+                        conn.heartbeat_check()
+                        elapsed += safety_interval
+                        if elapsed >= timeout:
+                            raise
                 except OSError:
                     if not self.should_stop:
                         raise

t/unit/test_mixins.py

@@ -89,6 +89,26 @@ def se(*args, **kwargs):
         c.connection.drain_events.side_effect = se
         with pytest.raises(socket.error):
             next(it)
+        c.connection.heartbeat_check.assert_called()
+
+    def test_consume_drain_no_heartbeat_check(self):
+        c, Acons, Bcons = self._context()
+        c.should_stop = False
+        it = c.consume(no_ack=True, timeout=None)
+
+        def se(*args, **kwargs):
+            c.should_stop = True
+            raise socket.timeout()
+        c.connection.drain_events.side_effect = se
+        with pytest.raises(StopIteration):
+            next(it)
+        c.connection.heartbeat_check.assert_not_called()
+
+        it = c.consume(no_ack=True, timeout=0)
+        c.connection.drain_events.side_effect = se
+        with pytest.raises(StopIteration):
+            next(it)
+        c.connection.heartbeat_check.assert_not_called()
 
     def test_Consumer_context(self):
         c, Acons, Bcons = self._context()