Merge pull request #15 from andrewwhitehead/transport-updates

Transport updates

.circleci/config.yml

@@ -28,7 +28,7 @@ jobs:
           name: Run Agent Tests
           command: |
             mkdir test-reports
-            mkdir test-reports/pytest
+            mkdir .coverage
             . venv/bin/activate
             pytest
 
@@ -40,11 +40,17 @@ jobs:
       - store_test_results:
           path: test-reports
 
+      - store_test_results:
+          path: .coverage
+
       - store_artifacts:
           path: test-reports
 
+      - store_artifacts:
+          path: .coverage
+
 workflows:
  version: 2
  aries_cloudagent:
    jobs:
-     - agent-build
+     - agent-build

.coveragerc

@@ -1,7 +1,16 @@
 [run]
-omit = */tests/* demo/* docker/* docs/* scripts/*
+omit =
+    */tests/*
+    demo/*
+    docker/*
+    docs/*
+    scripts/*
+data_file = .coverage/data
 
 [report]
 exclude_lines =
     pragma: no cover
     @abstract
+
+[xml]
+output = .coverage/coverage.xml

README.md

@@ -28,20 +28,20 @@ Instructions forthcoming. `aries_cloudagent` will be made available in the futur
 
 # Running
 
-After installing the package, `icatagent` should be available in your PATH.
+After installing the package, `acagent` should be available in your PATH.
 
 Find out more about the available command line parameters by running:
 
 ```bash
-icatagent --help
+acagent --help
 ```
 
 Currently you must specify at least one _inbound_ and one _outbound_ transport.
 
 For example:
 
 ```bash
-icatagent   --inbound-transport http 0.0.0.0 8000 \
+acagent     --inbound-transport http 0.0.0.0 8000 \
             --inbound-transport http 0.0.0.0 8001 \
             --inbound-transport ws 0.0.0.0 8002 \
             --outbound-transport ws \

aries_cloudagent/__init__.py

@@ -4,6 +4,8 @@
 
 import argparse
 import asyncio
+import functools
+import signal
 
 from aiohttp import ClientSession
 
@@ -257,15 +259,16 @@
     help="Provide external protocol modules",
 )
 
+PARSER.add_argument(
+    "--webhook-url",
+    action="append",
+    metavar="<url>",
+    help="Send webhooks to a given URL"
+)
 
-async def start(
-    inbound_transport_configs: list, outbound_transports: list, settings: dict
-):
-    """Start."""
-    registry = default_protocol_registry()
-    conductor = Conductor(
-        inbound_transport_configs, outbound_transports, registry, settings
-    )
+
+async def start(conductor: Conductor):
+    """Start up."""
     await conductor.start()
 
 
@@ -281,6 +284,18 @@ async def get_genesis_transactions(
         return genesis_txns
 
 
+async def shutdown(conductor: Conductor):
+    """Shut down."""
+    print("\nShutting down")
+    await conductor.stop()
+    tasks = [task for task in asyncio.Task.all_tasks() if task is not
+             asyncio.tasks.Task.current_task()]
+    for task in tasks:
+        task.cancel()
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    asyncio.get_event_loop().stop()
+
+
 def main():
     """Entrypoint."""
     args = PARSER.parse_args()
@@ -290,9 +305,7 @@ def main():
 
     inbound_transports = args.inbound_transports
     for transport in inbound_transports:
-        module = transport[0]
-        host = transport[1]
-        port = transport[2]
+        module, host, port = transport
         inbound_transport_configs.append(
             InboundTransportConfiguration(module=module, host=host, port=port)
         )
@@ -347,6 +360,11 @@ def main():
             settings["admin.help_link"] = args.help_link
         if args.no_receive_invites:
             settings["admin.no_receive_invites"] = True
+        hook_urls = list(args.webhook_url) if args.webhook_url else []
+        hook_url = os.environ.get("WEBHOOK_URL")
+        if hook_url:
+            hook_urls.append(hook_url)
+        settings["admin.webhook_urls"] = hook_urls
 
     if args.debug:
         settings["debug.enabled"] = True
@@ -381,17 +399,21 @@ def main():
     if args.external_protocols:
         settings["external_protocols"] = args.external_protocols
 
+    registry = default_protocol_registry()
+    conductor = Conductor(
+        inbound_transport_configs, outbound_transports, registry, settings
+    )
     loop = asyncio.get_event_loop()
+    loop.add_signal_handler(
+        signal.SIGTERM,
+        functools.partial(asyncio.ensure_future, shutdown(conductor), loop=loop),
+    )
+    asyncio.ensure_future(start(conductor), loop=loop)
+
     try:
-        # asyncio.ensure_future(
-        #     start(inbound_transport_configs, outbound_transports, settings), loop=loop
-        # )
-        loop.run_until_complete(
-            start(inbound_transport_configs, outbound_transports, settings)
-        )
         loop.run_forever()
     except KeyboardInterrupt:
-        print("\nShutting down")
+        loop.run_until_complete(shutdown(conductor))
 
 
 if __name__ == "__main__":

aries_cloudagent/admin/base_server.py

@@ -2,6 +2,7 @@
 
 
 from abc import ABC, abstractmethod
+from typing import Sequence
 
 
 class BaseAdminServer(ABC):
@@ -22,5 +23,15 @@ async def stop(self) -> None:
         """Stop the webserver."""
 
     @abstractmethod
-    async def add_event(self, message: dict):
-        """Add an event to existing queues."""
+    def add_webhook_target(
+        self, target_url: str, topic_filter: Sequence[str] = None, retries: int = None
+    ):
+        """Add a webhook target."""
+
+    @abstractmethod
+    def remove_webhook_target(self, target_url: str):
+        """Remove a webhook target."""
+
+    @abstractmethod
+    async def send_webhook(self, topic: str, payload: dict):
+        """Add a webhook to the queue, to send to all registered targets."""