release: v0.2.1 (#10)

* fix: Use an exchange instead of a queue to handle real time updates (#9)

* refactor(domain): Update method in the real time queue interface

* chore: Update logging messages when connecting to PG database

* fix: Use an exchange instead of a queue to send submission status updates

Author: PedroChaparro

CHANGELOG.md

@@ -1,3 +1,12 @@
+## [0.2.1](https://github.com/upb-code-labs/submissions-status-updater-microservice/compare/v0.2.0...v0.2.1) (2024-01-06)
+
+
+### Bug Fixes
+
+* Use an exchange instead of a queue to handle real time updates ([#9](https://github.com/upb-code-labs/submissions-status-updater-microservice/issues/9)) ([3334eb4](https://github.com/upb-code-labs/submissions-status-updater-microservice/commit/3334eb4d88d04dfa84315a69ac6681f147000399))
+
+
+
 # [0.2.0](https://github.com/upb-code-labs/submissions-status-updater-microservice/compare/v0.1.0...v0.2.0) (2024-01-06)
 
 

src/application/submissions-status-updater-use-cases.go

@@ -21,7 +21,7 @@ func (useCases *SubmissionsStatusUpdaterUseCases) UpdateSubmissionStatus(dto *dt
 	}
 
 	// Send submission status update to the submissions real time updates queue
-	err = useCases.SubmissionsRealTimeUpdatesQueue.EnqueueUpdate(dto)
+	err = useCases.SubmissionsRealTimeUpdatesQueue.SendUpdate(dto)
 	if err != nil {
 		log.Println("[SubmissionsStatusUpdaterUseCases] Error enqueuing submission status update: " + err.Error())
 		return err

src/config/connections/postgres-connection.go

@@ -21,11 +21,11 @@ func GetPostgresConnection() *sql.DB {
 
 		// Check connection
 		if err = db.Ping(); err != nil {
-			log.Fatal(err.Error())
+			log.Fatal("[Postgres]: Error connecting to the database: " + err.Error())
 		}
 
 		// Set connection
-		log.Println("Connected to Postgres")
+		log.Println("[Postgres]: Connected")
 		pgConnection = db
 	}
 

src/domain/definitions/submissions-real-time-updates-queue.go

@@ -4,5 +4,5 @@ import "github.com/upb-code-labs/submissions-status-updater-microservice/src/dom
 
 type SubmissionsRealTimeUpdatesQueue interface {
 	// Publishes a message to the queue
-	EnqueueUpdate(updateDTO *dtos.SubmissionStatusUpdateDTO) error
+	SendUpdate(updateDTO *dtos.SubmissionStatusUpdateDTO) error
 }

src/infrastructure/implementations/rubmissions-real-time-updates-queue-implementation.go

@@ -0,0 +1,113 @@
+package implementations
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"time"
+
+	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/upb-code-labs/submissions-status-updater-microservice/src/config/connections"
+	"github.com/upb-code-labs/submissions-status-updater-microservice/src/domain/dtos"
+)
+
+var exchangeName = "submissions-real-time-updates"
+
+type SubmissionsRealTimeUpdatesRabbitmqMgr struct{}
+
+// ## Singleton instance ##
+
+// submissionsRealTimeUpdatesRabbitmqMgr Singleton instance
+var submissionsRealTimeUpdatesRabbitmqMgr *SubmissionsRealTimeUpdatesRabbitmqMgr
+
+// GetSubmissionsRealTimeUpdatesQueueMgrInstance returns the singleton instance of the submissionsRealTimeUpdatesQueueMgr
+func GetSubmissionsRealTimeUpdatesQueueMgrInstance() *SubmissionsRealTimeUpdatesRabbitmqMgr {
+	if submissionsRealTimeUpdatesRabbitmqMgr == nil {
+		// Declare exchange
+		declareRealTimeUpdatesExchange()
+
+		// Declare singleton instance
+		submissionsRealTimeUpdatesRabbitmqMgr = &SubmissionsRealTimeUpdatesRabbitmqMgr{}
+	}
+
+	return submissionsRealTimeUpdatesRabbitmqMgr
+}
+
+func declareRealTimeUpdatesExchange() {
+	ch := connections.GetRabbitMQChannel()
+
+	exchangeName := exchangeName
+	exchangeType := "fanout"
+	exchangeDurable := true
+	exchangeAutoDelete := false
+	exchangeInternal := false
+	exchangeNoWait := false
+	exchangeArgs := amqp.Table{}
+
+	err := ch.ExchangeDeclare(
+		exchangeName,
+		exchangeType,
+		exchangeDurable,
+		exchangeAutoDelete,
+		exchangeInternal,
+		exchangeNoWait,
+		exchangeArgs,
+	)
+
+	if err != nil {
+		log.Fatal(
+			"[RabbitMQ]: Error declaring real time updates exchange",
+			err.Error(),
+		)
+	}
+
+	log.Println("[RabbitMQ]: Real time updates exchange declared")
+}
+
+// ## Methods implementation ##
+
+// SendUpdate enqueues a submission status update
+func (queueMgr *SubmissionsRealTimeUpdatesRabbitmqMgr) SendUpdate(updateDTO *dtos.SubmissionStatusUpdateDTO) error {
+	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+	defer cancel()
+
+	ch := connections.GetRabbitMQChannel()
+
+	// Parse update to JSON
+	json, err := json.Marshal(updateDTO)
+	if err != nil {
+		log.Println(
+			"[RabbitMQ Submissions Real Time Updates]: Error parsing submission status update to JSON",
+			err.Error(),
+		)
+		return err
+	}
+
+	// Publish
+	msgExchange := exchangeName
+	msgRoutingKey := ""
+	msgMandatory := false
+	msgImmediate := false
+
+	err = ch.PublishWithContext(
+		ctx,
+		msgExchange,
+		msgRoutingKey,
+		msgMandatory,
+		msgImmediate,
+		amqp.Publishing{
+			ContentType: "application/json",
+			Body:        json,
+		},
+	)
+
+	if err != nil {
+		log.Println(
+			"[RabbitMQ Submissions Real Time Updates]: Error publishing submission status update",
+			err.Error(),
+		)
+		return err
+	}
+
+	return nil
+}

src/infrastructure/implementations/rubmissions-real-time-updates-rabbitmq-implementation.go

@@ -124,7 +124,7 @@ func (queueMgr *SubmissionStatusUpdatesQueueMgr) ListenForSubmissionStatusUpdate
 // processSubmissionStatusUpdates creates an infinite loop that processes submission status updates
 func (queueMgr *SubmissionStatusUpdatesQueueMgr) processSubmissionStatusUpdates() {
 	log.Println(
-		"[RabbitMQ Submissions Status Updates Queue]: Listening for submission status updates...",
+		"[RabbitMQ Submissions Status Updates Queue]: Listening for updates...",
 	)
 
 	// Process each message in a separate goroutine

src/infrastructure/submission-status-updates-queue.go

@@ -1,3 +1,3 @@
 {
-  "version": "0.2.0"
+  "version": "0.2.1"
 }