surveyService/util/rabbitmq/index.go

package rabbitmq

import (
	"context"
	"fmt"
	"os"
	"time"

	jsoniter "github.com/json-iterator/go"
	"github.com/streadway/amqp"
)

type WebhookMessage struct {
	Body   any `json:"body"`
	Config struct {
		Url string `json:"url"`
	} `json:"config"`
}

var json = jsoniter.ConfigCompatibleWithStandardLibrary

func Webhook(webhookMessage WebhookMessage) error {
	conn, err := Init()
	if err != nil {
		return err
	}
	p, err := conn.Producer(&Exchange{
		Name: os.Getenv("RABBITMQ_WEBHOOK_EXCHANGE"),
		Kind: amqp.ExchangeDirect,
		Key:  os.Getenv("RABBITMQ_WEBHOOK_ROUTING"),
	}, nil)
	if err != nil {
		return err
	}

	message, err := json.MarshalToString(webhookMessage)
	if err != nil {
		return err
	}
	fmt.Println("message", message)

	err = p.PublishExchange(amqp.Publishing{
		ContentType:  "text/plain",
		DeliveryMode: amqp.Persistent,
		Body:         []byte(message),
	})
	if err != nil {
		return err
	}
	// 关闭连接
	defer conn.conn.Close()
	return nil
}

func Init() (*Conn, error) {
	var broker2 = Broker{
		Ssl:       false,                          // bool
		Username:  os.Getenv("RABBITMQ_USER"),     // string
		Password:  os.Getenv("RABBITMQ_PASSWORD"), // string
		Server:    os.Getenv("RABBITMQ_HOST"),     // string
		Port:      os.Getenv("RABBITMQ_PORT"),     // string
		Vhost:     os.Getenv("RABBITMQ_VHOST"),    // string
		TSL:       nil,                            // *tls.Config
		ProxyAddr: "",                             // string
		Beatime:   15 * time.Second,               // time.Duration
	}

	conn, err := New(context.Background(), broker2)
	if err != nil {
		return nil, err
	}
	return conn, nil
}