Repository: mercadona/rele
Pull request #225 (2022-07-21 10:52:25)

[Changed] Modify Dockerfile to work when using make docker commands
### :tophat: What?
Modify Dockerfile to work fine when using make docker commands

### :thinking: Why?
When we use docker make commands, we are receiving this error:

<img width="1341" alt="image (3)" src="https://user-images.githubusercontent.com/32519590/180170839-cc229d0a-9702-45b3-a9de-6131bc237252.png">

We have fixed this file to work fine :)

I also updated the list of code owners.
**The deleted users are due to not having write access to the repository.


Patch:
diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS
index dd7b077..a648ec3 100644
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -1,1 +1,1 @@
-* @andrewgy8 @tobami @edgarlatorre @daniel-ruiz
+* @daniel-ruiz @EmilioCarrion @mgarcia0094 @AdamjGoddard
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 305b459..49926d4 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,6 +1,10 @@
 Changelog
 =========
 
+1.6.0 (2022-08-03)
+-------------------
+* [Added] Implement auto restart of the consumption when futures are done or cancelled. (#226)
+
 1.5.0 (2022-04-20)
 -------------------
 * [Added] Add filter expressions to subscriptions. (#207)
diff --git a/Dockerfile b/Dockerfile
index 9f25446..70b5386 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -4,11 +4,8 @@ SHELL ["/bin/bash", "-o", "pipefail", "-c"]
 WORKDIR /rele
 LABEL python_version=python
 
-COPY Makefile ./
-COPY requirements requirements/
+COPY . .
 
 RUN make install-dev-requirements
 
-COPY . .
-
 CMD ["make", "clean", "lint", "test"]
diff --git a/rele/__init__.py b/rele/__init__.py
index ae0f6a8..e9bbdbe 100644
--- a/rele/__init__.py
+++ b/rele/__init__.py
@@ -1,4 +1,4 @@
-__version__ = "1.5.0"
+__version__ = "1.6.0"
 
 try:
     import django
diff --git a/rele/worker.py b/rele/worker.py
index 5536b55..639dae5 100644
--- a/rele/worker.py
+++ b/rele/worker.py
@@ -31,7 +31,7 @@ class Worker:
         threads_per_subscription=None,
     ):
         self._subscriber = Subscriber(gc_project_id, credentials, default_ack_deadline)
-        self._futures = []
+        self._futures = {}
         self._subscriptions = subscriptions
         self.threads_per_subscription = threads_per_subscription
 
@@ -56,20 +56,7 @@ class Worker:
         """
         run_middleware_hook("pre_worker_start")
         for subscription in self._subscriptions:
-            executor_kwargs = {
-                "thread_name_prefix": "ThreadPoolExecutor-ThreadScheduler"
-            }
-            executor = futures.ThreadPoolExecutor(
-                max_workers=self.threads_per_subscription, **executor_kwargs
-            )
-            scheduler = ThreadScheduler(executor=executor)
-            self._futures.append(
-                self._subscriber.consume(
-                    subscription_name=subscription.name,
-                    callback=Callback(subscription),
-                    scheduler=scheduler,
-                )
-            )
+            self._boostrap_consumption(subscription)
         run_middleware_hook("post_worker_start")
 
     def run_forever(self, sleep_interval=1):
@@ -99,15 +86,36 @@ class Worker:
         :param frame: Needed for `signal.signal <https://docs.python.org/3/library/signal.html#signal.signal>`_  # noqa
         """
         run_middleware_hook("pre_worker_stop", self._subscriptions)
-        for future in self._futures:
+        for future in self._futures.values():
             future.cancel()
 
         run_middleware_hook("post_worker_stop")
         sys.exit(0)
 
+    def _boostrap_consumption(self, subscription):
+        if subscription in self._futures:
+            self._futures[subscription].cancel()
+
+        executor_kwargs = {"thread_name_prefix": "ThreadPoolExecutor-ThreadScheduler"}
+        executor = futures.ThreadPoolExecutor(
+            max_workers=self.threads_per_subscription, **executor_kwargs
+        )
+        scheduler = ThreadScheduler(executor=executor)
+
+        self._futures[subscription] = self._subscriber.consume(
+            subscription_name=subscription.name,
+            callback=Callback(subscription),
+            scheduler=scheduler,
+        )
+
     def _wait_forever(self, sleep_interval):
         logger.info("Consuming subscriptions...")
         while True:
+            for subscription, future in self._futures.items():
+                if future.cancelled() or future.done():
+                    logger.info(f"Restarting consumption of {subscription.name}.")
+                    self._boostrap_consumption(subscription)
+
             time.sleep(sleep_interval)
 
 


Test patch:
diff --git a/tests/test_worker.py b/tests/test_worker.py
index 3a63b6f..94d610d 100644
--- a/tests/test_worker.py
+++ b/tests/test_worker.py
@@ -1,11 +1,14 @@
+import time
 from concurrent import futures
 from unittest.mock import ANY, patch
 
 import pytest
+from google.cloud import pubsub_v1
 from google.cloud.pubsub_v1.subscriber.scheduler import ThreadScheduler
 
 from rele import Subscriber, Worker, sub
 from rele.middleware import register_middleware
+from rele.subscription import Callback
 from rele.worker import create_and_run
 
 
@@ -27,8 +30,10 @@ def worker(config):
 
 
 @pytest.fixture
-def mock_consume():
+def mock_consume(config):
     with patch.object(Subscriber, "consume") as m:
+        client = pubsub_v1.SubscriberClient(credentials=config.credentials)
+        m.return_value = client.subscribe("dummy-subscription", Callback(sub_stub))
         yield m
 
 
@@ -108,6 +113,38 @@ class TestWorker:
         assert worker._subscriber._gc_project_id == "rele-test"
 
 
+@pytest.mark.usefixtures("mock_create_subscription")
+class TestRestartConsumer:
+    @pytest.fixture(autouse=True)
+    def mock_sleep(self):
+        with patch.object(time, "sleep", side_effect=ValueError) as m:
+            yield m
+
+    def test_does_not_restart_consumption_when_everything_goes_well(
+        self, worker, mock_consume
+    ):
+        with pytest.raises(ValueError):
+            worker.run_forever()
+
+        assert len(mock_consume.call_args_list) == 1
+
+    def test_restarts_consumption_when_future_is_cancelled(self, worker, mock_consume):
+        mock_consume.return_value.cancel()
+
+        with pytest.raises(ValueError):
+            worker.run_forever()
+
+        assert len(mock_consume.call_args_list) == 2
+
+    def test_restarts_consumption_when_future_is_done(self, worker, mock_consume):
+        mock_consume.return_value.set_result(True)
+
+        with pytest.raises(ValueError):
+            worker.run_forever()
+
+        assert len(mock_consume.call_args_list) == 2
+
+
 class TestCreateAndRun:
     @pytest.fixture(autouse=True)
     def worker_wait_forever(self):