malmal200/ca4006-ca2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Refactored researcher.py. now recieves response

Browse Source
This commit is contained in:
Gaz 2023-04-10 20:47:18 +01:00
parent 2fa1b47edb
commit bc3f9f8dcf
2 changed files with 35 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
funding_agency.py
View File

@ -16,7 +16,7 @@ def process_proposal(ch, method, properties, body):
channel.basic_publish(exchange='', channel.basic_publish(exchange='',
routing_key=properties.reply_to, routing_key=properties.reply_to,
properties=pika.BasicProperties(correlation_id=properties.correlation_id), properties=pika.BasicProperties(correlation_id=properties.correlation_id),
body=f"{acronym} approved") body="approved")
project_info=f"{acronym},{title},{description},{amount}" project_info=f"{acronym},{title},{description},{amount}"
channel.basic_publish(exchange='', routing_key='approved_projects', body=project_info) channel.basic_publish(exchange='', routing_key='approved_projects', body=project_info)
print(f"{acronym} approved") print(f"{acronym} approved")
@ -24,7 +24,7 @@ def process_proposal(ch, method, properties, body):
channel.basic_publish(exchange='', channel.basic_publish(exchange='',
routing_key=properties.reply_to, routing_key=properties.reply_to,
properties=pika.BasicProperties(correlation_id=properties.correlation_id), properties=pika.BasicProperties(correlation_id=properties.correlation_id),
body=f"{acronym} rejected") body="rejected")
print(f"{acronym} rejected") print(f"{acronym} rejected")
channel.basic_consume(queue='research_proposals', on_message_callback=process_proposal, auto_ack=True) channel.basic_consume(queue='research_proposals', on_message_callback=process_proposal, auto_ack=True)

54
researcher.py
View File

@ -1,17 +1,37 @@
import pika import pika
import uuid import uuid
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', pika.PlainCredentials("user", "password")))
channel = connection.channel()
channel.queue_declare(queue='research_proposals')
def process_response(ch, method, properties, body): class ResearchProposal(object):
if corr_id == properties.correlation_id:
print(body) def __init__(self):
else: self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost', 5672, '/', pika.PlainCredentials("user", "password")))
print("response failed") self.channel = self.connection.channel()
result = self.channel.queue_declare(queue='', exclusive=True)
self.callback_queue = result.method.queue
self.channel.basic_consume(queue=self.callback_queue, on_message_callback=self.on_response, auto_ack=True)
self.response = None
self.corr_id = None
def on_response(self, ch, method, properties, body):
if self.corr_id == properties.correlation_id:
self.response = body
def call(self, n):
self.response = None
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(exchange='', routing_key="research_proposals",
properties=pika.BasicProperties(reply_to=self.callback_queue,
correlation_id=self.corr_id, ),
body=f"{n['acronym']},{n['title']},{n['description']},{n['amount']}")
self.connection.process_data_events(time_limit=None)
return str(self.response)
proposals = [ proposals = [
{"acronym": "PROJ1", "title": "Project 1", "description": "Description 1", "amount": "300000"}, {"acronym": "PROJ1", "title": "Project 1", "description": "Description 1", "amount": "300000"},
@ -22,17 +42,9 @@ proposals = [
{"acronym": "PROJ6", "title": "Project 6", "description": "Description 6", "amount": "420000"} {"acronym": "PROJ6", "title": "Project 6", "description": "Description 6", "amount": "420000"}
] ]
corr_id = str(uuid.uuid4())
for proposal in proposals: for proposal in proposals:
proposal_client = ResearchProposal()
response_queue = channel.queue_declare(queue='', exclusive=True).method.queue print(f"{proposal['acronym']} sends proposal")
response = proposal_client.call(proposal)
channel.basic_publish(exchange='', print(f"{proposal['acronym']} got {response}")
routing_key='research_proposals',
properties=pika.BasicProperties(reply_to=response_queue, correlation_id=corr_id),
body=f"{proposal['acronym']},{proposal['title']},{proposal['description']},{proposal['amount']}")
print(f"Proposal {proposal['acronym']} sent")
channel.basic_consume(queue=response_queue, on_message_callback=process_response)
connection.close()