Replaced time.time usage with time.monotonic

Author: eandersson

CHANGELOG.rst

@@ -25,6 +25,9 @@ Version 3.1
   short type-hint rendering in the Sphinx config.
 - SSL: hostname verification now on by default; unknown ``cert_reqs``
   URI values fall back to ``CERT_REQUIRED`` instead of ``CERT_NONE``.
+- Switched all elapsed-duration checks (RPC and connection-state
+  timeouts) from ``time.time()`` to ``time.monotonic()`` so system
+  clock changes can no longer cause spurious timeouts or hangs.
 
 Version 3.0
 -----------

amqpstorm/connection.py

@@ -430,9 +430,9 @@ def _wait_for_connection_state(
 
         :return:
         """
-        start_time = time.time()
+        start_time = time.monotonic()
         while self.current_state != state:
             self.check_for_errors()
-            if time.time() - start_time > rpc_timeout:
+            if time.monotonic() - start_time > rpc_timeout:
                 raise AMQPConnectionError('connection timed out')
             time.sleep(IDLE_WAIT)

amqpstorm/rpc.py

@@ -139,10 +139,10 @@ def _wait_for_request(
         :param obj connection_adapter: Provide custom connection adapter.
         :return:
         """
-        start_time = time.time()
+        start_time = time.monotonic()
         while not self._response[uuid]:
             connection_adapter.check_for_errors()
-            if time.time() - start_time > self._timeout:
+            if time.monotonic() - start_time > self._timeout:
                 self._raise_rpc_timeout_error(uuid)
             time.sleep(IDLE_WAIT)
 

amqpstorm/tests/functional/ssl/test_reliability.py

@@ -194,9 +194,9 @@ def test_functional_ssl_close_performance(self):
                 SSL_HOST, USERNAME, PASSWORD, timeout=60, port=5671, ssl=True,
                 ssl_options=ssl_options)
 
-            start_time = time.time()
+            start_time = time.monotonic()
             self.connection.close()
-            self.assertLess(time.time() - start_time, 3)
+            self.assertLess(time.monotonic() - start_time, 3)
 
     @setup(new_connection=False)
     def test_functional_ssl_uri_connection(self):
@@ -287,9 +287,9 @@ def test_functional_publish_1k_with_ssl(self):
             consumer_thread.daemon = True
             consumer_thread.start()
 
-        start_time = time.time()
+        start_time = time.monotonic()
         while self.messages_consumed != self.messages_to_send:
-            if time.time() - start_time >= 60:
+            if time.monotonic() - start_time >= 60:
                 break
             time.sleep(0.1)
 

amqpstorm/tests/functional/test_generic.py

@@ -339,9 +339,9 @@ def on_message_second(message):
         # Sleep for 0.1s to make sure RabbitMQ has time to catch up.
         time.sleep(0.1)
 
-        start_time = time.time()
+        start_time = time.monotonic()
         while len(inbound_messages) != 1:
-            if time.time() - start_time >= 30:
+            if time.monotonic() - start_time >= 30:
                 break
             time.sleep(0.1)
 
@@ -373,10 +373,10 @@ def on_message(message):
                                    queue=self.queue_name,
                                    no_ack=True)
 
-        start_time = time.time()
+        start_time = time.monotonic()
         while len(inbound_messages) == 0:
             self.channel.process_data_events()
-            if time.time() - start_time >= 30:
+            if time.monotonic() - start_time >= 30:
                 break
             time.sleep(0.1)
 

amqpstorm/tests/functional/test_reliability.py

@@ -136,9 +136,9 @@ def test_functional_close_performance(self):
         """
         for _ in range(10):
             self.connection = self._make_connection()
-            start_time = time.time()
+            start_time = time.monotonic()
             self.connection.close()
-            self.assertLess(time.time() - start_time, 3)
+            self.assertLess(time.monotonic() - start_time, 3)
 
     @setup(new_connection=False)
     def test_functional_close_after_channel_close_forced_by_server(self):
@@ -159,13 +159,13 @@ def test_functional_close_after_channel_close_forced_by_server(self):
                 )
             except (AMQPConnectionError, AMQPChannelError):
                 pass
-            start_time = time.time()
+            start_time = time.monotonic()
             channel.close()
-            self.assertLess(time.time() - start_time, 3)
+            self.assertLess(time.monotonic() - start_time, 3)
 
-            start_time = time.time()
+            start_time = time.monotonic()
             connection.close()
-            self.assertLess(time.time() - start_time, 3)
+            self.assertLess(time.monotonic() - start_time, 3)
 
     @setup(new_connection=False)
     def test_functional_uri_connection(self):
@@ -267,9 +267,9 @@ def test_functional_publish_and_consume_1k_messages(self):
             consumer_thread.daemon = True
             consumer_thread.start()
 
-        start_time = time.time()
+        start_time = time.monotonic()
         while self.messages_consumed != self.messages_to_send:
-            if time.time() - start_time >= 60:
+            if time.monotonic() - start_time >= 60:
                 break
             time.sleep(0.1)