Merge pull request #172 from swecc-uw/advaypatil27/eng-92-create-producermanager-singleton-in-swecc-server

Add a synchronous rabbit producer

Author: Advayp

server/members/views.py

@@ -10,6 +10,7 @@
 from django.http import JsonResponse
 from django.shortcuts import get_object_or_404
 from email_util.send_email import send_email
+from mq.producers import publish_verified_email
 from rest_framework import generics, status
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
@@ -329,4 +330,6 @@ def post(self, request, token):
                 status=status.HTTP_400_BAD_REQUEST,
             )
 
+        publish_verified_email(request.user.discord_id)
+
         return Response({"detail": "School email verified"}, status=200)

server/mq/consumers.py

@@ -50,3 +50,15 @@ async def reviewed_feedback_callback(
     logger.info(
         f"Feedback for resume {data['resume_id']} updated with value {data['feedback']}"
     )
+
+
+@mq.consumer(
+    queue="server.verified-email",
+    routing_key="server.verified-email",
+    exchange="swecc-server-exchange",
+)
+async def verified_email_callback(
+    body,
+    properties: BasicProperties,
+):
+    logger.info(f"Received verified email message: {body}")

server/mq/core/synchronous_producer.py

@@ -0,0 +1,32 @@
+import os
+
+import pika
+
+
+class SynchronousRabbitProducer:
+
+    _instance = None
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        self.host = os.getenv("RABBIT_HOST", "rabbitmq-host")
+        self._initialized = True
+        self._connection = pika.BlockingConnection(
+            pika.ConnectionParameters(host=self.host)
+        )
+        self._channel = self._connection.channel()
+
+    def __new__(cls):
+        if not cls._instance:
+            cls._instance = super(SynchronousRabbitProducer, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def publish(self, routing_key, body, exchange="swecc-server-exchange"):
+        self._channel.basic_publish(
+            exchange=exchange,
+            routing_key=routing_key,
+            body=body,
+        )

server/mq/producers.py

@@ -0,0 +1,6 @@
+from mq.core.synchronous_producer import SynchronousRabbitProducer
+
+
+def publish_verified_email(discord_id):
+    producer_manager = SynchronousRabbitProducer()
+    producer_manager.publish("server.verified-email", str(discord_id))