baize/framework/core/infrastructure/message_queue/common/common.go

package common

import (
	"encoding/json"
	"git.sxidc.com/go-tools/utils/strutils"
	"github.com/pkg/errors"
	"time"
)

type CloudEvent struct {
	SpecVersion     string `json:"specversion"`
	ID              string `json:"id"`
	Type            string `json:"type"`
	Source          string `json:"source"`
	DataContentType string `json:"datacontenttype"`
	Time            string `json:"time"`
	Data            string `json:"data"`
}

func NewCloudEvent(eventID string, eventType string, data string) *CloudEvent {
	return &CloudEvent{
		SpecVersion:     "v1.0",
		ID:              eventID,
		Type:            eventType,
		Source:          "baize.sxidc.com",
		DataContentType: "application/json",
		Time:            time.Now().Format(time.RFC3339),
		Data:            data,
	}
}

func UnmarshalCloudEvent(data []byte) (*CloudEvent, error) {
	event := new(CloudEvent)
	err := json.Unmarshal(data, event)
	if err != nil {
		return nil, errors.New(err.Error())
	}

	return event, nil
}

func (event *CloudEvent) Marshal() ([]byte, error) {
	if strutils.IsStringEmpty(event.SpecVersion) {
		return nil, errors.New("没有传递事件规范版本")
	}

	if strutils.IsStringEmpty(event.ID) {
		return nil, errors.New("没有传递事件ID")
	}

	if strutils.IsStringEmpty(event.Type) {
		return nil, errors.New("没有传递事件类型")
	}

	if strutils.IsStringEmpty(event.Source) {
		return nil, errors.New("没有传递事件源")
	}

	if strutils.IsStringEmpty(event.DataContentType) {
		return nil, errors.New("没有传递事件数据类型")
	}

	if strutils.IsStringEmpty(event.Time) {
		return nil, errors.New("没有传递事件时间")
	}

	eventJson, err := json.Marshal(event)
	if err != nil {
		return nil, errors.New(err.Error())
	}

	return eventJson, nil
}

// MessageHandler 消息处理函数
// 参数:
// - queue: 消息队列
// - topic: 主题
// - data: 消息数据
// 返回值: 无
type MessageHandler func(queue MessageQueue, topic string, event *CloudEvent)

// MessageQueue 消息队列接口
type MessageQueue interface {
	Subscribe(group string, topic string, handler MessageHandler) error
	UnSubscribe(group string, topic string) error
	Publish(topic string, event *CloudEvent) error
}