# Producer sends a message to the default exchange
# The default exchange ('') is a special direct exchange that routes to the queue
# matching the routing_key -- so routing_key='hello' delivers to the 'hello' queue
channel.basic_publish(
    exchange='',              # Default exchange -- every queue is automatically bound to it
    routing_key='hello',      # Which queue to deliver to (when using default exchange)
    body='Hello World!'       # The message payload -- can be any bytes (JSON, protobuf, plain text)
)

# Consumer receives messages from the 'hello' queue
def callback(ch, method, properties, body):
    # This function is called for each message delivered
    # ch = the channel object (used for acking)
    # method = delivery metadata (routing key, delivery tag, etc.)
    # properties = message properties (headers, content type, correlation ID)
    # body = the actual message content as bytes
    print(f"Received {body}")

channel.basic_consume(
    queue='hello',                  # Which queue to consume from
    on_message_callback=callback,   # Function called for each message
    auto_ack=True                   # DANGER: message deleted on delivery, not on processing
)

# Bind the queue to the exchange: "deliver messages with routing_key='error' to this queue"
# You can bind multiple queues to the same exchange with different routing keys
# You can also bind the same queue to multiple exchanges
channel.queue_bind(
    exchange='logs',           # The exchange to receive messages from
    queue=queue_name,          # The queue to deliver matching messages to
    routing_key='error'        # The matching criteria (exact match for direct exchange)
)

import pika

# Step 1: Establish a TCP connection to the RabbitMQ broker
# BlockingConnection is synchronous -- simple for learning, but for production
# consider async adapters (pika.SelectConnection) for better throughput
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')  # Default port is 5672
)

# Step 2: Open a channel on the connection
# Channels are lightweight and where all AMQP operations happen
channel = connection.channel()

# Step 3: Declare the queue -- this is IDEMPOTENT
# If the queue already exists with the same properties, this is a no-op
# If it does not exist, RabbitMQ creates it
# Both producer and consumer should declare the queue to handle startup order
channel.queue_declare(queue='hello')

# Step 4: Publish a message
# exchange='' uses the default exchange, which routes by queue name
# The message body is bytes -- encode JSON, protobuf, or plain text as needed
channel.basic_publish(
    exchange='',
    routing_key='hello',
    body='Hello World!'
)

print(" [x] Sent 'Hello World!'")

# Step 5: Always close the connection to flush buffers and release resources
connection.close()

import pika

# Connect to RabbitMQ (same as producer)
connection = pika.BlockingConnection(
    pika.ConnectionParameters('localhost')
)
channel = connection.channel()

# Declare the queue here too -- the consumer might start before the producer,
# and consuming from a non-existent queue causes an error
# This is idempotent: if the producer already created it, this is a no-op
channel.queue_declare(queue='hello')

# The callback function is invoked for each message delivered
def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    # With auto_ack=True, the message is already deleted from the queue
    # We do not need to (and cannot) acknowledge it manually

# Subscribe to the queue -- RabbitMQ will push messages to the callback
channel.basic_consume(
    queue='hello',
    on_message_callback=callback,
    auto_ack=True  # For this hello-world example only -- use manual ack in production
)

# start_consuming() blocks forever, waiting for messages
# Press Ctrl+C to exit
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

# Terminal 1: Start the consumer first (it will wait for messages)
python consumer.py

# Terminal 2: Send a message (the consumer will print it immediately)
python producer.py

# You can run the producer multiple times -- each message is delivered once
# You can also start multiple consumers -- messages are distributed round-robin

# Message is deleted from the queue the instant it is delivered to the consumer
# If the consumer crashes AFTER receiving but BEFORE processing, the message is gone forever
channel.basic_consume(
    queue='hello',
    on_message_callback=callback,
    auto_ack=True  # Only acceptable for non-critical, disposable messages
)

def callback(ch, method, properties, body):
    print(f"Processing {body}")
    # Do the actual work...
    time.sleep(5)
    # Acknowledge ONLY after processing is complete
    # If the worker crashes before this line, RabbitMQ re-delivers the message to another consumer
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(
    queue='tasks',
    on_message_callback=callback,
    auto_ack=False  # Manual ack -- the consumer controls when the message is deleted
)

# Declare queue as durable -- the queue DEFINITION survives a broker restart
# Without this, the queue itself disappears on restart (even if messages were persistent)
channel.queue_declare(queue='task_queue', durable=True)

# Mark the message as persistent -- the message CONTENT is written to disk
# delivery_mode=1 is transient (in-memory only, faster but lost on restart)
# delivery_mode=2 is persistent (written to disk, slower but survives restart)
channel.basic_publish(
    exchange='',
    routing_key='task_queue',
    body=message,
    properties=pika.BasicProperties(
        delivery_mode=2  # Persistent -- broker writes this message to disk
    )
)

# Limit to 1 unacknowledged message per worker
# RabbitMQ will not send the next message until this worker acks the current one
channel.basic_qos(prefetch_count=1)

def callback(ch, method, properties, body):
    print(f"Processing {body}")
    time.sleep(body.count(b'.'))  # Simulate variable-duration work
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(
    queue='task_queue',
    on_message_callback=callback
)

# Declare a direct exchange for routing log messages by severity
channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

# Publish a message with routing_key='error'
# Only queues bound with the exact key 'error' will receive this message
channel.basic_publish(
    exchange='direct_logs',
    routing_key='error',       # Exact match -- only 'error' bindings get this
    body='Error message'
)

# Bind a queue to receive only 'error' messages from this exchange
# You can bind the same queue with multiple routing keys to receive multiple levels
channel.queue_bind(
    exchange='direct_logs',
    queue=queue_name,
    routing_key='error'        # This queue only gets messages with routing_key='error'
)

# Declare a topic exchange for fine-grained log routing
channel.exchange_declare(exchange='topic_logs', exchange_type='topic')

# Publish with a structured routing key: <facility>.<severity>
channel.basic_publish(
    exchange='topic_logs',
    routing_key='kern.critical',  # Structured key: facility.severity
    body='Critical kernel error'
)

# Bind with a pattern -- * matches exactly one word, # matches zero or more
channel.queue_bind(
    exchange='topic_logs',
    queue=queue_name,
    routing_key='kern.*'  # Matches kern.critical, kern.info, kern.warning
                          # Does NOT match kern.sub.category (that is two words)
)

# More wildcard examples:
# '*.critical'  -- matches kern.critical, app.critical, db.critical
# 'kern.#'      -- matches kern, kern.critical, kern.sub.category
# '#'           -- matches everything (acts like a fanout)

# Declare a fanout exchange for broadcasting events
channel.exchange_declare(exchange='logs', exchange_type='fanout')

# Publish -- the routing_key is ignored, but the API still requires the parameter
channel.basic_publish(
    exchange='logs',
    routing_key='',            # Ignored by fanout -- set to empty string by convention
    body='Log message'
)

# Every queue bound to this exchange receives a copy of every message
# No routing key needed in the binding either
channel.queue_bind(exchange='logs', queue=queue_name)

# Publish with headers that describe the message's attributes
channel.basic_publish(
    exchange='headers_exchange',
    routing_key='',            # Routing key is ignored by headers exchanges
    body='Message',
    properties=pika.BasicProperties(
        headers={'format': 'pdf', 'type': 'report'}  # Routing criteria in headers
    )
)

# Bind with header matching rules
channel.queue_bind(
    exchange='headers_exchange',
    queue=queue_name,
    arguments={
        'x-match': 'all',     # 'all' = every header must match (AND logic)
                               # 'any' = at least one header must match (OR logic)
        'format': 'pdf',
        'type': 'report'
    }
)

# producer.py
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)

message = ' '.join(sys.argv[1:]) or "Hello World!"
channel.basic_publish(
    exchange='',
    routing_key='task_queue',
    body=message,
    properties=pika.BasicProperties(delivery_mode=2)  # Persistent
)

print(f" [x] Sent {message}")
connection.close()

# worker.py
import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='task_queue', durable=True)
channel.basic_qos(prefetch_count=1)

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    time.sleep(body.count(b'.'))  # Simulate work
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='task_queue', on_message_callback=callback)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

# emit_log.py
import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs', exchange_type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
channel.basic_publish(exchange='logs', routing_key='', body=message)

print(f" [x] Sent {message}")
connection.close()

# receive_logs.py
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='logs', exchange_type='fanout')

# Exclusive queue (auto-deleted when consumer disconnects)
result = channel.queue_declare(queue='', exclusive=True)
queue_name = result.method.queue

channel.queue_bind(exchange='logs', queue=queue_name)

def callback(ch, method, properties, body):
    print(f" [x] {body.decode()}")

channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)

print(' [*] Waiting for logs. To exit press CTRL+C')
channel.start_consuming()