fix: update docker-compose

Author: tituschewxj

apps/docker-compose.yml

@@ -68,6 +68,8 @@ services:
       - apps_network
     volumes:
       - ./history-service:/history-service
+    depends_on:
+      - rabbitmq
 
   signalling-service:
     build:
@@ -95,6 +97,8 @@ services:
     volumes:
       - ./execution-service:/execution-service
       - /var/run/docker.sock:/var/run/docker.sock
+    depends_on:
+      - rabbitmq
 
   redis:
     image: redis:latest
@@ -104,7 +108,7 @@ services:
       - 6379:6379
     container_name: redis-container
 
-  rabbit-mq:
+  rabbitmq:
     image: rabbitmq:3-management
     networks:
       - apps_network

apps/execution-service/.env.example

@@ -3,6 +3,8 @@ PORT=8083
 
 # If you are NOT USING docker, use the below variables
 # HISTORY_SERVICE_URL=http://localhost:8082/
+# RABBITMQ_URL=amqp://guest:guest@localhost:5672/
 
 # If you are USING docker, use the below variables
 HISTORY_SERVICE_URL=http://history-service:8082/
+RABBITMQ_URL=amqp://guest:guest@rabbitmq:5672/

apps/execution-service/main.go

@@ -25,9 +25,6 @@ func main() {
 	err := godotenv.Load()
 	utils.FailOnError(err, "Error loading .env file")
 
-	amqpChannel := messagequeue.InitRabbitMQServer()
-	defer amqpChannel.Close()
-
 	// Initialize Firestore client
 	ctx := context.Background()
 	client, err := initFirestore(ctx)
@@ -36,6 +33,10 @@ func main() {
 
 	service := &handlers.Service{Client: client}
 
+	amqpConnection, amqpChannel := messagequeue.InitRabbitMQServer()
+	defer amqpConnection.Close()
+	defer amqpChannel.Close()
+
 	r := initChiRouter(service)
 	initRestServer(r)
 }

apps/execution-service/messagequeue/rabbitmq.go

@@ -5,23 +5,23 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"time"
 
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
-const CODE_SUBMISSION_QUEUE_KEY = "code-submission"
+const (
+	CODE_SUBMISSION_QUEUE_KEY = "code-submission"
+	NUM_RETRIES               = 10
+)
 
 var (
 	codeSubmissionQueue amqp.Queue
 	rabbitMQChannel     *amqp.Channel
 )
 
-func InitRabbitMQServer() *amqp.Channel {
-	// Connect to RabbitMQ server
-	rabbitMQURL := os.Getenv("RABBITMQ_URL")
-	conn, err := amqp.Dial(rabbitMQURL)
-	utils.FailOnError(err, "Failed to connect to RabbitMQ")
-	defer conn.Close()
+func InitRabbitMQServer() (*amqp.Connection, *amqp.Channel) {
+	conn := connectToRabbitMQ()
 
 	// Create a channel
 	ch, err := conn.Channel()
@@ -40,7 +40,24 @@ func InitRabbitMQServer() *amqp.Channel {
 	utils.FailOnError(err, "Failed to declare a queue")
 	codeSubmissionQueue = q
 
-	return ch
+	return conn, ch
+}
+
+func connectToRabbitMQ() *amqp.Connection {
+	var conn *amqp.Connection
+	var err error
+	rabbitMQURL := os.Getenv("RABBITMQ_URL")
+	for i := 0; i < NUM_RETRIES; i++ { // Retry up to 10 times
+		conn, err = amqp.Dial(rabbitMQURL)
+		if err == nil {
+			log.Println("Connected to RabbitMQ")
+			return conn
+		}
+		log.Printf("Failed to connect to RabbitMQ, retrying in 5 seconds... (Attempt %d/%d)", i+1, NUM_RETRIES)
+		time.Sleep(5 * time.Second)
+	}
+	utils.FailOnError(err, fmt.Sprintf("Failed to connect to RabbitMQ after %d attempts", NUM_RETRIES))
+	return nil
 }
 
 func PublishSubmissionMessage(submission []byte) error {

apps/history-service/.env.example

@@ -1,2 +1,8 @@
 FIREBASE_CREDENTIAL_PATH=cs3219-staging-codehisto-bb61c-firebase-adminsdk-egopb-95cfaf9b87.json
-PORT=8082
+PORT=8082
+
+# If you are NOT USING docker, use the below variables
+# RABBITMQ_URL=amqp://guest:guest@localhost:5672/
+
+# If you are USING docker, use the below variables
+RABBITMQ_URL=amqp://guest:guest@rabbitmq:5672/

apps/history-service/main.go

@@ -33,9 +33,10 @@ func main() {
 
 	service := &handlers.Service{Client: client}
 
-	amqpChannel := messagequeue.InitRabbitMQServer()
+	amqpConnection, amqpChannel := messagequeue.InitRabbitMQServer()
+	defer amqpConnection.Close()
 	defer amqpChannel.Close()
-	messagequeue.ConsumeSubmissionMessages(client, databases.CreateHistory)
+	go messagequeue.ConsumeSubmissionMessages(client, databases.CreateHistory)
 
 	r := initChiRouter(service)
 	initRestServer(r)

apps/history-service/messagequeue/rabbitmq.go

@@ -3,34 +3,34 @@ package messagequeue
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"history-service/models"
 	"history-service/utils"
 	"log"
 	"os"
+	"time"
 
 	"cloud.google.com/go/firestore"
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
-const CODE_SUBMISSION_QUEUE_KEY = "code-submission"
+const (
+	CODE_SUBMISSION_QUEUE_KEY = "code-submission"
+	NUM_RETRIES               = 10
+)
 
 var (
 	codeSubmissionQueue amqp.Queue
 	rabbitMQChannel     *amqp.Channel
 )
 
-func InitRabbitMQServer() *amqp.Channel {
-	// Connect to RabbitMQ server
-	rabbitMQURL := os.Getenv("RABBITMQ_URL")
-	conn, err := amqp.Dial(rabbitMQURL)
-	utils.FailOnError(err, "Failed to connect to RabbitMQ")
-	defer conn.Close()
+func InitRabbitMQServer() (*amqp.Connection, *amqp.Channel) {
+	conn := connectToRabbitMQ()
 
 	// Create a channel
 	ch, err := conn.Channel()
 	utils.FailOnError(err, "Failed to open a channel")
 	rabbitMQChannel = ch
-	defer ch.Close()
 
 	// Declare a queue
 	q, err := ch.QueueDeclare(
@@ -44,7 +44,24 @@ func InitRabbitMQServer() *amqp.Channel {
 	utils.FailOnError(err, "Failed to declare a queue")
 	codeSubmissionQueue = q
 
-	return ch
+	return conn, ch
+}
+
+func connectToRabbitMQ() *amqp.Connection {
+	var conn *amqp.Connection
+	var err error
+	rabbitMQURL := os.Getenv("RABBITMQ_URL")
+	for i := 0; i < NUM_RETRIES; i++ { // Retry up to 10 times
+		conn, err = amqp.Dial(rabbitMQURL)
+		if err == nil {
+			log.Println("Connected to RabbitMQ")
+			return conn
+		}
+		log.Printf("Failed to connect to RabbitMQ, retrying in 5 seconds... (Attempt %d/%d)", i+1, NUM_RETRIES)
+		time.Sleep(5 * time.Second)
+	}
+	utils.FailOnError(err, fmt.Sprintf("Failed to connect to RabbitMQ after %d attempts", NUM_RETRIES))
+	return nil
 }
 
 func ConsumeSubmissionMessages(client *firestore.Client, createSubmission func(
@@ -62,7 +79,7 @@ func ConsumeSubmissionMessages(client *firestore.Client, createSubmission func(
 		false,                    // no-wait
 		nil,                      // args
 	)
-	utils.FailOnError(err, "Failed to register a consumer")
+	utils.FailOnError(err, "RabbitMQ: Failed to register a consumer")
 
 	// Create a channel to block indefinitely
 	forever := make(chan bool)