Normal flow of read message, do something with the message, and then write the message to another queue

[mikeoliversinag]

I am trying to build a python component that uses Pika to get messages from a queue, do some processing on the message fetched, then write the processed message to another queue.

I reading through the examples it looked like the Asynchronous Consume and Asynchronous Publish examples were the most robust and I tried to use them.

`def on_message(self, _unused_channel, basic_deliver, properties, body):
"""Invoked by pika when a message is delivered from RabbitMQ. The
channel is passed for your convenience. The basic_deliver object that
is passed in carries the exchange, routing key, delivery tag and
a redelivered flag for the message. The properties passed in is an
instance of BasicProperties with the message properties and the body
is the message that was sent.

    :param pika.channel.Channel _unused_channel: The channel object
    :param pika.Spec.Basic.Deliver: basic_deliver method
    :param pika.Spec.BasicProperties: properties
    :param bytes body: The message body

    """
    
    LOGGER.debug('>>>>>>>>>>>>>>>> on_message exit result = %s', ProcessMessage.processIt(body))
    LOGGER.info('Received message # %s from %s: %s',
                basic_deliver.delivery_tag, properties.app_id, body)
    self.acknowledge_message(basic_deliver.delivery_tag)`

in the Asynchronous Consume example, is where I thought I could get access to the message to process it. At the end of the processIt(body) script I call the Asynchronous Publish.

Without the >>>>> line the Asynchrous Consume does fine, it fetches all the messages in the queue. But when I add that line either before or after the ack it only reads one message and writes one message and stops.

Is there an example of what I perceive as a common use of Pika to read a message, process it and write it to another queue?

[trexmo]

I have a similar request for an example of that kind of flow from MQ to process the message to write to a different queue

[trexmo]

Simple is as simple does.

import pika

# Establish a connection with RabbitMQ server
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare the input queue to read from
channel.queue_declare(queue='APQ')

# Declare the output queue to write to
channel.queue_declare(queue='C1Q')

# Define a callback function to process messages
def callback(ch, method, properties, body):
    # Process the message
    processed_message = body.upper()

    # Write the processed message to the output queue
    channel.basic_publish(exchange='', routing_key='C1Q', body=processed_message)

    # Acknowledge the message has been processed
    channel.basic_ack(delivery_tag=method.delivery_tag)

# Set up a consumer to read from the input queue
channel.basic_consume(queue='APQ', on_message_callback=callback)

# Start consuming messages
channel.start_consuming()