development/03_queue_worker_refactoring.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RabbitMQ worker - Refactor\n",
    "\n",
    "We have three use cases in our app for the queues:\n",
    "1. The API will trigger creation of a new job.  So this will likely be a new connection to RabbitMQ, sending a message to the queue, and then closing the connection.\n",
    "2. The first worker will listen to a message, do a job and then post to another queue.\n",
    "3. The second worker will only listen to a message.  \n",
    "\n",
    "Both 2 and 3 can use blocking connections, in that they can remain open and listen for messages.\n",
    "\n",
    "So for refactoring we can split this into these building blocks:\n",
    "* A base that creates a connection to RabbitMQ, creates a channel and ensures the right queues are available.\n",
    "* A component for the API to send a new message\n",
    "* A component for the first worker that listens to a message, does a job and posts to another queue\n",
    "* A component for the second worker that listens to a message and then does a job\n",
    "\n",
    "We will have two queues:\n",
    "* `start_fetch` - to trigger a new fetch job\n",
    "* `data_processing` - to trigger a data processing on fetched data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pika "
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Base - Our connection building block"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def open_channel():\n",
    "    \"\"\"Opens a connection, a channel, creates queues and then returns this to the caller.\"\"\"\n",
    "    credentials = pika.PlainCredentials('DEV_USER', 'CHANGE_ME')\n",
    "    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost', credentials=credentials))\n",
    "    channel = connection.channel()\n",
    "    channel.queue_declare(queue='start_fetch', durable=True)\n",
    "    channel.queue_declare(queue='data_processing', durable=True)\n",
    "    return channel"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component - The API sends to start worker 1"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Message sent to queue: start_fetch\n"
     ]
    }
   ],
   "source": [
    "QUEUE = 'start_fetch'\n",
    "\n",
    "def start_first_job():\n",
    "    \"\"\"Starts the first worker.\"\"\"\n",
    "\n",
    "    # Get a new channel from the base\n",
    "    channel = open_channel()\n",
    "\n",
    "    # Send an empty message to the queue\n",
    "    channel.basic_publish(\n",
    "        exchange='',\n",
    "        routing_key=QUEUE, \n",
    "        body=\"\"\n",
    "        )\n",
    "\n",
    "    print(\"Message sent to queue: {}\".format(QUEUE))\n",
    "\n",
    "    # Close cleanly\n",
    "    channel.close()\n",
    "\n",
    "\n",
    "start_first_job()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component - worker 1 job management"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "LISTEN_QUEUE = 'start_fetch'\n",
    "PUSH_QUEUE = 'data_processing'\n",
    "\n",
    "def run_fetch_worker():\n",
    "    \"\"\"Does the worker 1 fetch process and manages the job queue.\"\"\"\n",
    "\n",
    "    # Get a new channel from the base\n",
    "    channel = open_channel()\n",
    "\n",
    "    # Callback that would be run when a message is received\n",
    "    def callback(ch, method, properties, body):\n",
    "        print(\"Received message from queue: {}\".format(LISTEN_QUEUE))\n",
    "\n",
    "        # Do the fetch process\n",
    "        print(\"Doing the fetch process\")\n",
    "        data = []\n",
    "\n",
    "        # Send data to the next queue\n",
    "        print(\"Sending message to queue: {}\".format(PUSH_QUEUE))\n",
    "        ch.basic_publish(\n",
    "            exchange='',\n",
    "            routing_key=PUSH_QUEUE, \n",
    "            body=data\n",
    "            )\n",
    "        print(\"Message sent to queue: {}\".format(PUSH_QUEUE))\n",
    "\n",
    "        # Acknowledge the incoming message to remove it from the queue\n",
    "        ch.basic_ack(delivery_tag = method.delivery_tag)\n",
    "\n",
    "    # Register the callback\n",
    "    channel.basic_consume(queue=LISTEN_QUEUE, on_message_callback=callback)\n",
    "\n",
    "    # This is a blocking connection\n",
    "    channel.start_consuming()\n",
    "\n",
    "\n",
    "run_fetch_worker()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Component - worker 2 job management"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "LISTEN_QUEUE = 'data_processing'\n",
    "\n",
    "def run_processing_worker():\n",
    "    \"\"\"Does the worker 2 processing of data.\"\"\"\n",
    "\n",
    "    # Get a new channel from the base\n",
    "    channel = open_channel()\n",
    "\n",
    "    # Callback that would be run when a message is received\n",
    "    def callback(ch, method, properties, body):\n",
    "        print(\"Received message from queue: {}\".format(LISTEN_QUEUE))\n",
    "\n",
    "        # Do the data processing\n",
    "        print(\"Doing the data processing\")\n",
    "        result = {}\n",
    "\n",
    "        # Do something with the data\n",
    "        print(\"Finished my processing job\")\n",
    "\n",
    "        # Acknowledge the incoming message to remove it from the queue\n",
    "        ch.basic_ack(delivery_tag = method.delivery_tag)\n",
    "\n",
    "    # Register the callback\n",
    "    channel.basic_consume(queue=LISTEN_QUEUE, on_message_callback=callback)\n",
    "\n",
    "    # This is a blocking connection\n",
    "    channel.start_consuming()\n",
    "\n",
    "\n",
    "run_processing_worker()\n"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "These all seem to work.\n",
    "\n",
    "Now I can move them into the bases and components"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "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.11.1"
  },
  "orig_nbformat": 4,
  "vscode": {
   "interpreter": {
    "hash": "e312d00c0035e1a075cca129be038e4a92c47787dd29c5d0f304be6e82a509c0"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}