platform/rabbit/topic.go

package rabbit

import (
	"encoding/json"
	"log"
	"time"

	"git.k6n.net/mats/platform/event"
	amqp "github.com/rabbitmq/amqp091-go"
)

// PublishEvent marshals an event.Event and publishes it to exchange with
// routingKey = string(ev.Type). Opens and closes a channel per call. The
// exchange must already exist (declare it with ListenToPattern or call
// ExchangeDeclare explicitly before the first publish).
func PublishEvent(c *amqp.Connection, exchange string, ev event.Event) error {
	return PublishEventTo(c, exchange, string(ev.Type), ev)
}

// PublishEventTo marshals an event.Event and publishes it to exchange with the
// caller-supplied routingKey. PublishEvent hardcodes routingKey =
// string(ev.Type); partitioned producers (track.partition.{N}) supply their own
// routing key here. Opens and closes a channel per call; the exchange must
// already exist.
func PublishEventTo(c *amqp.Connection, exchange, routingKey string, ev event.Event) error {
	body, err := json.Marshal(ev)
	if err != nil {
		return err
	}
	ch, err := c.Channel()
	if err != nil {
		return err
	}
	defer ch.Close()
	return ch.Publish(exchange, routingKey, true, false, amqp.Publishing{
		ContentType:  "application/json",
		Body:         body,
		DeliveryMode: amqp.Persistent,
		Timestamp:    time.Now(),
	})
}

// ListenToPattern binds an exclusive queue with a routing-key pattern (wildcard)
// on a durable topic exchange, so one subscription receives every message
// matching the pattern. A handler error nacks (drops) that single message and
// keeps consuming.
func ListenToPattern(ch *amqp.Channel, exchange, pattern string, handler func(amqp.Delivery) error) error {
	if err := ch.ExchangeDeclare(exchange, "topic", true, false, false, false, nil); err != nil {
		return err
	}
	q, err := ch.QueueDeclare("", false, false, true, false, nil)
	if err != nil {
		return err
	}
	if err := ch.QueueBind(q.Name, pattern, exchange, false, nil); err != nil {
		return err
	}
	msgs, err := ch.Consume(q.Name, "", false, false, false, false, nil)
	if err != nil {
		return err
	}
	go func() {
		for d := range msgs {
			if err := handler(d); err != nil {
				log.Printf("Error processing pattern message: %v", err)
				d.Nack(false, false)
			} else {
				d.Ack(false)
			}
		}
	}()
	return nil
}