[EventCore] Use no_wait when publishing to queues (#1076)

* Use `no_wait` when publishing to the event core queues

* Update readme flag version

Author: abhinavsingh

README.md

@@ -2298,7 +2298,7 @@ usage: -m [-h] [--tunnel-hostname TUNNEL_HOSTNAME] [--tunnel-port TUNNEL_PORT]
           [--filtered-client-ips FILTERED_CLIENT_IPS]
           [--filtered-url-regex-config FILTERED_URL_REGEX_CONFIG]
 
-proxy.py v2.4.0rc9.dev12+g558a430.d20220126
+proxy.py v2.4.0rc10.dev13+g96428ae.d20220126
 
 options:
   -h, --help            show this help message and exit

proxy/core/event/dispatcher.py

@@ -37,7 +37,10 @@ class EventDispatcher:
     When --enable-events is used, dispatcher module is automatically
     started.  Most importantly, dispatcher module ensures that queue is
     not flooded and doesn't utilize too much memory in case there are no
-    event subscriber is enabled.
+    event subscribers for published messages.
+
+    EventDispatcher ensures that subscribers will receive the messages
+    in the order they are published.
     """
 
     def __init__(

proxy/core/event/queue.py

@@ -51,7 +51,7 @@ def publish(
         event_payload: Dict[str, Any],
         publisher_id: Optional[str] = None,
     ) -> None:
-        self.queue.put({
+        self.queue.put_nowait({
             'process_id': os.getpid(),
             'thread_id': threading.get_ident(),
             'event_timestamp': time.time(),
@@ -71,7 +71,7 @@ def subscribe(
         sub_id is a subscription identifier which must be globally
         unique.  channel MUST be a multiprocessing connection.
         """
-        self.queue.put({
+        self.queue.put_nowait({
             'event_name': eventNames.SUBSCRIBE,
             'event_payload': {'sub_id': sub_id, 'conn': channel},
         })
@@ -81,7 +81,7 @@ def unsubscribe(
             sub_id: str,
     ) -> None:
         """Unsubscribe by subscriber id."""
-        self.queue.put({
+        self.queue.put_nowait({
             'event_name': eventNames.UNSUBSCRIBE,
             'event_payload': {'sub_id': sub_id},
         })

tutorial/README.md

@@ -0,0 +1,9 @@
+# PROXY.PY TUTORIAL
+
+This directory contains Jupyter Notebook based tutorial.
+
+- [Welcome](welcome.ipynb)
+- [Http Parser](http_parser.ipynb)
+- [Responses](responses.ipynb)
+- [Requests](requests.ipynb)
+- [Eventing](eventing.ipynb)

tutorial/eventing.ipynb

@@ -0,0 +1,26 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Eventing\n",
+    "\n",
+    "Under [working with proxy.py](welcome.ipynb) we saw that plugin instances are not global.  So, how to manage shared data?  You have 2 options:\n",
+    "\n",
+    "1. Use in-built Python provided concepts of shared memory in a `multiprocessing` environment.\n",
+    "2. Use `proxy.py` provided eventing core\n",
+    "\n",
+    "`proxy.py` provides an in-built mechanism to publish and subscribe to messages between processes within a running instance.  With a pubsub facility in hand, we can develop variety of models e.g. multi-producer-single-subscriber model.  Imagine, publishing one or multiple messages from your plugin instances (per-request) and then processing them in near real-time within a single subscriber thread."
+   ]
+  }
+ ],
+ "metadata": {
+  "language_info": {
+   "name": "python"
+  },
+  "orig_nbformat": 4
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

tutorial/http_parser.ipynb

@@ -13,7 +13,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": 2,
    "metadata": {},
    "outputs": [
     {
@@ -30,7 +30,7 @@
     "from proxy.common.constants import HTTP_1_1\n",
     "\n",
     "get_request = HttpParser(httpParserTypes.REQUEST_PARSER)\n",
-    "get_request.parse(b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\n\\r\\n')\n",
+    "get_request.parse(memoryview(b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\n\\r\\n'))\n",
     "\n",
     "print(get_request.build())\n",
     "\n",
@@ -55,7 +55,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": 3,
    "metadata": {},
    "outputs": [
     {
@@ -69,7 +69,7 @@
    ],
    "source": [
     "proxy_request = HttpParser(httpParserTypes.REQUEST_PARSER)\n",
-    "proxy_request.parse(b'GET http://jaxl.com/ HTTP/1.1\\r\\nHost: jaxl.com\\r\\n\\r\\n')\n",
+    "proxy_request.parse(memoryview(b'GET http://jaxl.com/ HTTP/1.1\\r\\nHost: jaxl.com\\r\\n\\r\\n'))\n",
     "\n",
     "print(proxy_request.build())\n",
     "print(proxy_request.build(for_proxy=True))\n",
@@ -111,7 +111,7 @@
    ],
    "source": [
     "connect_request = HttpParser(httpParserTypes.REQUEST_PARSER)\n",
-    "connect_request.parse(b'CONNECT jaxl.com:443 HTTP/1.1\\r\\nHost: jaxl.com:443\\r\\n\\r\\n')\n",
+    "connect_request.parse(memoryview(b'CONNECT jaxl.com:443 HTTP/1.1\\r\\nHost: jaxl.com:443\\r\\n\\r\\n'))\n",
     "\n",
     "print(connect_request.build())\n",
     "print(connect_request.build(for_proxy=True))\n",

tutorial/requests.ipynb

@@ -22,7 +22,7 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nUser-Agent: proxy.py v2.4.0rc9.dev8+gea0253d.d20220126\\r\\n\\r\\n'\n"
+      "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\n\\r\\n'\n"
      ]
     }
    ],
@@ -55,7 +55,7 @@
     {
      "data": {
       "text/plain": [
-       "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nUser-Agent: proxy.py v2.4.0rc9.dev8+gea0253d.d20220126\\r\\n\\r\\n'"
+       "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nUser-Agent: proxy.py v2.4.0rc10.dev13+g96428ae.d20220126\\r\\n\\r\\n'"
       ]
      },
      "execution_count": 2,
@@ -154,7 +154,7 @@
     {
      "data": {
       "text/plain": [
-       "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nUser-Agent: proxy.py v2.4.0rc9.dev8+gea0253d.d20220126\\r\\nConnection: close\\r\\n\\r\\n'"
+       "b'GET / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nUser-Agent: proxy.py v2.4.0rc10.dev13+g96428ae.d20220126\\r\\nConnection: close\\r\\n\\r\\n'"
       ]
      },
      "execution_count": 5,
@@ -186,7 +186,7 @@
     {
      "data": {
       "text/plain": [
-       "b'POST / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nContent-Type: application/x-www-form-urlencoded\\r\\nContent-Length: 21\\r\\nUser-Agent: proxy.py v2.4.0rc9.dev8+gea0253d.d20220126\\r\\nConnection: close\\r\\n\\r\\nkey=value&hello=world'"
+       "b'POST / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nContent-Type: application/x-www-form-urlencoded\\r\\nContent-Length: 21\\r\\nUser-Agent: proxy.py v2.4.0rc10.dev13+g96428ae.d20220126\\r\\nConnection: close\\r\\n\\r\\nkey=value&hello=world'"
       ]
      },
      "execution_count": 6,
@@ -214,16 +214,16 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 9,
+   "execution_count": 7,
    "metadata": {},
    "outputs": [
     {
      "data": {
       "text/plain": [
-       "b'POST / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nTransfer-Encoding: chunked\\r\\nContent-Type: application/x-www-form-urlencoded\\r\\nUser-Agent: proxy.py v2.4.0rc9.dev8+gea0253d.d20220126\\r\\nConnection: close\\r\\n\\r\\n5\\r\\nkey=v\\r\\n5\\r\\nalue&\\r\\n5\\r\\nhello\\r\\n5\\r\\n=worl\\r\\n1\\r\\nd\\r\\n0\\r\\n\\r\\n'"
+       "b'POST / HTTP/1.1\\r\\nHost: jaxl.com\\r\\nTransfer-Encoding: chunked\\r\\nContent-Type: application/x-www-form-urlencoded\\r\\nUser-Agent: proxy.py v2.4.0rc10.dev13+g96428ae.d20220126\\r\\nConnection: close\\r\\n\\r\\n5\\r\\nkey=v\\r\\n5\\r\\nalue&\\r\\n5\\r\\nhello\\r\\n5\\r\\n=worl\\r\\n1\\r\\nd\\r\\n0\\r\\n\\r\\n'"
       ]
      },
-     "execution_count": 9,
+     "execution_count": 7,
      "metadata": {},
      "output_type": "execute_result"
     }

tutorial/responses.ipynb

@@ -15,7 +15,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 2,
+   "execution_count": 1,
    "metadata": {},
    "outputs": [
     {
@@ -49,7 +49,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 7,
+   "execution_count": 2,
    "metadata": {},
    "outputs": [
     {
@@ -77,7 +77,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 5,
+   "execution_count": 3,
    "metadata": {},
    "outputs": [
     {
@@ -107,7 +107,7 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 6,
+   "execution_count": 4,
    "metadata": {},
    "outputs": [
     {
@@ -142,14 +142,14 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 16,
+   "execution_count": 5,
    "metadata": {},
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "b'HTTP/1.1 200 OK\\r\\nX-Custom-Header: my value\\r\\nContent-Encoding: gzip\\r\\nContent-Length: 23\\r\\n\\r\\n\\x1f\\x8b\\x08\\x00F\\x0e\\xf1a\\x02\\xff\\xf3\\xf0\\xc0\\x02\\x00h\\x81?s\\x15\\x00\\x00\\x00'\n"
+      "b\"HTTP/1.1 200 OK\\r\\nX-Custom-Header: my value\\r\\nContent-Encoding: gzip\\r\\nContent-Length: 23\\r\\n\\r\\n\\x1f\\x8b\\x08\\x00\\x80'\\xf4a\\x02\\xff\\xf3\\xf0\\xc0\\x02\\x00h\\x81?s\\x15\\x00\\x00\\x00\"\n"
      ]
     }
    ],

tutorial/welcome.ipynb

@@ -14,9 +14,17 @@
     "\n",
     "Today, `proxy.py` has matured into a full blown networking library with focus on being lightweight, ability to deliver maximum performance while being extendible.  Unlike other Python servers, `proxy.py` doesn't need a `WSGI` or `UWSI` frontend, which then usually has to be placed behind a reverse proxy e.g. `Nginx` or `Apache`.  Of-course, `proxy.py` can be placed directly behind a load-balancer _(optionally capable of speaking HA proxy protocol)_.\n",
     "\n",
-    "## Asyncio\n",
+    "## Working with proxy.py\n",
     "\n",
-    "TBD\n",
+    "To work with `proxy.py`, you must follow these critical concepts:\n",
+    "\n",
+    "1. Avoid using synchronous IO operations within your code\n",
+    "\n",
+    "    `proxy.py` is asynchronous in nature and by making a synchronous call in your plugin code, you may block the entire core event loop.  For asynchronous operations, you must tie into the `proxy.py` event loop using the provided plugin APIs.\n",
+    "\n",
+    "2. Plugin instances are NOT global\n",
+    "\n",
+    "    Plugin instances are created for every request.  Hence, your plugin code must be written to handle execution of a single request.  `proxy.py` will internally take care of concurrency for you.\n",
     "\n",
     "## The Concept Of Work\n",
     "\n",