Merge pull request openwallet-foundation#1659 from shaangill025/cleanup_redis

Redis PQ Cleanup

Author: swcurran

OutboundQueue.md

@@ -1169,22 +1169,6 @@ def add_arguments(self, parser: ArgumentParser):
                 "Supported outbound transport types are 'http' and 'ws'."
             ),
         )
-        parser.add_argument(
-            "-oq",
-            "--outbound-queue",
-            dest="outbound_queue",
-            type=str,
-            env_var="ACAPY_OUTBOUND_TRANSPORT_QUEUE",
-            help=(
-                "Defines the location of the Outbound Queue Engine. This must be "
-                "a 'dotpath' to a Python module on the PYTHONPATH, followed by a "
-                "colon, followed by the name of a Python class that implements "
-                "BaseOutboundQueue. This commandline option is the official entry "
-                "point of ACA-py's pluggable queue interface. The default value is: "
-                "'aries_cloudagent.transport.outbound.queue.redis:RedisOutboundQueue'."
-                ""
-            ),
-        )
         parser.add_argument(
             "-l",
             "--label",
@@ -1264,20 +1248,10 @@ def get_settings(self, args: Namespace):
             settings["transport.inbound_configs"] = args.inbound_transports
         else:
             raise ArgsParseError("-it/--inbound-transport is required")
-        if not args.outbound_transports and not args.outbound_queue:
-            raise ArgsParseError(
-                "-ot/--outbound-transport or -oq/--outbound-queue is required"
-            )
-        if args.outbound_transports and args.outbound_queue:
-            raise ArgsParseError(
-                "-ot/--outbound-transport and -oq/--outbound-queue are "
-                "not allowed together"
-            )
         if args.outbound_transports:
             settings["transport.outbound_configs"] = args.outbound_transports
-        if args.outbound_queue:
-            settings["transport.outbound_queue"] = args.outbound_queue
-
+        else:
+            raise ArgsParseError("-ot/--outbound-transport is required")
         settings["transport.enable_undelivered_queue"] = args.enable_undelivered_queue
 
         if args.label:

aries_cloudagent/config/argparse.py

@@ -83,7 +83,6 @@ def print_banner(
         agent_label,
         inbound_transports,
         outbound_transports,
-        outbound_queue,
         public_did,
         admin_server=None,
         banner_length=40,
@@ -96,7 +95,6 @@ def print_banner(
             agent_label: Agent Label
             inbound_transports: Configured inbound transports
             outbound_transports: Configured outbound transports
-            outbound_queue: The outbound queue engine instance
             admin_server: Admin server info
             public_did: Public DID
             banner_length: (Default value = 40) Length of the banner
@@ -124,6 +122,18 @@ def print_banner(
         )
         banner.print_spacer()
 
+        external_in_transports = set().union(
+            *(
+                transport
+                for transport in inbound_transports.values()
+                if transport.is_external
+            )
+        )
+        if external_in_transports:
+            banner.print_spacer()
+            banner.print_list([f"{external_in_transports}"])
+            banner.print_spacer()
+
         # Outbound transports
         schemes = set().union(
             *(transport.schemes for transport in outbound_transports.values())
@@ -134,15 +144,16 @@ def print_banner(
             banner.print_list([f"{scheme}" for scheme in sorted(schemes)])
             banner.print_spacer()
 
-        # Outbound queue
-        if outbound_queue:
-            banner.print_subtitle("Outbound Queue")
-            banner.print_spacer()
-            banner.print_list(
-                [
-                    f"{outbound_queue}",
-                ]
+        external_out_transports = set().union(
+            *(
+                transport
+                for transport in outbound_transports.values()
+                if transport.is_external
             )
+        )
+        if external_out_transports:
+            banner.print_spacer()
+            banner.print_list([f"{external_out_transports}"])
             banner.print_spacer()
 
         # DID info

aries_cloudagent/config/logging.py

@@ -144,27 +144,6 @@ async def test_outbound_is_required(self):
         with self.assertRaises(argparse.ArgsParseError):
             settings = group.get_settings(result)
 
-    async def test_outbound_queue(self):
-        """Test outbound queue class path string."""
-        parser = argparse.create_argument_parser()
-        group = argparse.TransportGroup()
-        group.add_arguments(parser)
-
-        result = parser.parse_args(
-            [
-                "--inbound-transport",
-                "http",
-                "0.0.0.0",
-                "80",
-                "--outbound-queue",
-                "my_queue.mod.path",
-            ]
-        )
-
-        settings = group.get_settings(result)
-
-        assert settings.get("transport.outbound_queue") == "my_queue.mod.path"
-
     async def test_general_settings_file(self):
         """Test file argument parsing."""
 

aries_cloudagent/config/tests/test_argparse.py

@@ -66,38 +66,15 @@ def test_banner_did(self):
                 test_label,
                 {"in": mock_http},
                 {"out": mock_https},
-                None,
                 test_did,
                 mock_admin_server,
             )
             test_module.LoggingConfigurator.print_banner(
-                test_label, {"in": mock_http}, {"out": mock_https}, None, test_did
+                test_label, {"in": mock_http}, {"out": mock_https}, test_did
             )
         output = stdout.getvalue()
         assert test_did in output
 
-    def test_banner_outbound_queue(self):
-        stdout = StringIO()
-        mock_http = async_mock.MagicMock(scheme="http", host="1.2.3.4", port=8081)
-        mock_queue = "mocked queue text"
-        mock_admin_server = async_mock.MagicMock(host="1.2.3.4", port=8091)
-        with contextlib.redirect_stdout(stdout):
-            test_label = "Aries Cloud Agent"
-            test_did = "55GkHamhTU1ZbTbV2ab9DE"
-            test_module.LoggingConfigurator.print_banner(
-                test_label,
-                {"in": mock_http},
-                {},
-                mock_queue,
-                test_did,
-                mock_admin_server,
-            )
-            test_module.LoggingConfigurator.print_banner(
-                test_label, {"in": mock_http}, {}, mock_queue, test_did
-            )
-        output = stdout.getvalue()
-        assert "mocked queue text" in output
-
     def test_load_resource(self):
         with async_mock.patch("builtins.open", async_mock.MagicMock()) as mock_open:
             test_module.load_resource("abc", encoding="utf-8")

aries_cloudagent/config/tests/test_logging.py

@@ -56,8 +56,6 @@
 from ..transport.outbound.base import OutboundDeliveryError
 from ..transport.outbound.manager import OutboundTransportManager, QueuedOutboundMessage
 from ..transport.outbound.message import OutboundMessage
-from ..transport.outbound.queue.base import BaseOutboundQueue
-from ..transport.outbound.queue.loader import get_outbound_queue
 from ..transport.outbound.status import OutboundSendStatus
 from ..transport.wire_format import BaseWireFormat
 from ..utils.stats import Collector
@@ -97,7 +95,6 @@ def __init__(self, context_builder: ContextBuilder) -> None:
         self.outbound_transport_manager: OutboundTransportManager = None
         self.root_profile: Profile = None
         self.setup_public_did: DIDInfo = None
-        self.outbound_queue: BaseOutboundQueue = None
 
     @property
     def context(self) -> InjectionContext:
@@ -211,8 +208,6 @@ async def setup(self):
             DocumentLoader, DocumentLoader(self.root_profile)
         )
 
-        self.outbound_queue = get_outbound_queue(self.root_profile)
-
         # Admin API
         if context.settings.get("admin.enabled"):
             try:
@@ -273,13 +268,6 @@ async def start(self) -> None:
             LOGGER.exception("Unable to start outbound transports")
             raise
 
-        if self.outbound_queue:
-            try:
-                await self.outbound_queue.start()
-            except Exception:
-                LOGGER.exception("Unable to start outbound queue")
-                raise
-
         # Start up Admin server
         if self.admin_server:
             try:
@@ -303,7 +291,6 @@ async def start(self) -> None:
             default_label,
             self.inbound_transport_manager.registered_transports,
             self.outbound_transport_manager.registered_transports,
-            self.outbound_queue,
             self.setup_public_did and self.setup_public_did.did,
             self.admin_server,
         )
@@ -489,8 +476,6 @@ async def stop(self, timeout=1.0):
             shutdown.run(self.inbound_transport_manager.stop())
         if self.outbound_transport_manager:
             shutdown.run(self.outbound_transport_manager.stop())
-        if self.outbound_queue:
-            shutdown.run(self.outbound_queue.stop())
 
         # close multitenant profiles
         multitenant_mgr = self.context.inject_or(BaseMultitenantManager)
@@ -669,44 +654,14 @@ async def queue_outbound(
                     self.admin_server.notify_fatal_error()
                 raise
             del conn_mgr
-        # If ``self.outbound_queue`` is specified (usually set via
-        # outbound queue `-oq` commandline option), use that external
-        # queue. Else save the message to an internal queue. This
-        # internal queue usually results in the message to be sent over
-        # ACA-py `-ot` transport.
-        if self.outbound_queue:
-            return await self._queue_external(profile, outbound)
-        else:
-            return self._queue_internal(profile, outbound)
-
-    async def _queue_external(
-        self,
-        profile: Profile,
-        outbound: OutboundMessage,
-    ) -> OutboundSendStatus:
-        """Save the message to an external outbound queue."""
-        async with self.outbound_queue:
-            targets = (
-                [outbound.target] if outbound.target else (outbound.target_list or [])
-            )
-            for target in targets:
-                encoded_outbound_message = (
-                    await self.outbound_transport_manager.encode_outbound_message(
-                        profile, outbound, target
-                    )
-                )
-                await self.outbound_queue.enqueue_message(
-                    encoded_outbound_message.payload, target.endpoint
-                )
-
-            return OutboundSendStatus.SENT_TO_EXTERNAL_QUEUE
+        return await self._queue_message(profile, outbound)
 
-    def _queue_internal(
+    async def _queue_message(
         self, profile: Profile, outbound: OutboundMessage
     ) -> OutboundSendStatus:
         """Save the message to an internal outbound queue."""
         try:
-            self.outbound_transport_manager.enqueue_message(profile, outbound)
+            await self.outbound_transport_manager.enqueue_message(profile, outbound)
             return OutboundSendStatus.QUEUED_FOR_DELIVERY
         except OutboundDeliveryError:
             LOGGER.warning("Cannot queue message for delivery, no supported transport")