execlib/notebooks/router.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "718618b7-132f-44e0-8cad-6a912a623c82",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "import logging\n",
    "from pathlib import Path\n",
    "from functools import partial\n",
    "\n",
    "from execlog import ChainRouter, Event\n",
    "from execlog.routers import PathRouter\n",
    "from execlog.listeners import PathListener\n",
    "\n",
    "logging.basicConfig(level=logging.DEBUG)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3fd536f-75d6-408d-88b2-1e4a1b62a51e",
   "metadata": {},
   "source": [
    "# Router setup\n",
    "Create individual \"frame\" routers, and attach them in a chain.\n",
    "\n",
    "A matching event will first be processed by matching callbacks in `router1` in parallel, blocking until all are completed, and then pass on to the next router (`router2`) to repeat the same process. This trajectory can occur in parallel for several events."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "f8b834c2-808e-4bd0-87eb-932dc42ba390",
   "metadata": {},
   "outputs": [],
   "source": [
    "router1 = PathRouter()\n",
    "router2 = PathRouter()\n",
    "router3 = PathRouter()\n",
    "\n",
    "chain_router = ChainRouter([router1, router2, router3])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "5a4a1e7a-ee8b-4eec-97dd-ed9abac73989",
   "metadata": {},
   "source": [
    "Register callbacks to each of the routers. The `Router` objects are of type `PathRouter`, so `.register` takes a path endpoint and a function that accepts `Event`s.\n",
    "\n",
    "Events are created with the registered endpoint path, and a `name` parameter with the filename at that path target. Here one callback is attached to Router 1, two to Router 2, and three to Router 3. A given matching event should have a trajectory that looks like:\n",
    "\n",
    "```\n",
    "Event -> Router-1 -> C1-1 -- blocking --> Router-2 --> C2-1 -- blocking --> Router-3 --> C3-1 -->\n",
    "                                                   \\-> C2-2 -/                        \\-> C3-2 -/\n",
    "                                                                                      \\-> C3-3 -/\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "c5bd34bb-ce1d-4719-a77a-ee13a95c3c78",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "router1.register('endpoint_proxy', partial(print, 'R1 ::'))\n",
    "router2.register('endpoint_proxy', partial(print, 'R2 ::'))\n",
    "router3.register('endpoint_proxy', partial(print, 'R3 ::'))\n",
    "\n",
    "events = [\n",
    "    Event(endpoint='endpoint_proxy', name='file1'),\n",
    "    Event(endpoint='endpoint_proxy', name='file2'),\n",
    "    Event(endpoint='endpoint_proxy', name='file3'),\n",
    "]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3f98138-e381-4a98-af33-0e5310f305ce",
   "metadata": {},
   "source": [
    "Submit the event list to an individual router. Each event will be handled in its own thread, until the thread limit is reached, at which point the events remain in a queue until they can be processed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "d7354e95-b7ce-4b8a-bcc2-cb8b161b8bae",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:execlog.router:Event [file1] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R1 ::')]\n",
      "INFO:execlog.router:Event [file2] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R1 ::')]\n",
      "INFO:execlog.router:Event [file3] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R1 ::')]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[<Future at 0x7802dbfc62d0 state=running>,\n",
       " <Future at 0x7802dbfc6900 state=running>,\n",
       " <Future at 0x7802e03660c0 state=running>]"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R1 :: Event(endpoint='endpoint_proxy', name='file1', action=None)\n",
      "R1 :: Event(endpoint='endpoint_proxy', name='file2', action=None)\n",
      "R1 :: Event(endpoint='endpoint_proxy', name='file3', action=None)\n",
      "R1 :: Event(endpoint='endpoint_proxy', name='fileA', action=[<flags.CREATE: 256>])\n"
     ]
    }
   ],
   "source": [
    "# multi-event single router\n",
    "router1.submit(events)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e4de6182-aae6-43e0-9d14-04f1e291fe41",
   "metadata": {},
   "source": [
    "Submit the event list to the chain router. Each event will be processed independently and in parallel, so long as there are threads available. Each event will make its way through the router chain, blocking until all matching callbacks for a given router are completed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "092a21a7-8052-4826-911c-6e4423b075ce",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:execlog.router:Event [file1] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R2 ::')]\n",
      "INFO:execlog.router:Event [file1] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R3 ::')]\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[<Future at 0x7802dbfec500 state=running>,\n",
       " <Future at 0x7802dbfecb90 state=running>,\n",
       " <Future at 0x7802dbfec800 state=running>]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:execlog.router:Event [file2] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R2 ::')]\n",
      "INFO:execlog.router:Event [file2] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R3 ::')]\n",
      "INFO:execlog.router:Event [file3] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R2 ::')]\n",
      "INFO:execlog.router:Event [file3] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R3 ::')]\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "R2 :: Event(endpoint='endpoint_proxy', name='file1', action=None)\n",
      "R2 :: Event(endpoint='endpoint_proxy', name='file2', action=None)\n",
      "R2 :: Event(endpoint='endpoint_proxy', name='file3', action=None)\n",
      "R3 :: Event(endpoint='endpoint_proxy', name='file1', action=None)\n",
      "R3 :: Event(endpoint='endpoint_proxy', name='file2', action=None)\n",
      "R3 :: Event(endpoint='endpoint_proxy', name='file3', action=None)\n",
      "R2 :: Event(endpoint='endpoint_proxy', name='fileA', action=[<flags.CREATE: 256>])\n",
      "R3 :: Event(endpoint='endpoint_proxy', name='fileA', action=[<flags.CREATE: 256>])\n"
     ]
    }
   ],
   "source": [
    "chain_router.submit(events)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "52f83d9a-7b12-4891-8b90-986a9ed399d7",
   "metadata": {
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:execlog.listeners.path:Starting listener for 1 paths\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "defaultdict(<function PathListener.__init__.<locals>.<lambda> at 0x7802dbfdfba0>, {1: defaultdict(<class 'int'>, {(PosixPath('endpoint_proxy'), PosixPath('.')): 1986})})\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:execlog.listeners.path:> Listening on path endpoint_proxy for flags [<flags.MODIFY: 2>, <flags.MOVED_FROM: 64>, <flags.MOVED_TO: 128>, <flags.CREATE: 256>, <flags.DELETE: 512>, <flags.DELETE_SELF: 1024>]\n"
     ]
    }
   ],
   "source": [
    "listener = chain_router.get_listener()\n",
    "listener.start()\n",
    "\n",
    "print(listener.watchmap)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "00bb2889-f266-4fb1-9a89-7d7539aba9cf",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "DEBUG:execlog.listeners.path:Watcher fired for [fileA]: [<flags.CREATE: 256>]\n",
      "INFO:execlog.router:Event [fileA] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R1 ::')]\n",
      "DEBUG:execlog.listeners.path:Watcher fired for [fileA]: [<flags.MODIFY: 2>]\n",
      "INFO:execlog.router:Event [fileA] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R2 ::')]\n",
      "INFO:execlog.router:Event [fileA] \u001b[1m\u001b[32mmatched [**/!(.*|*.tmp|*~)] under [endpoint_proxy] for [functools.partial(<built-in function print>, 'R3 ::')]\n",
      "DEBUG:execlog.listeners.path:Watcher fired for [fileA]: [<flags.DELETE: 512>]\n"
     ]
    }
   ],
   "source": [
    "file_a = Path('endpoint_proxy/fileA')\n",
    "file_a.write_text('test text')\n",
    "file_a.unlink()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4e993450-bdb7-4860-ba23-dbc2e5676ace",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "execlog",
   "language": "python",
   "name": "execlog"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}