diff --git a/README.MD b/README.MD index e663696..7ff57b8 100644 --- a/README.MD +++ b/README.MD @@ -72,7 +72,7 @@ Notifications should be published to the RabbitMQ exchange with the following JS } ``` -Python example +Python example rabbitmq broker ```python from kombu import Connection, Exchange, Producer @@ -93,6 +93,28 @@ producer.publish(message) print("Message sent to all workers!") ``` +Python example redis broker +```python +import redis +import json + +# Redis ulanishi +r = redis.StrictRedis(host='127.0.0.1', port=6379, db=0) + +# Xabar tayyorlash +message = { + 'type': 'email', + 'message': "Subject: test\r\n\r\nclasscom.uz sayti va mobil ilovasiga ro'yxatdan o'tishingiz uchun tasdiqlash kodi: 1234", + 'to': ["JscorpTech@gmail.com", "admin@jscorp.uz"] +} + +# Xabarni JSON formatga o‘tkazib, Redis listga push qilish +r.rpush('notification', json.dumps(message)) + +print("Message pushed to Redis list!") + +``` + Available notification types: - `email`: For email notifications - `sms`: For SMS notifications diff --git a/internal/broker/rabbitmq.go b/internal/broker/rabbitmq.go new file mode 100644 index 0000000..fea3564 --- /dev/null +++ b/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, ¬ification); err != nil { + fmt.Print(err.Error()) + } + go handler(notification) + } + }() +} diff --git a/internal/broker/redis.go b/internal/broker/redis.go new file mode 100644 index 0000000..8d1aee6 --- /dev/null +++ b/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]), ¬ification); err != nil { + fmt.Print(err.Error()) + return + } + go handler(notification) + } + }() + +} diff --git a/internal/consumer/notification.go b/internal/consumer/notification.go index 7543116..488e17b 100644 --- a/internal/consumer/notification.go +++ b/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, ¬ification) - if err != nil { - fmt.Print(err.Error()) - } +func (n *notificationConsumer) Handler(notification domain.NotificationMsg) { var ntf domain.NotifierPort switch notification.Type { case "sms": diff --git a/internal/domain/broker.go b/internal/domain/broker.go new file mode 100644 index 0000000..29ebb5a --- /dev/null +++ b/internal/domain/broker.go @@ -0,0 +1,6 @@ +package domain + +type BrokerPort interface { + Subscribe(string, func(NotificationMsg)) + // Publish() +} diff --git a/internal/domain/notification.go b/internal/domain/notification.go index 8442b1d..2f23dc6 100644 --- a/internal/domain/notification.go +++ b/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 { diff --git a/test.py b/test.py index 5062681..ab43dbf 100644 --- a/test.py +++ b/test.py @@ -1,17 +1,36 @@ -from kombu import Connection, Exchange, Producer +# from kombu import Connection, Exchange, Producer -# RabbitMQ ulanishi -rabbit_url = 'amqp://guest:guest@127.0.0.1:5672/' -connection = Connection(rabbit_url) -channel = connection.channel() +# # RabbitMQ ulanishi +# rabbit_url = 'amqp://guest:guest@127.0.0.1:5672/' +# connection = Connection(rabbit_url) +# channel = connection.channel() -exchange = Exchange('notification', type='direct') +# exchange = Exchange('notification', type='direct') -# Producer yaratish -producer = Producer(channel, exchange=exchange, routing_key="notification") +# # Producer yaratish +# producer = Producer(channel, exchange=exchange, routing_key="notification") -# Xabar yuborish -message = {'type': 'email', 'message': "Subject: test\r\n\r\nclasscom.uz sayti va mobil ilovasiga ro'yxatdan o'tishingingiz uchun tasdiqlash kodi: 1234", "to": ["JscorpTech@gmail.com", "admin@jscorp.uz"]} -producer.publish(message) +# # Xabar yuborish +# message = {'type': 'email', 'message': "Subject: test\r\n\r\nclasscom.uz sayti va mobil ilovasiga ro'yxatdan o'tishingingiz uchun tasdiqlash kodi: 1234", "to": ["JscorpTech@gmail.com", "admin@jscorp.uz"]} +# producer.publish(message) -print("Message sent to all workers!") \ No newline at end of file +# print("Message sent to all workers!") + + +import redis +import json + +# Redis ulanishi +r = redis.StrictRedis(host='127.0.0.1', port=6379, db=0) + +# Xabar tayyorlash +message = { + 'type': 'email', + 'message': "Subject: test\r\n\r\nclasscom.uz sayti va mobil ilovasiga ro'yxatdan o'tishingiz uchun tasdiqlash kodi: 1234", + 'to': ["JscorpTech@gmail.com", "admin@jscorp.uz"] +} + +# Xabarni JSON formatga o‘tkazib, Redis listga push qilish +r.rpush('notification', json.dumps(message)) + +print("Message pushed to Redis list!")