65 lines
1.2 KiB
Go
65 lines
1.2 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"time"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
type AmqpOrderHandler struct {
|
|
conn *amqp.Connection
|
|
queue *amqp.Queue
|
|
}
|
|
|
|
func NewAmqpOrderHandler(conn *amqp.Connection) *AmqpOrderHandler {
|
|
return &AmqpOrderHandler{
|
|
conn: conn,
|
|
}
|
|
}
|
|
|
|
func (h *AmqpOrderHandler) DefineTopics() error {
|
|
ch, err := h.conn.Channel()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to open a channel: %w", err)
|
|
}
|
|
defer ch.Close()
|
|
|
|
queue, err := ch.QueueDeclare(
|
|
"orders", // name
|
|
false,
|
|
false,
|
|
false,
|
|
false,
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to declare an exchange: %w", err)
|
|
}
|
|
h.queue = &queue
|
|
return nil
|
|
}
|
|
|
|
func (h *AmqpOrderHandler) OrderCompleted(body []byte) error {
|
|
ch, err := h.conn.Channel()
|
|
if err != nil {
|
|
return fmt.Errorf("failed to open a channel: %w", err)
|
|
}
|
|
defer ch.Close()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
return ch.PublishWithContext(ctx,
|
|
"", // exchange
|
|
"orders", // routing key
|
|
false, // mandatory
|
|
false, // immediate
|
|
amqp.Publishing{
|
|
DeliveryMode: amqp.Persistent,
|
|
ContentType: "application/json",
|
|
Body: body,
|
|
})
|
|
|
|
}
|