message broker sifatida redis qo'shildi

internal/broker/rabbitmq.go

@@ -0,0 +1,46 @@
+package broker
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+
+	"github.com/JscorpTech/notification/internal/domain"
+	"github.com/JscorpTech/notification/internal/rabbitmq"
+)
+
+type rabbitMQBroker struct {
+	Ctx context.Context
+}
+
+func NewRabbitMQBroker(ctx context.Context) domain.BrokerPort {
+	return &rabbitMQBroker{
+		Ctx: ctx,
+	}
+}
+
+func (r rabbitMQBroker) Subscribe(topic string, handler func(domain.NotificationMsg)) {
+	conn, ch, err := rabbitmq.Connect()
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer conn.Close()
+	defer ch.Close()
+
+	ch.ExchangeDeclare(topic, "direct", true, false, false, false, nil)
+	q, _ := ch.QueueDeclare(topic, true, false, false, false, nil)
+	ch.QueueBind(q.Name, topic, topic, false, nil)
+
+	msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)
+
+	go func() {
+		for msg := range msgs {
+			var notification domain.NotificationMsg
+			if err := json.Unmarshal(msg.Body, &notification); err != nil {
+				fmt.Print(err.Error())
+			}
+			go handler(notification)
+		}
+	}()
+}

internal/broker/redis.go

@@ -0,0 +1,39 @@
+package broker
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/JscorpTech/notification/internal/domain"
+	"github.com/JscorpTech/notification/internal/redis"
+)
+
+type redisBroker struct {
+	Ctx context.Context
+}
+
+func NewRedisBroker(ctx context.Context) domain.BrokerPort {
+	return &redisBroker{
+		Ctx: ctx,
+	}
+}
+
+func (r redisBroker) Subscribe(topic string, handler func(domain.NotificationMsg)) {
+	go func() {
+		for {
+			var notification domain.NotificationMsg
+			val, err := redis.RDB.BLPop(r.Ctx, 0, topic).Result()
+			if err != nil {
+				fmt.Print(err.Error())
+				return
+			}
+			if err := json.Unmarshal([]byte(val[1]), &notification); err != nil {
+				fmt.Print(err.Error())
+				return
+			}
+			go handler(notification)
+		}
+	}()
+
+}

internal/consumer/notification.go

@@ -2,14 +2,12 @@ package consumer
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"log"
+	"os"
 
+	"github.com/JscorpTech/notification/internal/broker"
 	"github.com/JscorpTech/notification/internal/domain"
 	"github.com/JscorpTech/notification/internal/notifier"
-	"github.com/JscorpTech/notification/internal/rabbitmq"
-	"github.com/streadway/amqp"
 )
 
 type notificationConsumer struct {
@@ -23,39 +21,26 @@ func NewNotificationConsumer(ctx context.Context) domain.NotificationConsumerPor
 }
 
 func (n *notificationConsumer) Start() {
-	conn, ch, err := rabbitmq.Connect()
-	if err != nil {
-		log.Fatal(err)
+
+	brokerName := os.Getenv("BROKER")
+	if brokerName == "" {
+		brokerName = "redis"
 	}
-	defer conn.Close()
-	defer ch.Close()
-
-	exchangeName := "notification"
-	queueName := "notification"
-	routingKey := "notification"
-
-	ch.ExchangeDeclare(exchangeName, "direct", true, false, false, false, nil)
-	q, _ := ch.QueueDeclare(queueName, true, false, false, false, nil)
-	ch.QueueBind(q.Name, routingKey, exchangeName, false, nil)
-
-	msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)
-
-	go func() {
-		for msg := range msgs {
-			go n.Handler(msg)
-		}
-	}()
-
+	var brokerService domain.BrokerPort
+	switch brokerName {
+	case "redis":
+		brokerService = broker.NewRedisBroker(n.Ctx)
+	case "rabbitmq":
+		brokerService = broker.NewRabbitMQBroker(n.Ctx)
+	default:
+		brokerService = broker.NewRedisBroker(n.Ctx)
+	}
+	brokerService.Subscribe(os.Getenv("TOPIC"), n.Handler)
 	fmt.Println("🚀 Server started. Ctrl+C to quit.")
 	select {}
 }
 
-func (n *notificationConsumer) Handler(msg amqp.Delivery) {
-	var notification domain.NotificationMsg
-	err := json.Unmarshal(msg.Body, &notification)
-	if err != nil {
-		fmt.Print(err.Error())
-	}
+func (n *notificationConsumer) Handler(notification domain.NotificationMsg) {
 	var ntf domain.NotifierPort
 	switch notification.Type {
 	case "sms":

internal/domain/broker.go

@@ -0,0 +1,6 @@
+package domain
+
+type BrokerPort interface {
+	Subscribe(string, func(NotificationMsg))
+	// Publish()
+}

internal/domain/notification.go

@@ -1,10 +1,8 @@
 package domain
 
-import "github.com/streadway/amqp"
-
 type NotificationConsumerPort interface {
 	Start()
-	Handler(amqp.Delivery)
+	Handler(NotificationMsg)
 }
 
 type SMSServicePort interface {