Repository: mercadona/rele
Pull request #137 (2019-12-12 09:22:46)

Add publishing timeout
### :tophat: What?

Configurable timeout while waiting for a successful result from the pubsub future. https://googleapis.dev/python/pubsub/1.1.0/publisher/api/futures.html?highlight=result#google.cloud.pubsub_v1.publisher.futures.Future.result

### :thinking: Why?

We would like to control the timeout for publishing.

### :link: Related issue

Addresses #136 



Patch:
diff --git a/docs/api/settings.rst b/docs/api/settings.rst
index 7c95496..176177d 100644
--- a/docs/api/settings.rst
+++ b/docs/api/settings.rst
@@ -31,6 +31,7 @@ Example::
         'APP_NAME': 'myappname',
         'ENCODER_PATH': 'rest_framework.utils.encoders.JSONEncoder',
         'ACK_DEADLINE': 120,
+        'PUBLISHER_TIMEOUT': 3.0
     }
 
 
@@ -116,10 +117,20 @@ serializing your Python data structure to a json object when publishing.
 
 Ack deadline for all subscribers in seconds.
 
-
-
 .. seealso:: The `Google Pub/Sub documentation <https://cloud.google.com/pubsub/docs/subscriber>`_
     which states that *The subscriber has a configurable, limited amount of time --
     known as the ackDeadline -- to acknowledge the outstanding message. Once the deadline
     passes, the message is no longer considered outstanding, and Cloud Pub/Sub will attempt
     to redeliver the message.*
+
+``PUBLISHER_TIMEOUT``
+---------------------
+
+**Optional**
+
+Default: 3.0 seconds
+
+Timeout that the publishing result will wait on the future to publish successfully while blocking.
+
+`See Google PubSub documentation for more info
+<https://googleapis.dev/python/pubsub/1.1.0/publisher/api/futures.html?highlight=result#google.cloud.pubsub_v1.publisher.futures.Future.result>`_
diff --git a/rele/client.py b/rele/client.py
index af36101..749cf1f 100644
--- a/rele/client.py
+++ b/rele/client.py
@@ -85,10 +85,10 @@ class Publisher:
     :param gc_project_id: string Google Cloud Project ID.
     :param credentials: string Google Cloud Credentials.
     :param encoder: A valid `json.encoder.JSONEncoder subclass <https://docs.python.org/3/library/json.html#json.JSONEncoder>`_  # noqa
-    :param timeout: integer, default 3.0 seconds.
+    :param timeout: float
     """
 
-    def __init__(self, gc_project_id, credentials, encoder, timeout=3.0):
+    def __init__(self, gc_project_id, credentials, encoder, timeout):
         self._gc_project_id = gc_project_id
         self._timeout = timeout
         self._encoder = encoder
@@ -114,7 +114,7 @@ class Publisher:
         Usage::
 
             publisher = Publisher()
-            future = publisher.publish('topic_name', {'foo': 'bar'}, blocking=True) # noqa
+            future = publisher.publish('topic_name', {'foo': 'bar'}, blocking=True, timeout=10.0) # noqa
 
         However, it should be noted that using `blocking=True` may incur a
         significant performance hit.
@@ -125,7 +125,7 @@ class Publisher:
 
         :param topic: string topic to publish the data.
         :param data: dict with the content of the message.
-        :param blocking: boolean, default False.
+        :param blocking: boolean
         :param attrs: Extra parameters to be published.
         :return: `Future <https://googleapis.github.io/google-cloud-python/latest/pubsub/subscriber/api/futures.html>`_  # noqa
         """
diff --git a/rele/config.py b/rele/config.py
index e9706fb..1d0c8c9 100644
--- a/rele/config.py
+++ b/rele/config.py
@@ -28,6 +28,7 @@ class Config:
             "ACK_DEADLINE", os.environ.get("DEFAULT_ACK_DEADLINE", 60)
         )
         self._encoder_path = setting.get("ENCODER_PATH", DEFAULT_ENCODER_PATH)
+        self.publisher_timeout = setting.get("PUBLISHER_TIMEOUT", 3.0)
 
     @property
     def encoder(self):
diff --git a/rele/publishing.py b/rele/publishing.py
index 0625dcc..e2bf1d4 100644
--- a/rele/publishing.py
+++ b/rele/publishing.py
@@ -10,6 +10,7 @@ def init_global_publisher(config):
             gc_project_id=config.gc_project_id,
             credentials=config.credentials,
             encoder=config.encoder,
+            timeout=config.publisher_timeout,
         )
     return _publisher
 
@@ -33,6 +34,8 @@ def publish(topic, data, **kwargs):
 
     :param topic: str PubSub topic name
     :param data: dict-like Data to be sent as the message.
+    :param timeout: float. Default None, falls back to RELE['PUBLISHER_TIMEOUT'] value
+    :param blocking: boolean. Default False
     :param kwargs: Any optional key-value pairs that are included as attributes
         in the message
     :return: None


Test patch:
diff --git a/tests/conftest.py b/tests/conftest.py
index 1e5d53c..c342704 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -41,7 +41,12 @@ def subscriber(project_id, credentials):
 
 @pytest.fixture
 def publisher(config):
-    publisher = Publisher(config.gc_project_id, config.credentials, config.encoder)
+    publisher = Publisher(
+        gc_project_id=config.gc_project_id,
+        credentials=config.credentials,
+        encoder=config.encoder,
+        timeout=config.publisher_timeout,
+    )
     publisher._client = MagicMock(spec=PublisherClient)
     publisher._client.publish.return_value = concurrent.futures.Future()
 
diff --git a/tests/test_client.py b/tests/test_client.py
index a46e58d..954ff06 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,5 +1,4 @@
 import decimal
-import os
 import concurrent
 from unittest.mock import ANY, patch
 
@@ -51,6 +50,19 @@ class TestPublisher:
 
         publisher._client.publish.assert_called_with(ANY, b"1.2", published_at=ANY)
 
+    @patch.object(concurrent.futures.Future, "result")
+    def test_publishes_data_with_client_timeout_when_blocking(
+        self, mock_future, publisher
+    ):
+        publisher._timeout = 100.0
+        publisher.publish(topic="order-cancelled", data={"foo": "bar"}, blocking=True)
+
+        publisher._client.publish.return_value = mock_future
+        publisher._client.publish.assert_called_with(
+            ANY, b'{"foo": "bar"}', published_at=ANY
+        )
+        mock_future.assert_called_once_with(timeout=100)
+
 
 class TestSubscriber:
     @patch.object(SubscriberClient, "create_subscription")
diff --git a/tests/test_publishing.py b/tests/test_publishing.py
index 3f012b1..f1befe2 100644
--- a/tests/test_publishing.py
+++ b/tests/test_publishing.py
@@ -3,7 +3,7 @@ from unittest.mock import MagicMock, patch
 from rele import Publisher, publishing
 
 
-class TestPublish:
+class TestInitGlobalPublisher:
     @patch("rele.publishing.Publisher", autospec=True)
     def test_creates_global_publisher_when_published_called(
         self, mock_publisher, config
diff --git a/tests/test_subscription.py b/tests/test_subscription.py
index 368ca96..ec33e67 100644
--- a/tests/test_subscription.py
+++ b/tests/test_subscription.py
@@ -119,7 +119,7 @@ class TestCallback:
                 "topic": "some-cool-topic",
                 "status": "received",
                 "subscription": "rele-some-cool-topic",
-                "attributes": {"lang": "es", "published_at": str(published_at),},
+                "attributes": {"lang": "es", "published_at": str(published_at)},
             },
         }
 
@@ -153,7 +153,7 @@ class TestCallback:
                 "status": "succeeded",
                 "subscription": "rele-some-cool-topic",
                 "duration_seconds": pytest.approx(0.5, abs=0.5),
-                "attributes": {"lang": "es", "published_at": str(published_at),},
+                "attributes": {"lang": "es", "published_at": str(published_at)},
             },
         }
 
@@ -182,7 +182,7 @@ class TestCallback:
                 "status": "failed",
                 "subscription": "rele-some-cool-topic",
                 "duration_seconds": pytest.approx(0.5, abs=0.5),
-                "attributes": {"lang": "es", "published_at": str(published_at),},
+                "attributes": {"lang": "es", "published_at": str(published_at)},
             },
         }
         assert failed_log.subscription_message == message_wrapper