Merge pull request #62 from OpenMined/feature/add_listener

Feature/add listener

Author: wip-abramson

.gitignore

@@ -14,7 +14,4 @@ libs/om-aries-controller/demo/bob/image_received.png
 # exception to the rule
 !libs/om-aries-controller/demo/bob/received_files/.gitkeep
 
-**/model.pt
-**/trained_model.pt
-**/untrained_model.pt
-**/part_trained.pt
+**/*.pt

libs/aries-basic-controller/aries_basic_controller/aries_controller.py

@@ -93,7 +93,21 @@ def register_listeners(self, listeners, defaults=True):
                 pub.subscribe(self.proofs.default_handler, "present_proof")
 
         for listener in listeners:
-            pub.subscribe(listener["handler"], listener["topic"])
+            self.add_listener(listener)
+
+    def add_listener(self, listener):
+        pub.subscribe(listener["handler"], listener["topic"])
+
+    def remove_listener(self, listener):
+        if pub.isSubscribed(listener["handler"], listener["topic"]):
+            pub.unsubscribe(listener["handler"], listener["topic"])
+        else:
+            logger.debug("Listener not subscribed", listener)
+
+    def remove_all_listeners(self, topic: str = None):
+        # Note advanced use of function can include both listenerFilter and topicFilter for this
+        # Add when needed
+        pub.unsubAll(topicName=topic)
 
     async def listen_webhooks(self):
         app = web.Application()
@@ -106,13 +120,13 @@ async def listen_webhooks(self):
     async def _receive_webhook(self, request: ClientRequest):
         topic = request.match_info["topic"]
         payload = await request.json()
-        await self.handle_webhook(topic, payload)
+        await self._handle_webhook(topic, payload)
         return web.Response(status=200)
 
-    async def handle_webhook(self, topic, payload):
+    async def _handle_webhook(self, topic, payload):
         logging.debug(f"Handle Webhook - {topic}", payload)
         pub.sendMessage(topic, payload=payload)
-        return web.Response(status=200)
+        # return web.Response(status=200)
 
     async def terminate(self):
         await self.client_session.close()

tutorials/aries-basic-controller/notebooks/alice/Part 2 - Aries Basic Controller.ipynb

@@ -227,6 +227,85 @@
     "agent_controller.register_listeners([some_listener], defaults=True)"
    ]
   },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### You can also add listeners one at a time.\n",
+    "\n",
+    "And add more than one listener for the same topic. This is useful in larger applications where multiple processes need to react to a trigger from the agent.\n",
+    "\n",
+    "![PyPubSub figure](https://pypubsub.readthedocs.io/en/v4.0.3/_images/pubsub_concept.png)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 1,
+   "metadata": {},
+   "outputs": [
+    {
+     "ename": "NameError",
+     "evalue": "name 'agent_controller' is not defined",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mNameError\u001b[0m                                 Traceback (most recent call last)",
+      "\u001b[0;32m<ipython-input-1-e5ea8d68770d>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m     10\u001b[0m }\n\u001b[1;32m     11\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m---> 12\u001b[0;31m \u001b[0magent_controller\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0madd_listener\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msome_listener2\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
+      "\u001b[0;31mNameError\u001b[0m: name 'agent_controller' is not defined"
+     ]
+    }
+   ],
+   "source": [
+    "def some_handler2(payload):\n",
+    "\n",
+    "    print(\"listener 2\")\n",
+    "    # Some other custom function\n",
+    "\n",
+    "    \n",
+    "some_listener2 = {\n",
+    "    \"handler\": some_handler2,\n",
+    "    \"topic\": \"sometopic\"\n",
+    "}\n",
+    "\n",
+    "agent_controller.add_listener(some_listener2)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### Listeners can be removed individually"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "agent_controller.remove_listener(some_listener2)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### And by topic\n",
+    "\n",
+    "If topic == None then all listeners will be removed."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "topic = \"basicmessages\"\n",
+    "agent_controller.remove_all(topic)\n",
+    "\n"
+   ]
+  },
   {
    "cell_type": "markdown",
    "metadata": {},

tutorials/aries-basic-controller/notebooks/alice/Part 3 - Establishing a Connection.ipynb

@@ -45,9 +45,17 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 1,
    "metadata": {},
-   "outputs": [],
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "IPython autoawait is `on`, and set to use `asyncio`\n"
+     ]
+    }
+   ],
    "source": [
     "%autoawait\n",
     "import time\n",
@@ -69,7 +77,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 2,
    "metadata": {},
    "outputs": [],
    "source": [
@@ -104,7 +112,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 3,
    "metadata": {},
    "outputs": [],
    "source": [
@@ -141,9 +149,48 @@
   },
   {
    "cell_type": "code",
-   "execution_count": null,
+   "execution_count": 5,
    "metadata": {},
-   "outputs": [],
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Connection ID 0b42a8c8-d25f-4a2e-9ba1-d7aa74ebb9d4\n",
+      "Invitation\n",
+      "{'@type': 'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/connections/1.0/invitation', '@id': '0f6250e8-f996-4a33-b798-9d5218cfd264', 'label': 'Alice', 'recipientKeys': ['5oBk3KC87hLHjbU14oetYAm2CRrxZTtiHituhyCCvEcs'], 'serviceEndpoint': 'http://172.17.0.1:8020'}\n",
+      "Connection Handler Called\n",
+      "Connection 0b42a8c8-d25f-4a2e-9ba1-d7aa74ebb9d4 in State invitation\n",
+      "Connection Handler 2 Called\n",
+      "Connection 0b42a8c8-d25f-4a2e-9ba1-d7aa74ebb9d4 in State invitation\n"
+     ]
+    }
+   ],
+   "source": [
+    "# Create Invitation\n",
+    "invite = await agent_controller.connections.create_invitation()\n",
+    "connection_id = invite[\"connection_id\"]\n",
+    "invite_message = invite['invitation']\n",
+    "print(\"Connection ID\", connection_id)\n",
+    "print(\"Invitation\")\n",
+    "print(invite_message)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 14,
+   "metadata": {},
+   "outputs": [
+    {
+     "name": "stdout",
+     "output_type": "stream",
+     "text": [
+      "Connection ID b2bed5c0-09e5-42e8-afaf-93304b84c5a6\n",
+      "Invitation\n",
+      "{'@type': 'did:sov:BzCbsNYhMrjHiqZDTUASHg;spec/connections/1.0/invitation', '@id': '66d5629d-5ea6-4a46-8f19-a5f0b1758a58', 'label': 'Alice', 'recipientKeys': ['31qfD7BQ3i2sAQzNR3BZocynrc7vgDwWYAQ7xXcYyKtk'], 'serviceEndpoint': 'http://172.17.0.1:8020'}\n"
+     ]
+    }
+   ],
    "source": [
     "# Create Invitation\n",
     "invite = await agent_controller.connections.create_invitation()\n",