50 lines
1.7 KiB
Go
50 lines
1.7 KiB
Go
package mq
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"git.ipao.vip/rogee/go-sip/internal/tenant"
|
|
"log/slog"
|
|
"strings"
|
|
|
|
amqp "github.com/rabbitmq/amqp091-go"
|
|
)
|
|
|
|
type InvalidMessageHandler func(context.Context, string, string, []byte, error) error
|
|
|
|
// Configure before starting any consumers.
|
|
func (b *Broker) OwnsRejectionRoute(queue, routing string) bool {
|
|
if b.validateQueue(queue, true) != nil {
|
|
return false
|
|
}
|
|
if queue == b.controlQueue {
|
|
return routing == "d."+b.dispatcherID+".control.in"
|
|
}
|
|
prefix := "agent-call.d." + b.dispatcherID + ".task."
|
|
id := strings.TrimSuffix(strings.TrimPrefix(queue, prefix), ".v1")
|
|
route, err := tenant.TaskRoute(b.dispatcherID, id)
|
|
return err == nil && route.BindingKey == routing
|
|
}
|
|
|
|
func (b *Broker) SetInvalidMessageHandler(handler InvalidMessageHandler) { b.invalidHandler = handler }
|
|
func (b *Broker) rejectInbound(ctx context.Context, queue string, d amqp.Delivery, cause error) error {
|
|
if b.invalidHandler == nil {
|
|
if err := d.Nack(false, true); err != nil {
|
|
return errors.Join(cause, err)
|
|
}
|
|
return fmt.Errorf("invalid data has no durable rejection handler: %w", cause)
|
|
}
|
|
if err := b.invalidHandler(ctx, queue, d.RoutingKey, d.Body, cause); err != nil {
|
|
if nack := d.Nack(false, true); nack != nil {
|
|
return errors.Join(err, nack)
|
|
}
|
|
return fmt.Errorf("durable rejected-data handling failed: %w", err)
|
|
}
|
|
slog.WarnContext(ctx, "MQ invalid data handled", append(messageIdentity(d.Body), "transport", "mq", "direction", "mq->dispatcher", "queue", queue, "stage", "rejected_handled", "bytes", len(d.Body), "reason", cause)...)
|
|
if err := d.Ack(false); err != nil {
|
|
return fmt.Errorf("ack handled rejected data: %w", err)
|
|
}
|
|
return nil
|
|
}
|