Revert "Store s3echo.upload messages in DLQ"

This reverts commit 3c503cd.

Author: isikhar

src/dlstbx/services/s3echo_uploader.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
 
 from pathlib import Path
+from pprint import pformat
 
 import workflows.recipe
 from minio.error import S3Error
 from workflows.services.common_service import CommonService
 
 from dlstbx.util.iris import (
     get_minio_client,
+    get_presigned_urls,
     remove_objects_from_s3,
     retrieve_results_from_s3,
 )
@@ -33,13 +35,13 @@ def initializing(self):
         """
         self.log.info(f"{S3EchoUploader._service_name} starting")
 
-        workflows.recipe.wrap_subscribe(
-            self._transport,
-            "s3echo.upload",
-            self.on_upload,
-            acknowledgement=True,
-            log_extender=self.extend_log,
-        )
+        # workflows.recipe.wrap_subscribe(
+        #    self._transport,
+        #    "s3echo.upload",
+        #    self.on_upload,
+        #    acknowledgement=True,
+        #    log_extender=self.extend_log,
+        # )
 
         workflows.recipe.wrap_subscribe(
             self._transport,
@@ -55,26 +57,26 @@ def on_upload(self, rw, header, message):
         """
         # Conditionally acknowledge receipt of the message
         txn = rw.transport.transaction_begin(subscription_id=header["subscription"])
-        rw.transport.nack(header, transaction=txn)
-
-        # params = rw.recipe_step["parameters"]
-        # minio_client = get_minio_client(S3EchoUploader._s3echo_credentials)
-        # try:
-        #    s3_urls = get_presigned_urls(
-        #        minio_client,
-        #        params["bucket"],
-        #        params["rpid"],
-        #        rw.environment["s3echo_upload"].values(),
-        #        self.log,
-        #    )
-        # except S3Error:
-        #    self.log.exception(
-        #        f"Error uploading following files to S3 bucket {params['bucket']}:\n{pformat(rw.environment['s3_upload'])}"
-        #    )
-        #    rw.send_to("failure", message, transaction=txn)
-        # else:
-        #    rw.environment["s3_urls"] = s3_urls
-        #    rw.send_to("success", message, transaction=txn)
+        rw.transport.ack(header, transaction=txn)
+
+        params = rw.recipe_step["parameters"]
+        minio_client = get_minio_client(S3EchoUploader._s3echo_credentials)
+        try:
+            s3_urls = get_presigned_urls(
+                minio_client,
+                params["bucket"],
+                params["rpid"],
+                rw.environment["s3echo_upload"].values(),
+                self.log,
+            )
+        except S3Error:
+            self.log.exception(
+                f"Error uploading following files to S3 bucket {params['bucket']}:\n{pformat(rw.environment['s3_upload'])}"
+            )
+            rw.send_to("failure", message, transaction=txn)
+        else:
+            rw.environment["s3_urls"] = s3_urls
+            rw.send_to("success", message, transaction=txn)
 
         # Commit transaction
         rw.transport.transaction_commit(txn)