dataRT/data/publish_event.go

package data

import (
	"context"
	"datart/data/rabbit"
	"datart/log"
	"encoding/json"
	"errors"
	"fmt"

	rmq "github.com/rabbitmq/rabbitmq-amqp-go-client/pkg/rabbitmqamqp"
)

var eventNotifyXQK = rabbit.XQK{
	Exchange: "event_notify_fanout",
}

var eventNotifyPublisher *rmq.Publisher

func init() {
	ctx := context.Background()

	m := new(rabbit.Manage)

	rm := rabbit.DefaultManagement()

	rx := &rmq.FanOutExchangeSpecification{
		Name: eventNotifyXQK.Exchange,
	}

	m.Init(ctx, rm, rx, nil, nil)

	m.DeclareExchange(ctx)

	publisher, err := rabbit.NewPublisher(ctx, "default", &eventNotifyXQK)
	if err != nil {
		panic(err)
	}
	eventNotifyPublisher = publisher
}

func PublishEvent(ctx context.Context, event any) error {
	data, err := json.Marshal(event)
	if err != nil {
		return err
	}

	result, err := eventNotifyPublisher.Publish(ctx, rmq.NewMessage(data))
	if err != nil {
		return err
	}

	switch result.Outcome.(type) {
	case *rmq.StateAccepted:
		// "message accepted"
	case *rmq.StateReleased:
		// "message released"
	case *rmq.StateRejected:
		return errors.New("message rejected")
	default:
		return fmt.Errorf("invalid message state: %v", result.Outcome)
	}

	return nil
}

func CloseEventPublisher(ctx context.Context) {
	if err := eventNotifyPublisher.Close(ctx); err != nil {
		log.Error(err)
	}
}