Add runRequest to RabbitMQ.

Author: Abhinav-ark

controller/ea.go

@@ -123,5 +123,11 @@ func CreateEA(res http.ResponseWriter, req *http.Request) {
 
 	// TODO: Schedule the run.
 
+	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
+		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
+		return
+	}
+
 	util.JSONResponse(res, http.StatusOK, "It works! 👍🏻", data)
+
 }

controller/gp.go

@@ -122,6 +122,10 @@ func CreateGP(res http.ResponseWriter, req *http.Request) {
 	}
 
 	// TODO: Schedule the run.
+	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
+		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
+		return
+	}
 
 	util.JSONResponse(res, http.StatusOK, "It works! 👍🏻", data)
 }

controller/ml.go

@@ -122,6 +122,10 @@ func CreateML(res http.ResponseWriter, req *http.Request) {
 	}
 
 	// TODO: Schedule the run.
+	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
+		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
+		return
+	}
 
 	util.JSONResponse(res, http.StatusOK, "It works! 👍🏻", data)
 }

controller/pso.go

@@ -122,6 +122,10 @@ func CreatePSO(res http.ResponseWriter, req *http.Request) {
 	}
 
 	// TODO: Schedule the run.
+	if err := util.EnqueueRunRequest(req.Context(), runID, "code", "py"); err != nil {
+		util.JSONResponse(res, http.StatusInternalServerError, err.Error(), nil)
+		return
+	}
 
 	util.JSONResponse(res, http.StatusOK, "It works! 👍🏻", data)
 }

go.mod

@@ -22,6 +22,7 @@ require (
 	github.com/klauspost/cpuid/v2 v2.2.9 // indirect
 	github.com/minio/crc64nvme v1.0.1 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
+	github.com/rabbitmq/amqp091-go v1.10.0 // indirect
 	github.com/rs/xid v1.6.0 // indirect
 	golang.org/x/crypto v0.33.0 // indirect
 	golang.org/x/net v0.35.0 // indirect

go.sum

@@ -38,6 +38,8 @@ github.com/minio/minio-go/v7 v7.0.87 h1:nkr9x0u53PespfxfUqxP3UYWiE2a41gaofgNnC4Y
 github.com/minio/minio-go/v7 v7.0.87/go.mod h1:33+O8h0tO7pCeCWwBVa07RhVVfB/3vS4kEX7rwYKmIg=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/rabbitmq/amqp091-go v1.10.0 h1:STpn5XsHlHGcecLmMFCtg7mqq0RnD+zFr4uzukfVhBw=
+github.com/rabbitmq/amqp091-go v1.10.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
 github.com/rs/cors v1.11.1 h1:eU3gRzXLRK57F5rKMGMZURNdIG4EoAmX8k94r9wXWHA=
 github.com/rs/cors v1.11.1/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/rs/xid v1.6.0 h1:fV591PaemRlL6JfRxGDEPl69wICngIQ3shQtzfy2gxU=

util/enqueue.go

@@ -0,0 +1,93 @@
+package util
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"os"
+	"time"
+	amqp "github.com/rabbitmq/amqp091-go"
+)
+
+func EnqueueRunRequest(ctx context.Context, runID string, fileName string, extension string) error {
+
+	// Message represents the structure of our message
+	type Message struct {
+		RunId	string	`json:"runId"`
+		FileName	string	`json:"fileName"`
+		Extension	string	`json:"extension"`
+		Timestamp	time.Time	`json:"timestamp"`
+	}
+	
+	// Get RabbitMQ connection string from environment variable or use default
+	rabbitMQURL := os.Getenv("RABBITMQ_URL")
+	if rabbitMQURL == "" {
+		rabbitMQURL = "amqp://guest:guest@localhost:5672/"
+	}
+
+	// Connect to RabbitMQ server
+	conn, err := amqp.Dial(rabbitMQURL)
+	if err != nil {
+		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
+	}
+	defer conn.Close()
+
+	// Create a channel
+	ch, err := conn.Channel()
+	if err != nil {
+		log.Fatalf("Failed to open a channel: %v", err)
+	}
+	defer ch.Close()
+
+	// Declare a queue
+	queueName := "task_queue"
+	q, err := ch.QueueDeclare(
+		queueName, // name
+		true,      // durable
+		false,     // delete when unused
+		false,     // exclusive
+		false,     // no-wait
+		nil,       // arguments
+	)
+	
+	if err != nil {
+		log.Fatalf("Failed to declare a queue: %v", err)
+	}
+
+
+	// Create a new message
+	msg := Message{
+		RunId: runID,
+		FileName: fileName,
+		Extension: extension,
+		Timestamp: time.Now(),
+	}
+
+	// Convert message to JSON
+	body, err := json.Marshal(msg)
+	if err != nil {
+		log.Printf("Error marshaling message: %v", err)
+	}
+
+	// Publish message
+	err = ch.PublishWithContext(
+		ctx,
+		"",     // exchange
+		q.Name, // routing key
+		false,  // mandatory
+		false,  // immediate
+		amqp.Publishing{
+			DeliveryMode: amqp.Persistent, 
+			Body: body,
+		},
+	)
+			
+	if err != nil {
+		log.Printf("Failed to publish a message: %v", err)
+	}
+	
+	log.Printf("Published message: %s", msg.RunId)
+		
+
+	return nil
+}

util/minio.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
 )