Publisher is getting blocked after after consuming 1st message_reply. It is not sending any messages to subscriber.
I have already tried that if I don't use 'start_consuming()' method publisher sends data continuously but it does not print reply_message from subscriber. If I used 'start_consuming()' method it just blocks the publisher and waits.
#publisher.py
#!/usr/bin/env python
import pika
import sys
import uuid
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
channel.queue_declare(queue='reply_queue', durable=True)
message = "Hello World!"
corr_id = str(uuid.uuid4())
def on_response(ch, method, properties, body):
print("----- On_response -----")
print("Received CORR_ID : ", properties.correlation_id)
if(corr_id == properties.correlation_id):
resp = body.decode('utf-8')
print("RESPONSE : ", resp)
ch.basic_ack(delivery_tag=method.delivery_tag)
def consume_response(channel):
channel.basic_consume(queue='reply_queue', on_message_callback=on_response)
channel.start_consuming()
while True:
channel.basic_publish(
exchange='',
routing_key='task_queue',
body=message,
properties=pika.BasicProperties(
reply_to='reply_queue',
correlation_id=corr_id
))
print(" [x] Sent %r" % message)
consume_response(channel)
#subscriber.py
#!/usr/bin/env python
import pika
import time
connection = pika.BlockingConnection(
pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
channel.queue_declare(queue='task_queue', durable=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
def callback(ch, method, properties, body):
print(" [x] Received %r" % body.decode('utf-8'))
#time.sleep(body.count(b'.'))
print(" [x] Done")
print("CORR_ID : ",str(properties.correlation_id))
print("Reply : ", str(properties.reply_to))
ch.basic_ack(delivery_tag=method.delivery_tag)
res = "Received "+str(body.decode('utf-8'))
ch.basic_publish(
exchange='',
routing_key='reply_queue',
properties = pika.BasicProperties(
correlation_id=\
properties.correlation_id
),
body=res
)
channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='task_queue', on_message_callback=callback)
channel.start_consuming()
Actual result is-
publisher subscriber
(1)----------publish message 'Hello'--------------> |
| <---------reply_to_publisher_queue---------------(2)
(3)prints_the message
(4)publisher waits for consuming messages
I want to design this-
(expected)
publisher subscriber
(1)----------publish message 'Hello'--------------> |
| <---------reply_to_publisher_queue---------------(2)
(3)prints_the message
(4)----------publish message 'Hello'--------------> |
| <---------reply_to_publisher_queue---------------(5)
.
.
.
continues....
Kindly help.
I replaced the line channel.start_consuming()
with channel.connection.process_data_events(time_limit=1)
in publisher.py
and it worked as I expected.
I have also followed https://github.com/pika/pika/issues/770.