From 76d1f7b1ec9a69a009777cfe686b6403cddef106 Mon Sep 17 00:00:00 2001 From: nutcas3 Date: Sat, 25 Apr 2026 06:15:11 +0300 Subject: [PATCH] feat: Add RabbitMQ message queue integration with exchange and queue management support - Add rabbitmq/amqp091-go v1.11.0 dependency to go.mod - Add MessageQueue struct with connection, channel, and url fields - Add Message struct with Type, Payload, and Timestamp fields - Add NewMessageQueue constructor with RABBITMQ_URL environment variable support and amqp://guest:guest@localhost:5672/ default - Add Close method for connection cleanup - Add DeclareQueue method with durable configuration - Add Publish method with 5 --- apps/api-server/go.mod | 1 + apps/api-server/go.sum | 3 + apps/api-server/internal/mq/mq.go | 213 ++++++++++++++++++++++++++++++ 3 files changed, 217 insertions(+) create mode 100644 apps/api-server/internal/mq/mq.go diff --git a/apps/api-server/go.mod b/apps/api-server/go.mod index 6358c92..b001fbc 100644 --- a/apps/api-server/go.mod +++ b/apps/api-server/go.mod @@ -140,6 +140,7 @@ require ( github.com/prometheus/client_golang v1.23.2 github.com/quic-go/qpack v0.6.0 // indirect github.com/quic-go/quic-go v0.59.0 // indirect + github.com/rabbitmq/amqp091-go v1.11.0 github.com/stripe/stripe-go/v82 v82.5.1 github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.3.1 // indirect diff --git a/apps/api-server/go.sum b/apps/api-server/go.sum index cb13f4a..75b780d 100644 --- a/apps/api-server/go.sum +++ b/apps/api-server/go.sum @@ -137,6 +137,7 @@ github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJn github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY= github.com/go-playground/validator/v10 v10.30.2 h1:JiFIMtSSHb2/XBUbWM4i/MpeQm9ZK2xqPNk8vgvu5JQ= github.com/go-playground/validator/v10 v10.30.2/go.mod h1:mAf2pIOVXjTEBrwUMGKkCWKKPs9NheYGabeB04txQSc= +github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo= github.com/go-sql-driver/mysql v1.9.3 h1:U/N249h2WzJ3Ukj8SowVFjdtZKfu9vlLZxjPXV1aweo= github.com/go-sql-driver/mysql v1.9.3/go.mod h1:qn46aNg1333BRMNU69Lq93t8du/dwxI64Gl8i5p1WMU= github.com/go-viper/mapstructure/v2 v2.5.0 h1:vM5IJoUAy3d7zRSVtIwQgBj7BiWtMPfmPEgAXnvj1Ro= @@ -263,6 +264,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8= github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII= github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw= github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU= +github.com/rabbitmq/amqp091-go v1.11.0 h1:HxIctVm9Gid/Vtn706necmZ7Wj6pgGI2eqplRbEY8O8= +github.com/rabbitmq/amqp091-go v1.11.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= diff --git a/apps/api-server/internal/mq/mq.go b/apps/api-server/internal/mq/mq.go new file mode 100644 index 0000000..f777b84 --- /dev/null +++ b/apps/api-server/internal/mq/mq.go @@ -0,0 +1,213 @@ +package mq + +import ( + "context" + "encoding/json" + "fmt" + "log" + "os" + "time" + + amqp "github.com/rabbitmq/amqp091-go" +) + +// MessageQueue provides a simple interface for message queue operations +type MessageQueue struct { + connection *amqp.Connection + channel *amqp.Channel + url string +} + +// Message represents a message to be sent/received +type Message struct { + Type string `json:"type"` + Payload map[string]interface{} `json:"payload"` + Timestamp time.Time `json:"timestamp"` +} + +// NewMessageQueue creates a new message queue connection +func NewMessageQueue(url string) (*MessageQueue, error) { + if url == "" { + url = os.Getenv("RABBITMQ_URL") + if url == "" { + url = "amqp://guest:guest@localhost:5672/" + } + } + + conn, err := amqp.Dial(url) + if err != nil { + return nil, fmt.Errorf("failed to connect to RabbitMQ: %w", err) + } + + ch, err := conn.Channel() + if err != nil { + conn.Close() + return nil, fmt.Errorf("failed to open channel: %w", err) + } + + mq := &MessageQueue{ + connection: conn, + channel: ch, + url: url, + } + + return mq, nil +} + +// Close closes the connection to the message queue +func (mq *MessageQueue) Close() error { + if mq.channel != nil { + mq.channel.Close() + } + if mq.connection != nil { + return mq.connection.Close() + } + return nil +} + +// DeclareQueue declares a queue +func (mq *MessageQueue) DeclareQueue(name string) error { + _, err := mq.channel.QueueDeclare( + name, + true, // durable + false, // autoDelete + false, // exclusive + false, // noWait + nil, // args + ) + if err != nil { + return fmt.Errorf("failed to declare queue %s: %w", name, err) + } + return nil +} + +// Publish publishes a message to a queue +func (mq *MessageQueue) Publish(queue string, msg Message) error { + msg.Timestamp = time.Now() + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("failed to marshal message: %w", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err = mq.channel.PublishWithContext( + ctx, + "", // exchange + queue, // routing key + false, // mandatory + false, // immediate + amqp.Publishing{ + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Body: body, + Timestamp: time.Now(), + }, + ) + if err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + + return nil +} + +// Consume starts consuming messages from a queue +func (mq *MessageQueue) Consume(queue string, handler func(Message) error) error { + msgs, err := mq.channel.Consume( + queue, + "", // consumerTag + false, // autoAck + false, // exclusive + false, // noLocal + false, // noWait + nil, // args + ) + if err != nil { + return fmt.Errorf("failed to register consumer: %w", err) + } + + for msg := range msgs { + var mqMsg Message + if err := json.Unmarshal(msg.Body, &mqMsg); err != nil { + log.Printf("Failed to unmarshal message: %v", err) + msg.Nack(false, false) + continue + } + + if err := handler(mqMsg); err != nil { + log.Printf("Handler error: %v", err) + msg.Nack(false, true) // requeue on error + continue + } + + msg.Ack(false) + } + + return nil +} + +// PublishToExchange publishes a message to an exchange +func (mq *MessageQueue) PublishToExchange(exchange, routingKey string, msg Message) error { + msg.Timestamp = time.Now() + + body, err := json.Marshal(msg) + if err != nil { + return fmt.Errorf("failed to marshal message: %w", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + err = mq.channel.PublishWithContext( + ctx, + exchange, // exchange + routingKey, // routing key + false, // mandatory + false, // immediate + amqp.Publishing{ + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Body: body, + Timestamp: time.Now(), + }, + ) + if err != nil { + return fmt.Errorf("failed to publish message: %w", err) + } + + return nil +} + +// DeclareExchange declares an exchange +func (mq *MessageQueue) DeclareExchange(name, kind string) error { + err := mq.channel.ExchangeDeclare( + name, + kind, + true, // durable + false, // autoDelete + false, // internal + false, // noWait + nil, // args + ) + if err != nil { + return fmt.Errorf("failed to declare exchange %s: %w", name, err) + } + return nil +} + +// BindQueue binds a queue to an exchange +func (mq *MessageQueue) BindQueue(queue, exchange, routingKey string) error { + err := mq.channel.QueueBind( + queue, + routingKey, + exchange, + false, // noWait + nil, // args + ) + if err != nil { + return fmt.Errorf("failed to bind queue %s to exchange %s: %w", queue, exchange, err) + } + return nil +}