This commit is contained in:
commit
f752668cb5
@ -28,6 +28,7 @@ def message_callback(channel, method, properties, body):
|
|||||||
routing_key='classification-result',
|
routing_key='classification-result',
|
||||||
body=json.dumps(results).encode("utf-8")
|
body=json.dumps(results).encode("utf-8")
|
||||||
)
|
)
|
||||||
|
channel.basic_ack(delivery_tag=method.delivery_tag)
|
||||||
|
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
@ -56,14 +57,16 @@ def main():
|
|||||||
logging.info("Connecting to MQ service...")
|
logging.info("Connecting to MQ service...")
|
||||||
connection = pika.BlockingConnection(pika.connection.URLParameters(os.environ['PIKA_URL']))
|
connection = pika.BlockingConnection(pika.connection.URLParameters(os.environ['PIKA_URL']))
|
||||||
channel = connection.channel()
|
channel = connection.channel()
|
||||||
channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='fanout')
|
channel.exchange_declare(exchange=os.environ['PIKA_INPUT_EXCHANGE'], exchange_type='direct')
|
||||||
|
|
||||||
queue_declare_result = channel.queue_declare(queue='', exclusive=True)
|
queue_declare_result = channel.queue_declare(queue='cnnqueue', exclusive=False)
|
||||||
queue_name = queue_declare_result.method.queue
|
queue_name = queue_declare_result.method.queue
|
||||||
|
|
||||||
channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'], queue=queue_name)
|
channel.queue_bind(exchange=os.environ['PIKA_INPUT_EXCHANGE'], routing_key='feature', queue=queue_name)
|
||||||
|
|
||||||
channel.basic_consume(queue=queue_name, on_message_callback=message_callback, auto_ack=True)
|
channel.basic_qos(prefetch_count=1)
|
||||||
|
|
||||||
|
channel.basic_consume(queue=queue_name, on_message_callback=message_callback, auto_ack=False)
|
||||||
|
|
||||||
logging.info("Connection complete! Listening to messages...")
|
logging.info("Connection complete! Listening to messages...")
|
||||||
try:
|
try:
|
||||||
|
Loading…
Reference in New Issue
Block a user