add cloud_pubsub_dead_letter_ack, check topic before send

roticagas · Jan 11, 2019 · b9e9cdb · b9e9cdb
1 parent 92a35af
commit b9e9cdb
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 9 deletions.
diff --git a/PubSubRunner/runner_application.py b/PubSubRunner/runner_application.py
@@ -26,7 +26,7 @@ def verify_config(self):
             logging.warning('publish topic are not set, check your config carefully')
         if self.config.cloud_pubsub_dead_letter_topic == '':
             logging.warning('dead letter topic are not set, check your config carefully')
-            assert not self.config.cloud_pubsub_ack, 'dead letter should not be sent if message not ack'
+            assert not self.config.cloud_pubsub_dead_letter_ack, 'dead letter should not be sent if message not ack'
 
     def check_pubsub(self):
         """
@@ -72,14 +72,15 @@ def subscribe_processing(self, message):
         except JSONDecodeError as e:
             # NOTE: if payload not in json format: send to dead letter topic
             logging.error(str(e))
-            CloudUtil.publish_dead_letter(self.config.cloud_project,
-                                          self.config.cloud_pubsub_dead_letter_topic,
-                                          self.config.cloud_pubsub_publish_topic,
-                                          message,
-                                          e,
-                                          'Invalid json format')
-            if self.config.cloud_pubsub_ack:
-                message.ack()
+            if self.config.cloud_pubsub_dead_letter_topic != '':
+                CloudUtil.publish_dead_letter(self.config.cloud_project,
+                                              self.config.cloud_pubsub_dead_letter_topic,
+                                              self.config.cloud_pubsub_publish_topic,
+                                              message,
+                                              e,
+                                              'Invalid json format')
+                if self.config.cloud_pubsub_dead_letter_ack:
+                    message.ack()
         except Exception as e:
             logging.error(str(e))
             message.nack()

diff --git a/PubSubRunner/runner_config.py b/PubSubRunner/runner_config.py
@@ -13,6 +13,7 @@ class RunnerConfig:
         ('CLOUD_PUBSUB_MAX_DEADLINE', '600'),
         ('CLOUD_PUBSUB_CHECK', 'true'),
         ('CLOUD_PUBSUB_ACK', 'true'),
+        ('CLOUD_PUBSUB_DEAD_LETTER_ACK', 'false'),
     ]
 
     def __init__(self):
@@ -27,3 +28,4 @@ def __init__(self):
         self.cloud_pubsub_ack_deadline = int(env[7])
         self.cloud_pubsub_check = env[8] in ['True', 'true', 'TRUE']
         self.cloud_pubsub_ack = env[9] in ['True', 'true', 'TRUE']
+        self.cloud_pubsub_dead_letter_ack = env[10] in ['True', 'true', 'TRUE']