feat: Add RabbitMQ message queue integration with profile event publishing

- Add mq package with MessageQueue struct containing connection, channel, and url fields
- Add Message struct with Type, Payload, and Timestamp fields
- Add NewMessageQueue constructor with RABBITMQ_URL environment variable and amqp://guest:guest@localhost:5672/ default
- Add Close method to cleanup channel and connection resources
- Add DeclareQueue method with durable queue configuration
- Add Publish method with 5-second context
This commit is contained in:
nutcas3
2026-04-25 06:49:21 +03:00
parent 1fc84859df
commit 0ea496b627
+133
View File
@@ -0,0 +1,133 @@
package mq
import (
"context"
"encoding/json"
"fmt"
"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]any `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
}
// PublishProfileEvent publishes a profile-related event
func (mq *MessageQueue) PublishProfileEvent(eventType string, profileID string, payload map[string]any) error {
if mq == nil {
return nil // Skip if message queue not initialized
}
msg := Message{
Type: eventType,
Payload: map[string]any{
"profile_id": profileID,
},
}
for k, v := range payload {
msg.Payload[k] = v
}
return mq.Publish("profile-events", msg)
}