feat: implement WebSocket support and RabbitMQ integration for message handling

Author: npv2k1

src/modules/api/__init__.py

@@ -1,3 +1,14 @@
-"""API module initialization."""
-print("API module loaded.")
-import src.modules.api.index
+import uvicorn
+from .app import app as api_app
+import logging
+
+async def server():
+    """Main application entry point."""
+    try:
+        print("Start api...")
+        config = uvicorn.Config(api_app, host="0.0.0.0", port=8080, log_level="info")
+        server = uvicorn.Server(config)
+        await server.serve()
+    except Exception as e:
+        logging.error(f"Application failed to start: {e}", exc_info=True)
+        raise

src/modules/api/app.py

@@ -1,10 +1,26 @@
 """FastAPI application factory."""
 
-from fastapi import FastAPI
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 from fastapi.middleware.cors import CORSMiddleware
 
 from .routes import v1_router
 
+class ConnectionManager:
+    def __init__(self):
+        self.active_connections: List[WebSocket] = []
+
+    async def connect(self, websocket: WebSocket):
+        await websocket.accept()
+        self.active_connections.append(websocket)
+
+    def disconnect(self, websocket: WebSocket):
+        self.active_connections.remove(websocket)
+
+    async def broadcast(self, message: str):
+        for connection in self.active_connections:
+            await connection.send_text(message)
+
+manager = ConnectionManager()
 
 def create_app() -> FastAPI:
     """Create and configure the FastAPI application."""
@@ -31,6 +47,18 @@ async def startup_event():
     async def shutdown_event():
         """Run shutdown events."""
         pass
+    
+    @app.websocket("/ws")
+    async def websocket_endpoint(websocket: WebSocket):
+        await manager.connect(websocket)
+        try:
+            while True:
+                data = await websocket.receive_text()
+                print(f"Received: {data}")
+                await manager.broadcast(f"Message from client: {data}")
+        except WebSocketDisconnect:
+            manager.disconnect(websocket)
+            await manager.broadcast("A client disconnected")
 
     return app
 

src/modules/api/index.py

@@ -2,14 +2,15 @@
 
 from fastapi import APIRouter
 from typing import Dict, List
-
+from src.modules.transporter import publish_message
 
 # Create v1 router
 router = APIRouter(prefix='/v1', tags=['v1'])
 
 @router.get("/hello")
 async def hello_world() -> Dict[str, str]:
     """Hello world endpoint."""
+    publish_message("hello-python", "Hello from FastAPI!")
     return {"message": "Hello, World!"}
 
 @router.get("/health")

src/modules/api/routes/v1.py

@@ -0,0 +1,26 @@
+# app/utils/__init__.py
+
+import os
+import importlib
+import pkgutil
+
+# Lấy tên package hiện tại
+__all__ = []
+
+# Duyệt tất cả module con trong thư mục hiện tại (không đệ quy)
+package_dir = os.path.dirname(__file__)
+for _, module_name, is_pkg in pkgutil.iter_modules([package_dir]):
+    if not is_pkg and module_name != "__init__":
+        module = importlib.import_module(f".{module_name}", package=__name__)
+        
+        # Nếu module có biến __all__, import những gì nó khai báo
+        if hasattr(module, "__all__"):
+            for attr in module.__all__:
+                globals()[attr] = getattr(module, attr)
+                __all__.append(attr)
+        else:
+            # Import tất cả public attributes (không bắt đầu bằng _)
+            for attr in dir(module):
+                if not attr.startswith("_"):
+                    globals()[attr] = getattr(module, attr)
+                    __all__.append(attr)

src/modules/transporter/__init__.py

@@ -0,0 +1,57 @@
+import pika
+import threading
+
+credentials = pika.PlainCredentials('guest', 'guest')
+parameters = pika.ConnectionParameters(
+    host='ubuntu',
+    port=5672,  # default RabbitMQ port
+    virtual_host='/',  # default vhost, change if needed
+    credentials=credentials
+)
+
+def queue_listener(queue_name, host="localhost"):
+    def decorator(func):
+        def start_consumer():
+            connection = pika.BlockingConnection(parameters)
+            channel = connection.channel()
+            channel.queue_declare(queue=queue_name, durable=True)
+
+            def callback(ch, method, properties, body):
+                try:
+                    func(body.decode())  # Gọi hàm xử lý
+                    ch.basic_ack(delivery_tag=method.delivery_tag)
+                except Exception as e:
+                    print(f"Error handling message: {e}")
+                    ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)
+
+            channel.basic_qos(prefetch_count=1)
+            channel.basic_consume(queue=queue_name, on_message_callback=callback)
+            print(f"[+] Listening on queue '{queue_name}'")
+            channel.start_consuming()
+
+        # chạy consumer trên thread riêng (không chặn main thread)
+        threading.Thread(target=start_consumer, daemon=True).start()
+        return func
+    return decorator
+
+@queue_listener("hello-python")
+def process_message(msg):
+    print(f"[x] Received: {msg}")
+
+
+# publish a message to the queue
+def publish_message(queue_name, message):
+    connection = pika.BlockingConnection(parameters)
+    channel = connection.channel()
+    # channel.queue_declare(queue=queue_name, durable=True)
+    channel.basic_publish(
+        exchange='',
+        routing_key=queue_name,
+        body=message,
+        properties=pika.BasicProperties(
+            delivery_mode=2,  # make message persistent
+        )
+    )
+    print(f"[x] Sent: {message}")
+
+publish_message("hello-python", "RabbitMQ is running!")