107 lines
2.0 KiB
Go
107 lines
2.0 KiB
Go
package rabbitmq
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"feedsystem_video_go/internal/config"
|
|
"strconv"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
type RabbitMQ struct {
|
|
conn *amqp.Connection
|
|
ch *amqp.Channel
|
|
}
|
|
|
|
func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) {
|
|
if cfg == nil {
|
|
return nil, errors.New("rabbitmq config is nil")
|
|
}
|
|
url := "amqp://" + cfg.Username + ":" + cfg.Password + "@" + cfg.Host + ":" + strconv.Itoa(cfg.Port) + "/"
|
|
conn, err := amqp.Dial(url)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ch, err := conn.Channel()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &RabbitMQ{conn: conn, ch: ch}, nil
|
|
}
|
|
|
|
func (r *RabbitMQ) Close() error {
|
|
if r == nil || r.ch == nil || r.conn == nil {
|
|
return nil
|
|
}
|
|
if err := r.ch.Close(); err != nil {
|
|
return err
|
|
}
|
|
if err := r.conn.Close(); err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error {
|
|
if r == nil || r.ch == nil {
|
|
return errors.New("rabbitmq is not initialized")
|
|
}
|
|
if exchange == "" || queue == "" || bindingKey == "" {
|
|
return errors.New("exchange/queue/bindingKey is required")
|
|
}
|
|
|
|
if err := r.ch.ExchangeDeclare(
|
|
exchange,
|
|
"topic",
|
|
true,
|
|
false,
|
|
false,
|
|
false,
|
|
nil,
|
|
); err != nil {
|
|
return err
|
|
}
|
|
|
|
q, err := r.ch.QueueDeclare(
|
|
queue,
|
|
true,
|
|
false,
|
|
false,
|
|
false,
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return r.ch.QueueBind(
|
|
q.Name,
|
|
bindingKey,
|
|
exchange,
|
|
false,
|
|
nil,
|
|
)
|
|
}
|
|
|
|
func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error {
|
|
if r == nil || r.ch == nil {
|
|
return errors.New("rabbitmq is not initialized")
|
|
}
|
|
if exchange == "" || routingKey == "" {
|
|
return errors.New("exchange and routingKey are required")
|
|
}
|
|
b, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return r.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{
|
|
ContentType: "application/json",
|
|
DeliveryMode: amqp.Persistent,
|
|
Timestamp: time.Now(),
|
|
Body: b,
|
|
})
|
|
}
|