I'm hoping someone can help he with my RabbitMQ problem; I've got some
code based on the tutorial #6 code, and it's not doing what I'm expecting.
I tried to create a shell where I can send strings to the server and
receive the MD5 digest, but I only ever seem to get the first reply.

Here's the source:

##########################################################################
# server.py

import pika
import hashlib

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

channel = connection.channel()
channel.queue_declare(queue='my_request')

def on_request(ch, method, props, body):

response = hashlib.md5(body).hexdigest()

print("Received '%r', sending '%r'" % (body, response))

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)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='my_request')

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

##########################################################################
# client.py

import pika
import uuid

class RabbitClient(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(" [.] Received correlation ID %r" % (props.correlation_id,))
if self.corr_id == props.correlation_id:
self.response = body
self.channel.stop_consuming()
else:
print(" [~] Correlation IDs do not match")

def call(self, n):
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(exchange='',
routing_key='my_request',
properties=pika.BasicProperties(
reply_to=self.callback_queue,
correlation_id=self.corr_id,
),
body=n)
self.channel.start_consuming()
return self.response

auth = RabbitClient()

while True:
send = raw_input('> ')
if send == 'quit': break
print(" [x] sending '%s'" % (send))
response = auth.call(send)
print(" [^] sent with correlation ID %s" % (auth.corr_id,))
print(" [.] Got %r" % (response,))
##########################################################################

Here's the output from the server:

##########################################################################
[x] Awaiting requests
Received ''Hello World'', sending ''b10a8db164e0754105b7a99be72e3fe5''
Received ''Hello Rabbit'', sending ''5bec01f45b43f6202e84242fcb71b04f''
##########################################################################

And here's the output from the server:
##########################################################################
Hello World
[x] sending 'Hello World'
[.] Received correlation ID '1805f131-1fbc-40a3-86fb-2a113b5f9b9a'
[^] sent with correlation ID 1805f131-1fbc-40a3-86fb-2a113b5f9b9a
[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
Hello Rabbit
[x] sending 'Hello Rabbit'
[^] sent with correlation ID fdb089ff-44ea-4b81-885b-be7e0178d5a1
[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
quit
##########################################################################

So, my question is, what am I doing wrong that I didn't get the '5bec'
response to the second message?

Thanks for your help,
--
Doug Gorley | doug at gorley.ca

Search Discussions

  • Marek Majkowski at Aug 24, 2011 at 8:32 am
    Doug,

    Thanks for the report. It does look like a problem caused
    by a bug in Pika or my misunderstanding how start/stop_consuming
    work.

    I'll try to come up with a workaround.

    Cheers,
    Marek

    On Wed, Aug 24, 2011 at 02:49, Doug Gorley wrote:
    I'm hoping someone can help he with my RabbitMQ problem; I've got some
    code based on the tutorial #6 code, and it's not doing what I'm expecting.
    I tried to create a shell where I can send strings to the server and
    receive the MD5 digest, but I only ever seem to get the first reply.

    Here's the source:

    ##########################################################################
    # server.py

    import pika
    import hashlib

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

    channel = connection.channel()
    channel.queue_declare(queue='my_request')

    def on_request(ch, method, props, body):

    ? ?response = hashlib.md5(body).hexdigest()

    ? ?print("Received '%r', sending '%r'" % (body, response))

    ? ?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)

    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(on_request, queue='my_request')

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

    ##########################################################################
    # client.py

    import pika
    import uuid

    class RabbitClient(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(" [.] Received correlation ID %r" % (props.correlation_id,))
    ? ? ? ?if self.corr_id == props.correlation_id:
    ? ? ? ? ? ?self.response = body
    ? ? ? ? ? ?self.channel.stop_consuming()
    ? ? ? ?else:
    ? ? ? ? ? ?print(" [~] Correlation IDs do not match")

    ? ?def call(self, n):
    ? ? ? ?self.corr_id = str(uuid.uuid4())
    ? ? ? ?self.channel.basic_publish(exchange='',
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? routing_key='my_request',
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? properties=pika.BasicProperties(
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? reply_to=self.callback_queue,
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? correlation_id=self.corr_id,
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ),
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? body=n)
    ? ? ? ?self.channel.start_consuming()
    ? ? ? ?return self.response

    auth = RabbitClient()

    while True:
    ? ?send = raw_input('> ')
    ? ?if send == 'quit': break
    ? ?print(" [x] sending '%s'" % (send))
    ? ?response = auth.call(send)
    ? ?print(" [^] sent with correlation ID %s" % (auth.corr_id,))
    ? ?print(" [.] Got %r" % (response,))
    ##########################################################################

    Here's the output from the server:

    ##########################################################################
    ?[x] Awaiting requests
    Received ''Hello World'', sending ''b10a8db164e0754105b7a99be72e3fe5''
    Received ''Hello Rabbit'', sending ''5bec01f45b43f6202e84242fcb71b04f''
    ##########################################################################

    And here's the output from the server:
    ##########################################################################
    Hello World
    ?[x] sending 'Hello World'
    ?[.] Received correlation ID '1805f131-1fbc-40a3-86fb-2a113b5f9b9a'
    ?[^] sent with correlation ID 1805f131-1fbc-40a3-86fb-2a113b5f9b9a
    ?[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    Hello Rabbit
    ?[x] sending 'Hello Rabbit'
    ?[^] sent with correlation ID fdb089ff-44ea-4b81-885b-be7e0178d5a1
    ?[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    quit
    ##########################################################################

    So, my question is, what am I doing wrong that I didn't get the '5bec'
    response to the second message?

    Thanks for your help,
    --
    Doug Gorley | doug at gorley.ca



    _______________________________________________
    rabbitmq-discuss mailing list
    rabbitmq-discuss at lists.rabbitmq.com
    https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
  • Marek Majkowski at Aug 24, 2011 at 10:52 am

    On Wed, Aug 24, 2011 at 09:32, Marek Majkowski wrote:
    Thanks for the report. It does look like a problem caused
    by a bug in Pika or my misunderstanding how start/stop_consuming
    work.
    Yup. I did misuse "stop_consuming". Apparently, it calls
    basic.cancel, which was completely not inteded:
    https://github.com/pika/pika/blob/v0.9.5/pika/adapters/blocking_connection.py#L287

    The fix is here:
    https://github.com/rabbitmq/rabbitmq-tutorials/commit/c1b178d9deec664cdf12770c24319c7d12d67efd

    Thanks again for the report!

    Cheers,
    Marek
    On Wed, Aug 24, 2011 at 02:49, Doug Gorley wrote:
    I'm hoping someone can help he with my RabbitMQ problem; I've got some
    code based on the tutorial #6 code, and it's not doing what I'm expecting.
    I tried to create a shell where I can send strings to the server and
    receive the MD5 digest, but I only ever seem to get the first reply.

    Here's the source:

    ##########################################################################
    # server.py

    import pika
    import hashlib

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

    channel = connection.channel()
    channel.queue_declare(queue='my_request')

    def on_request(ch, method, props, body):

    ? ?response = hashlib.md5(body).hexdigest()

    ? ?print("Received '%r', sending '%r'" % (body, response))

    ? ?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)

    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(on_request, queue='my_request')

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

    ##########################################################################
    # client.py

    import pika
    import uuid

    class RabbitClient(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(" [.] Received correlation ID %r" % (props.correlation_id,))
    ? ? ? ?if self.corr_id == props.correlation_id:
    ? ? ? ? ? ?self.response = body
    ? ? ? ? ? ?self.channel.stop_consuming()
    ? ? ? ?else:
    ? ? ? ? ? ?print(" [~] Correlation IDs do not match")

    ? ?def call(self, n):
    ? ? ? ?self.corr_id = str(uuid.uuid4())
    ? ? ? ?self.channel.basic_publish(exchange='',
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? routing_key='my_request',
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? properties=pika.BasicProperties(
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? reply_to=self.callback_queue,
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? correlation_id=self.corr_id,
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ),
    ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? body=n)
    ? ? ? ?self.channel.start_consuming()
    ? ? ? ?return self.response

    auth = RabbitClient()

    while True:
    ? ?send = raw_input('> ')
    ? ?if send == 'quit': break
    ? ?print(" [x] sending '%s'" % (send))
    ? ?response = auth.call(send)
    ? ?print(" [^] sent with correlation ID %s" % (auth.corr_id,))
    ? ?print(" [.] Got %r" % (response,))
    ##########################################################################

    Here's the output from the server:

    ##########################################################################
    ?[x] Awaiting requests
    Received ''Hello World'', sending ''b10a8db164e0754105b7a99be72e3fe5''
    Received ''Hello Rabbit'', sending ''5bec01f45b43f6202e84242fcb71b04f''
    ##########################################################################

    And here's the output from the server:
    ##########################################################################
    Hello World
    ?[x] sending 'Hello World'
    ?[.] Received correlation ID '1805f131-1fbc-40a3-86fb-2a113b5f9b9a'
    ?[^] sent with correlation ID 1805f131-1fbc-40a3-86fb-2a113b5f9b9a
    ?[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    Hello Rabbit
    ?[x] sending 'Hello Rabbit'
    ?[^] sent with correlation ID fdb089ff-44ea-4b81-885b-be7e0178d5a1
    ?[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    quit
    ##########################################################################

    So, my question is, what am I doing wrong that I didn't get the '5bec'
    response to the second message?

    Thanks for your help,
    --
    Doug Gorley | doug at gorley.ca



    _______________________________________________
    rabbitmq-discuss mailing list
    rabbitmq-discuss at lists.rabbitmq.com
    https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
  • Doug Gorley at Aug 24, 2011 at 1:32 pm
    Made the change as per your update and it works great. Thanks Marek!
    --
    Doug Gorley | doug at gorley.ca
    On Wed, August 24, 2011 3:52 am, Marek Majkowski wrote:
    On Wed, Aug 24, 2011 at 09:32, Marek Majkowski wrote:
    Thanks for the report. It does look like a problem caused
    by a bug in Pika or my misunderstanding how start/stop_consuming
    work.
    Yup. I did misuse "stop_consuming". Apparently, it calls
    basic.cancel, which was completely not inteded:
    https://github.com/pika/pika/blob/v0.9.5/pika/adapters/blocking_connection.py#L287

    The fix is here:
    https://github.com/rabbitmq/rabbitmq-tutorials/commit/c1b178d9deec664cdf12770c24319c7d12d67efd

    Thanks again for the report!

    Cheers,
    Marek
    On Wed, Aug 24, 2011 at 02:49, Doug Gorley wrote:
    I'm hoping someone can help he with my RabbitMQ problem; I've got some
    code based on the tutorial #6 code, and it's not doing what I'm
    expecting.
    I tried to create a shell where I can send strings to the server and
    receive the MD5 digest, but I only ever seem to get the first reply.

    Here's the source:

    ##########################################################################
    # server.py

    import pika
    import hashlib

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

    channel = connection.channel()
    channel.queue_declare(queue='my_request')

    def on_request(ch, method, props, body):

    ?? ??response = hashlib.md5(body).hexdigest()

    ?? ??print("Received '%r', sending '%r'" % (body, response))

    ?? ??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)

    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(on_request, queue='my_request')

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

    ##########################################################################
    # client.py

    import pika
    import uuid

    class RabbitClient(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(" [.] Received correlation ID %r" %
    (props.correlation_id,))
    ?? ?? ?? ??if self.corr_id == props.correlation_id:
    ?? ?? ?? ?? ?? ??self.response = body
    ?? ?? ?? ?? ?? ??self.channel.stop_consuming()
    ?? ?? ?? ??else:
    ?? ?? ?? ?? ?? ??print(" [~] Correlation IDs do not match")

    ?? ??def call(self, n):
    ?? ?? ?? ??self.corr_id = str(uuid.uuid4())
    ?? ?? ?? ??self.channel.basic_publish(exchange='',
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ??
    routing_key='my_request',
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ??
    properties=pika.BasicProperties(
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ??
    reply_to=self.callback_queue,
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ??
    correlation_id=self.corr_id,
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ),
    ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? ?? body=n)
    ?? ?? ?? ??self.channel.start_consuming()
    ?? ?? ?? ??return self.response

    auth = RabbitClient()

    while True:
    ?? ??send = raw_input('> ')
    ?? ??if send == 'quit': break
    ?? ??print(" [x] sending '%s'" % (send))
    ?? ??response = auth.call(send)
    ?? ??print(" [^] sent with correlation ID %s" % (auth.corr_id,))
    ?? ??print(" [.] Got %r" % (response,))
    ##########################################################################

    Here's the output from the server:

    ##########################################################################
    ??[x] Awaiting requests
    Received ''Hello World'', sending ''b10a8db164e0754105b7a99be72e3fe5''
    Received ''Hello Rabbit'', sending ''5bec01f45b43f6202e84242fcb71b04f''
    ##########################################################################

    And here's the output from the server:
    ##########################################################################
    Hello World
    ??[x] sending 'Hello World'
    ??[.] Received correlation ID '1805f131-1fbc-40a3-86fb-2a113b5f9b9a'
    ??[^] sent with correlation ID 1805f131-1fbc-40a3-86fb-2a113b5f9b9a
    ??[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    Hello Rabbit
    ??[x] sending 'Hello Rabbit'
    ??[^] sent with correlation ID fdb089ff-44ea-4b81-885b-be7e0178d5a1
    ??[.] Got 'b10a8db164e0754105b7a99be72e3fe5'
    quit
    ##########################################################################

    So, my question is, what am I doing wrong that I didn't get the '5bec'
    response to the second message?

    Thanks for your help,
    --
    Doug Gorley | doug at gorley.ca



    _______________________________________________
    rabbitmq-discuss mailing list
    rabbitmq-discuss at lists.rabbitmq.com
    https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss

Related Discussions

Discussion Navigation
viewthread | post
Discussion Overview
grouprabbitmq-discuss @
categoriesrabbitmq
postedAug 24, '11 at 1:49a
activeAug 24, '11 at 1:32p
posts4
users2
websiterabbitmq.com
irc#rabbitmq

2 users in discussion

Doug Gorley: 2 posts Marek Majkowski: 2 posts

People

Translate

site design / logo © 2022 Grokbase