Клиент-> Сервер + Клиент-> Сервер Передача сообщений

Question

Я пробую следующий сценарий, используя механизм запроса-ответа RabbitMQ

Шаг 1: Процесс ведет себя как простой клиент, который отправляет сообщение другому процессу B с указанием своей очереди ответа.Шаг 2: процесс B ведет себя как сервер для A и клиент для обработки C. Он получает сообщение от A и запрашивает другой процесс C. Шаг 3: процесс C принимает сообщение от B и отправляет ответ B на полученный ответ.очередь.Шаг 4: B добавляет что-то к этому ответу и отправляет ответ в очередь ответов A.

Я столкнулся с проблемой на шаге 3, т.е. процесс B не получает ответ от C.

Фрагмент кодаДля процесса A:

#!/usr/bin/env python
import pika
import uuid
import threading

class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        print ("on_response: ", threading.current_thread())
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        print("sending message:" , threading.current_thread())
        self.channel.basic_publish(exchange='',
                                   routing_key='serverclient_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=str(n))
        print ("waiting for response", threading.current_thread())
        while self.response is None:
            self.connection.process_data_events()
        return self.response

try:

    fibonacci_rpc = FibonacciRpcClient()

    print(" [x] Requesting ", threading.current_thread())
    response = fibonacci_rpc.call(30)
    print(" [.] Got %r" % response)

except Exception as e:
    print("Exception : ", e)
    raise(e)

Фрагмент кода для процесса B:

class FibonacciRpcClient2(object):

    def __init__(self):
        #self.connection = pika.SelectConnection(pika.ConnectionParameters(host='localhost'))
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
        #self.connection.ioloop.start()

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)
        self.channel.queue_declare(queue='serverclient_queue')
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(self.on_request, queue='serverclient_queue')
        print(" [x] Awaiting RPC requests: ", threading.current_thread())
        self.channel.start_consuming()
        self.response = ""

    def on_response(self, ch, method, props, body):
        print ("Response received from server")
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        print ("sending the message: thread : ", threading.current_thread())
        print ("queue:  : ", self.callback_queue)
        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)
        #with self.internal_lock:
        self.channel.basic_publish(exchange='',
                                   routing_key='server_queue',
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,
                                         correlation_id = self.corr_id,
                                         ),
                                   body=str(n))
        print("waiting for response", threading.current_thread())
        while self.response is None:
            print("poll data events. ... ", threading.current_thread())
            #self.connection.process_data_events()
            time.sleep(5)
        return self.response


    def on_request(self, ch, method, props, body):
        n = int(body)

        print ("server client is requested", threading.current_thread())

        print ("requesting the server")

        self.response = ""

        self.response =  self.call(30)

        print ("response from server", self.response)

        self.response = "hello from serverclient ::: " + self.response

        print ("sending the response : "+ self.response)

        ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=self.response)
        ch.basic_ack(delivery_tag = method.delivery_tag)


try:
    fibonacci_rpc = FibonacciRpcClient2()
except Exception as e:
    print ("Excepton : ", e)
    raise(e)

Код для процесса C выглядит следующим образом:

#!/usr/bin/env python
import pika
import time

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

channel = connection.channel()

channel.queue_declare(queue='server_queue')

def on_request(ch, method, props, body):
    n = int(body)

    print ("received the request")

    response = "hello from server"

    print ("sending the response : ", response)
    print ("reply queue : ", props.reply_to)

    for i in range(3):

        time.sleep(15)
        print ("sending to reply queue : ", i)

        ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id = \
                                                         props.correlation_id),
                     body=response)
        ch.basic_ack(delivery_tag = method.delivery_tag)
        print ("response sending over ... ")

try:
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(on_request, queue='server_queue')

    print(" [x] Awaiting RPC requests")
    channel.start_consuming()

except Excepiotn as e:
    print ("Exception: ", e)
    raise(e)