0-Pre-requisites/3-RabbitMQ.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# RabbitMQ Basics with Docker: A Hands-On Tutorial\n",
    "\n",
    "## Introduction\n",
    "\n",
    "RabbitMQ is a robust messaging broker that allows applications to communicate asynchronously. It supports a variety of messaging protocols and is widely used for managing message queues between distributed systems.\n",
    "\n",
    "In this tutorial, you will:\n",
    "- Set up RabbitMQ using Docker.\n",
    "- Interact with RabbitMQ using Python to send and receive messages.\n",
    "- Understand basic concepts like producers, consumers, and queues.\n",
    "\n",
    "## Prerequisites\n",
    "\n",
    "Ensure that Docker is installed on your system. You can install Docker by following [this guide](https://docs.docker.com/get-docker/).\n",
    "\n",
    "Additionally, ensure Python is installed with the `pika` library, which is used to interact with RabbitMQ.\n",
    "\n",
    "Install `pika`:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Requirement already satisfied: pika in c:\\python\\python311\\lib\\site-packages (1.3.2)\n",
      "Note: you may need to restart the kernel to use updated packages.\n"
     ]
    }
   ],
   "source": [
    "%pip install pika"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Setting Up RabbitMQ Server Using Docker\n",
    "\n",
    "We will use Docker to set up a RabbitMQ server, which is easy and avoids any manual installation.\n",
    "\n",
    "### Pull RabbitMQ Docker Image\n",
    "\n",
    "We’ll start by pulling the official RabbitMQ Docker image."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "3-management: Pulling from library/rabbitmq\n",
      "Digest: sha256:b55a09baeef15e567a9403ce6c3c84ab06a97b48265954ae99876892c04fdfdf\n",
      "Status: Image is up to date for rabbitmq:3-management\n",
      "docker.io/library/rabbitmq:3-management\n"
     ]
    }
   ],
   "source": [
    "!docker pull rabbitmq:3-management"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `3-management` tag includes the RabbitMQ Management Plugin, allowing you to monitor and manage the RabbitMQ server through a web interface.\n",
    "\n",
    "### Run the RabbitMQ Container\n",
    "\n",
    "Now, let's run RabbitMQ in a Docker container. We will expose two ports:\n",
    "- **5672**: The default port for RabbitMQ.\n",
    "- **15672**: The web management interface."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "336d1ecdff8b8f9d427006fc58cf948ded14e0fdcf3a665d1c12be41c12a166f\n"
     ]
    }
   ],
   "source": [
    "!docker run -d --name rabbitmq-server -p 5672:5672 -p 15672:15672 rabbitmq:3-management"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Access the RabbitMQ Management Interface\n",
    "\n",
    "After running the above command, RabbitMQ will be accessible through the web interface. Open a browser and visit:\n",
    "[http://localhost:15672](http://localhost:15672)\n",
    "\n",
    "Use the default credentials to log in:\n",
    "- Username: `guest`\n",
    "- Password: `guest`\n",
    "\n",
    "## Understanding RabbitMQ Concepts\n",
    "\n",
    "Before we jump into code, let's review the core RabbitMQ concepts:\n",
    "\n",
    "- **Producer**: A producer sends messages to RabbitMQ.\n",
    "- **Consumer**: A consumer receives messages from RabbitMQ.\n",
    "- **Queue**: A buffer that stores messages until they are consumed.\n",
    "- **Exchange**: A mechanism that routes messages to one or more queues based on rules.\n",
    "\n",
    "## Sending Messages to RabbitMQ (Producer)\n",
    "\n",
    "Now that the RabbitMQ server is up and running, let's write a Python script to send a message to a queue.\n",
    "\n",
    "First, we'll import the necessary libraries:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pika"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Establish Connection to RabbitMQ\n",
    "\n",
    "We need to establish a connection to the RabbitMQ server running on `localhost`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_connection():\n",
    "    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))\n",
    "    return connection\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Sending a Message to a Queue\n",
    "\n",
    "Let's write a function to send a message to a queue called `hello`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " [x] Sent Hello from RabbitMQ!\n"
     ]
    }
   ],
   "source": [
    "def send_message(message):\n",
    "    connection = get_connection()\n",
    "    channel = connection.channel()\n",
    "\n",
    "    # Declare a queue\n",
    "    channel.queue_declare(queue='hello')\n",
    "\n",
    "    # Publish a message to the queue\n",
    "    channel.basic_publish(exchange='',\n",
    "                          routing_key='hello',\n",
    "                          body=message)\n",
    "    print(f\" [x] Sent {message}\")\n",
    "    connection.close()\n",
    "\n",
    "# Test sending a message\n",
    "send_message(\"Hello from RabbitMQ!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Explanation**:\n",
    "- **`queue_declare`**: This ensures that the queue exists before we send a message to it. If the queue doesn’t exist, it will be created.\n",
    "- **`basic_publish`**: This sends the message to the `hello` queue.\n",
    "\n",
    "## Receiving Messages from RabbitMQ (Consumer)\n",
    "\n",
    "Now let's create a consumer that listens for messages from the `hello` queue.\n",
    "\n",
    "### Defining a Consumer"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " [*] Waiting for messages. To exit press CTRL+C\n",
      " [x] Received b'Hello from RabbitMQ!'\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[9], line 21\u001b[0m\n\u001b[0;32m     18\u001b[0m     channel\u001b[38;5;241m.\u001b[39mstart_consuming()\n\u001b[0;32m     20\u001b[0m \u001b[38;5;66;03m# Start receiving messages\u001b[39;00m\n\u001b[1;32m---> 21\u001b[0m \u001b[43mreceive_message\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "Cell \u001b[1;32mIn[9], line 18\u001b[0m, in \u001b[0;36mreceive_message\u001b[1;34m()\u001b[0m\n\u001b[0;32m     13\u001b[0m channel\u001b[38;5;241m.\u001b[39mbasic_consume(queue\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mhello\u001b[39m\u001b[38;5;124m'\u001b[39m,\n\u001b[0;32m     14\u001b[0m                       on_message_callback\u001b[38;5;241m=\u001b[39mcallback,\n\u001b[0;32m     15\u001b[0m                       auto_ack\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mTrue\u001b[39;00m)\n\u001b[0;32m     17\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m [*] Waiting for messages. To exit press CTRL+C\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[1;32m---> 18\u001b[0m \u001b[43mchannel\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mstart_consuming\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:1883\u001b[0m, in \u001b[0;36mBlockingChannel.start_consuming\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1880\u001b[0m \u001b[38;5;66;03m# Process events as long as consumers exist on this channel\u001b[39;00m\n\u001b[0;32m   1881\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_consumer_infos:\n\u001b[0;32m   1882\u001b[0m     \u001b[38;5;66;03m# This will raise ChannelClosed if channel is closed by broker\u001b[39;00m\n\u001b[1;32m-> 1883\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_process_data_events\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtime_limit\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;28;43;01mNone\u001b[39;49;00m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:2044\u001b[0m, in \u001b[0;36mBlockingChannel._process_data_events\u001b[1;34m(self, time_limit)\u001b[0m\n\u001b[0;32m   2025\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_process_data_events\u001b[39m(\u001b[38;5;28mself\u001b[39m, time_limit):\n\u001b[0;32m   2026\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"Wrapper for `BlockingConnection.process_data_events()` with common\u001b[39;00m\n\u001b[0;32m   2027\u001b[0m \u001b[38;5;124;03m    channel-specific logic that raises ChannelClosed if broker closed this\u001b[39;00m\n\u001b[0;32m   2028\u001b[0m \u001b[38;5;124;03m    channel.\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m   2042\u001b[0m \n\u001b[0;32m   2043\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[1;32m-> 2044\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mconnection\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mprocess_data_events\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtime_limit\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mtime_limit\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   2045\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mis_closed \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closing_reason,\n\u001b[0;32m   2046\u001b[0m                                      exceptions\u001b[38;5;241m.\u001b[39mChannelClosedByBroker):\n\u001b[0;32m   2047\u001b[0m         LOGGER\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mChannel close by broker detected, raising \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[38;5;124m; \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[38;5;124m'\u001b[39m,\n\u001b[0;32m   2048\u001b[0m                      \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closing_reason, \u001b[38;5;28mself\u001b[39m)\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:842\u001b[0m, in \u001b[0;36mBlockingConnection.process_data_events\u001b[1;34m(self, time_limit)\u001b[0m\n\u001b[0;32m    838\u001b[0m common_terminator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mlambda\u001b[39;00m: \u001b[38;5;28mbool\u001b[39m(dispatch_acquired \u001b[38;5;129;01mand\u001b[39;00m\n\u001b[0;32m    839\u001b[0m                                  (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_channels_pending_dispatch \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    840\u001b[0m                                   \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_ready_events))\n\u001b[0;32m    841\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m time_limit \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m--> 842\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_flush_output\u001b[49m\u001b[43m(\u001b[49m\u001b[43mcommon_terminator\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    843\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    844\u001b[0m     \u001b[38;5;28;01mwith\u001b[39;00m _IoloopTimerContext(time_limit, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_impl) \u001b[38;5;28;01mas\u001b[39;00m timer:\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:514\u001b[0m, in \u001b[0;36mBlockingConnection._flush_output\u001b[1;34m(self, *waiters)\u001b[0m\n\u001b[0;32m    512\u001b[0m \u001b[38;5;66;03m# Process I/O until our completion condition is satisfied\u001b[39;00m\n\u001b[0;32m    513\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m is_done():\n\u001b[1;32m--> 514\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_impl\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mioloop\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    515\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_impl\u001b[38;5;241m.\u001b[39mioloop\u001b[38;5;241m.\u001b[39mprocess_timeouts()\n\u001b[0;32m    517\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closed_result\u001b[38;5;241m.\u001b[39mready:\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\select_connection.py:579\u001b[0m, in \u001b[0;36mIOLoop.poll\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    572\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mpoll\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[0;32m    573\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"[Extension] Wait for events of interest on registered file\u001b[39;00m\n\u001b[0;32m    574\u001b[0m \u001b[38;5;124;03m    descriptors until an event of interest occurs or next timer deadline or\u001b[39;00m\n\u001b[0;32m    575\u001b[0m \u001b[38;5;124;03m    `_PollerBase._MAX_POLL_TIMEOUT`, whichever is sooner, and dispatch the\u001b[39;00m\n\u001b[0;32m    576\u001b[0m \u001b[38;5;124;03m    corresponding event handlers.\u001b[39;00m\n\u001b[0;32m    577\u001b[0m \n\u001b[0;32m    578\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[1;32m--> 579\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_poller\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\select_connection.py:953\u001b[0m, in \u001b[0;36mSelectPoller.poll\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    949\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m    950\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mREAD] \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    951\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mWRITE] \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    952\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mERROR]):\n\u001b[1;32m--> 953\u001b[0m         read, write, error \u001b[38;5;241m=\u001b[39m select\u001b[38;5;241m.\u001b[39mselect(\n\u001b[0;32m    954\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mREAD],\n\u001b[0;32m    955\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mWRITE],\n\u001b[0;32m    956\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mERROR], \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_max_wait())\n\u001b[0;32m    957\u001b[0m     \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    958\u001b[0m         \u001b[38;5;66;03m# NOTE When called without any FDs, select fails on\u001b[39;00m\n\u001b[0;32m    959\u001b[0m         \u001b[38;5;66;03m# Windows with error 10022, 'An invalid argument was\u001b[39;00m\n\u001b[0;32m    960\u001b[0m         \u001b[38;5;66;03m# supplied'.\u001b[39;00m\n\u001b[0;32m    961\u001b[0m         time\u001b[38;5;241m.\u001b[39msleep(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_max_wait())\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "def receive_message():\n",
    "    connection = get_connection()\n",
    "    channel = connection.channel()\n",
    "\n",
    "    # Declare the same queue\n",
    "    channel.queue_declare(queue='hello')\n",
    "\n",
    "    # Callback function to process messages\n",
    "    def callback(ch, method, properties, body):\n",
    "        print(f\" [x] Received {body}\")\n",
    "\n",
    "    # Tell RabbitMQ to deliver messages from the queue\n",
    "    channel.basic_consume(queue='hello',\n",
    "                          on_message_callback=callback,\n",
    "                          auto_ack=True)\n",
    "\n",
    "    print(' [*] Waiting for messages. To exit press Interrupt')\n",
    "    channel.start_consuming()\n",
    "\n",
    "# Start receiving messages\n",
    "receive_message()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Example producer script"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Writing send.py\n"
     ]
    }
   ],
   "source": [
    "%%writefile send.py\n",
    "import pika\n",
    "def get_connection():\n",
    "    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))\n",
    "    return connection\n",
    "\n",
    "def send_message(message):\n",
    "    connection = get_connection()\n",
    "    channel = connection.channel()\n",
    "\n",
    "    # Declare a queue\n",
    "    channel.queue_declare(queue='hello')\n",
    "\n",
    "    # Publish a message to the queue\n",
    "    channel.basic_publish(exchange='',\n",
    "                          routing_key='hello',\n",
    "                          body=message)\n",
    "    print(f\" [x] Sent {message}\")\n",
    "    connection.close()\n",
    "\n",
    "# Test sending a message\n",
    "send_message(\"Hello from RabbitMQ!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Sending and Receiving Messages Together\n",
    "\n",
    "Now that we have both producer and consumer code, let's see them working together:\n",
    "1. Run the consumer in the next cell.\n",
    "2. Open a terminal in the current folder, and run `python send.py`. Repeat as you wish, to see the multiple messages arriving."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      " [*] Waiting for messages. To exit press CTRL+C\n",
      " [x] Received b'Hello from RabbitMQ!'\n",
      " [x] Received b'Hello from RabbitMQ!'\n",
      " [x] Received b'Hello from RabbitMQ!'\n",
      " [x] Received b'Hello from RabbitMQ!'\n"
     ]
    },
    {
     "ename": "KeyboardInterrupt",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m                         Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[11], line 2\u001b[0m\n\u001b[0;32m      1\u001b[0m \u001b[38;5;66;03m# Start receiving messages\u001b[39;00m\n\u001b[1;32m----> 2\u001b[0m \u001b[43mreceive_message\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "Cell \u001b[1;32mIn[9], line 18\u001b[0m, in \u001b[0;36mreceive_message\u001b[1;34m()\u001b[0m\n\u001b[0;32m     13\u001b[0m channel\u001b[38;5;241m.\u001b[39mbasic_consume(queue\u001b[38;5;241m=\u001b[39m\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mhello\u001b[39m\u001b[38;5;124m'\u001b[39m,\n\u001b[0;32m     14\u001b[0m                       on_message_callback\u001b[38;5;241m=\u001b[39mcallback,\n\u001b[0;32m     15\u001b[0m                       auto_ack\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mTrue\u001b[39;00m)\n\u001b[0;32m     17\u001b[0m \u001b[38;5;28mprint\u001b[39m(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124m [*] Waiting for messages. To exit press CTRL+C\u001b[39m\u001b[38;5;124m'\u001b[39m)\n\u001b[1;32m---> 18\u001b[0m \u001b[43mchannel\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mstart_consuming\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:1883\u001b[0m, in \u001b[0;36mBlockingChannel.start_consuming\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m   1880\u001b[0m \u001b[38;5;66;03m# Process events as long as consumers exist on this channel\u001b[39;00m\n\u001b[0;32m   1881\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_consumer_infos:\n\u001b[0;32m   1882\u001b[0m     \u001b[38;5;66;03m# This will raise ChannelClosed if channel is closed by broker\u001b[39;00m\n\u001b[1;32m-> 1883\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_process_data_events\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtime_limit\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[38;5;28;43;01mNone\u001b[39;49;00m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:2044\u001b[0m, in \u001b[0;36mBlockingChannel._process_data_events\u001b[1;34m(self, time_limit)\u001b[0m\n\u001b[0;32m   2025\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21m_process_data_events\u001b[39m(\u001b[38;5;28mself\u001b[39m, time_limit):\n\u001b[0;32m   2026\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"Wrapper for `BlockingConnection.process_data_events()` with common\u001b[39;00m\n\u001b[0;32m   2027\u001b[0m \u001b[38;5;124;03m    channel-specific logic that raises ChannelClosed if broker closed this\u001b[39;00m\n\u001b[0;32m   2028\u001b[0m \u001b[38;5;124;03m    channel.\u001b[39;00m\n\u001b[1;32m   (...)\u001b[0m\n\u001b[0;32m   2042\u001b[0m \n\u001b[0;32m   2043\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[1;32m-> 2044\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mconnection\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mprocess_data_events\u001b[49m\u001b[43m(\u001b[49m\u001b[43mtime_limit\u001b[49m\u001b[38;5;241;43m=\u001b[39;49m\u001b[43mtime_limit\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m   2045\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mis_closed \u001b[38;5;129;01mand\u001b[39;00m \u001b[38;5;28misinstance\u001b[39m(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closing_reason,\n\u001b[0;32m   2046\u001b[0m                                      exceptions\u001b[38;5;241m.\u001b[39mChannelClosedByBroker):\n\u001b[0;32m   2047\u001b[0m         LOGGER\u001b[38;5;241m.\u001b[39mdebug(\u001b[38;5;124m'\u001b[39m\u001b[38;5;124mChannel close by broker detected, raising \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[38;5;124m; \u001b[39m\u001b[38;5;132;01m%r\u001b[39;00m\u001b[38;5;124m'\u001b[39m,\n\u001b[0;32m   2048\u001b[0m                      \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closing_reason, \u001b[38;5;28mself\u001b[39m)\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:842\u001b[0m, in \u001b[0;36mBlockingConnection.process_data_events\u001b[1;34m(self, time_limit)\u001b[0m\n\u001b[0;32m    838\u001b[0m common_terminator \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mlambda\u001b[39;00m: \u001b[38;5;28mbool\u001b[39m(dispatch_acquired \u001b[38;5;129;01mand\u001b[39;00m\n\u001b[0;32m    839\u001b[0m                                  (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_channels_pending_dispatch \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    840\u001b[0m                                   \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_ready_events))\n\u001b[0;32m    841\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m time_limit \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m--> 842\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_flush_output\u001b[49m\u001b[43m(\u001b[49m\u001b[43mcommon_terminator\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    843\u001b[0m \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    844\u001b[0m     \u001b[38;5;28;01mwith\u001b[39;00m _IoloopTimerContext(time_limit, \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_impl) \u001b[38;5;28;01mas\u001b[39;00m timer:\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\blocking_connection.py:514\u001b[0m, in \u001b[0;36mBlockingConnection._flush_output\u001b[1;34m(self, *waiters)\u001b[0m\n\u001b[0;32m    512\u001b[0m \u001b[38;5;66;03m# Process I/O until our completion condition is satisfied\u001b[39;00m\n\u001b[0;32m    513\u001b[0m \u001b[38;5;28;01mwhile\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m is_done():\n\u001b[1;32m--> 514\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_impl\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mioloop\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n\u001b[0;32m    515\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_impl\u001b[38;5;241m.\u001b[39mioloop\u001b[38;5;241m.\u001b[39mprocess_timeouts()\n\u001b[0;32m    517\u001b[0m \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_closed_result\u001b[38;5;241m.\u001b[39mready:\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\select_connection.py:579\u001b[0m, in \u001b[0;36mIOLoop.poll\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    572\u001b[0m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mpoll\u001b[39m(\u001b[38;5;28mself\u001b[39m):\n\u001b[0;32m    573\u001b[0m \u001b[38;5;250m    \u001b[39m\u001b[38;5;124;03m\"\"\"[Extension] Wait for events of interest on registered file\u001b[39;00m\n\u001b[0;32m    574\u001b[0m \u001b[38;5;124;03m    descriptors until an event of interest occurs or next timer deadline or\u001b[39;00m\n\u001b[0;32m    575\u001b[0m \u001b[38;5;124;03m    `_PollerBase._MAX_POLL_TIMEOUT`, whichever is sooner, and dispatch the\u001b[39;00m\n\u001b[0;32m    576\u001b[0m \u001b[38;5;124;03m    corresponding event handlers.\u001b[39;00m\n\u001b[0;32m    577\u001b[0m \n\u001b[0;32m    578\u001b[0m \u001b[38;5;124;03m    \"\"\"\u001b[39;00m\n\u001b[1;32m--> 579\u001b[0m     \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43m_poller\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mpoll\u001b[49m\u001b[43m(\u001b[49m\u001b[43m)\u001b[49m\n",
      "File \u001b[1;32mc:\\Python\\Python311\\Lib\\site-packages\\pika\\adapters\\select_connection.py:953\u001b[0m, in \u001b[0;36mSelectPoller.poll\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    949\u001b[0m \u001b[38;5;28;01mtry\u001b[39;00m:\n\u001b[0;32m    950\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m (\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mREAD] \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    951\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mWRITE] \u001b[38;5;129;01mor\u001b[39;00m\n\u001b[0;32m    952\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mERROR]):\n\u001b[1;32m--> 953\u001b[0m         read, write, error \u001b[38;5;241m=\u001b[39m select\u001b[38;5;241m.\u001b[39mselect(\n\u001b[0;32m    954\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mREAD],\n\u001b[0;32m    955\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mWRITE],\n\u001b[0;32m    956\u001b[0m             \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_fd_events[PollEvents\u001b[38;5;241m.\u001b[39mERROR], \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_max_wait())\n\u001b[0;32m    957\u001b[0m     \u001b[38;5;28;01melse\u001b[39;00m:\n\u001b[0;32m    958\u001b[0m         \u001b[38;5;66;03m# NOTE When called without any FDs, select fails on\u001b[39;00m\n\u001b[0;32m    959\u001b[0m         \u001b[38;5;66;03m# Windows with error 10022, 'An invalid argument was\u001b[39;00m\n\u001b[0;32m    960\u001b[0m         \u001b[38;5;66;03m# supplied'.\u001b[39;00m\n\u001b[0;32m    961\u001b[0m         time\u001b[38;5;241m.\u001b[39msleep(\u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39m_get_max_wait())\n",
      "\u001b[1;31mKeyboardInterrupt\u001b[0m: "
     ]
    }
   ],
   "source": [
    "# Start receiving messages\n",
    "receive_message()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## RabbitMQ Web Management Interface\n",
    "\n",
    "You can monitor the status of queues, connections, and messages through the RabbitMQ Management Interface:\n",
    "\n",
    "- Go to http://localhost:15672\n",
    "- Log in with the default credentials (`guest`/`guest`).\n",
    "- You can see details about the `hello` queue and the messages being sent and received.\n",
    "\n",
    "## Stopping the RabbitMQ Docker Container\n",
    "\n",
    "After you're done with RabbitMQ, you can stop and remove the Docker container:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "rabbitmq-server\n",
      "rabbitmq-server\n"
     ]
    }
   ],
   "source": [
    "!docker stop rabbitmq-server\n",
    "!docker rm rabbitmq-server"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "\n",
    "This will stop the RabbitMQ server and remove the container to free up resources.\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}