From ed42bf7eb7f2461ca75fec84104c61e867fc907c Mon Sep 17 00:00:00 2001
From: James Kirk <james.kirk@noc.ac.uk>
Date: Mon, 23 Jan 2023 16:40:36 +0000
Subject: [PATCH] refactor: changed endpoints to use rmq funcs refactor:
deleted hello.py fix: minor comment typo
---
endpoints/hello.py | 6 ------
endpoints/notify.py | 12 +++++-------
endpoints/receive.py | 30 +++++-------------------------
endpoints/send.py | 15 +++++++--------
rmq.py | 2 +-
5 files changed, 18 insertions(+), 47 deletions(-)
delete mode 100644 endpoints/hello.py
diff --git a/endpoints/hello.py b/endpoints/hello.py
deleted file mode 100644
index 7bb0bab..0000000
--- a/endpoints/hello.py
+++ /dev/null
@@ -1,6 +0,0 @@
-from flask_restful import Resource
-
-
-class HelloWorld(Resource):
- def get(self):
- return {"hello": "world"}
diff --git a/endpoints/notify.py b/endpoints/notify.py
index a953db8..22bcb8d 100644
--- a/endpoints/notify.py
+++ b/endpoints/notify.py
@@ -1,8 +1,10 @@
import json
-from flask_restful import request, abort
+
+from flask_restful import abort, request
from marshmallow import Schema, fields
-import pika
+
from endpoints.auth_resource import AuthResource
+from rmq import write_to_queue
class NotifySchema(Schema):
@@ -33,8 +35,4 @@ class Notify(AuthResource):
if allow:
notify_queue = self.client['client_id'] + "-broadcast"
- connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
- channel = connection.channel()
- channel.queue_declare(queue=notify_queue, durable=True)
- channel.basic_publish(exchange="", routing_key=notify_queue, body=json.dumps(message))
- connection.close()
+ write_to_queue(queue_name=notify_queue, msg=json.dumps(message))
\ No newline at end of file
diff --git a/endpoints/receive.py b/endpoints/receive.py
index bc1feba..5c1f99c 100644
--- a/endpoints/receive.py
+++ b/endpoints/receive.py
@@ -1,9 +1,8 @@
-from flask_restful import request, abort
+from flask_restful import abort, request
from marshmallow import Schema, fields
-import pika
-import json
-from models.token import TokenModel
+
from endpoints.auth_resource import AuthResource
+from rmq import read_from_queue
class ReceiveQuerySchema(Schema):
@@ -23,28 +22,9 @@ class Receive(AuthResource):
if errors:
abort(400, message=str(errors))
- messages = []
max_messages = request.args.get("max_messages", 10)
-
+
allow = self.auth(request)
if allow:
inbox_queue = self.client['client_id'] + "-inbox"
-
- if allow:
- connection = pika.BlockingConnection(
- pika.ConnectionParameters(host="localhost")
- )
- channel = connection.channel()
- channel.queue_declare(queue=inbox_queue, durable=True)
- while len(messages) < max_messages:
- method_frame, header_frame, body = channel.basic_get(inbox_queue)
- if method_frame:
- print(method_frame, header_frame, body)
- channel.basic_ack(method_frame.delivery_tag)
- messages.append(json.loads(body.decode()))
- else:
- print("No message returned")
- break
- channel.close()
- connection.close()
- return messages
+ return read_from_queue(queue_name=inbox_queue, max_msgs=max_messages)
\ No newline at end of file
diff --git a/endpoints/send.py b/endpoints/send.py
index 900e8a5..9fe91ca 100644
--- a/endpoints/send.py
+++ b/endpoints/send.py
@@ -1,8 +1,11 @@
import json
-from flask_restful import request, abort
+
+from flask_restful import abort, request
from marshmallow import Schema, fields
-import pika
+
from endpoints.auth_resource import AuthResource
+from rmq import write_to_queue
+
class SendSchema(Schema):
body = fields.Str(required=True)
@@ -28,13 +31,9 @@ class Send(AuthResource):
body = args.get("body")
topic = args.get("topic")
outbox_queue = self.client['client_id'] + "-outbox"
-
- connection = pika.BlockingConnection(pika.ConnectionParameters(host="localhost"))
- channel = connection.channel()
- channel.queue_declare(queue=outbox_queue, durable=True)
message = {
'topic': topic,
'message': body,
}
- channel.basic_publish(exchange="", routing_key=outbox_queue, body=json.dumps(message))
- connection.close()
+
+ write_to_queue(queue_name=outbox_queue, msg=json.dumps(message))
\ No newline at end of file
diff --git a/rmq.py b/rmq.py
index 3008d0f..8216ab5 100644
--- a/rmq.py
+++ b/rmq.py
@@ -53,7 +53,7 @@ def write_to_queue(queue_name, msg):
def read_from_queue(queue_name, max_msgs):
- # get messages off of a queue until the queue is empty of max_msgs is hit
+ # get messages off of a queue until the queue is empty or max_msgs is hit
connection, channel = pika_connect(host=host)
setup_queue(channel=channel, queue_name=queue_name)
--
GitLab