mq.go

package dgmq

import (
	"errors"
	"time"

	dgctx "github.com/darwinOrg/go-common/context"
)

const (
	RequestIdHeader = "request_id"
)

type SubscribeHandler func(ctx *dgctx.DgContext, message string) error
type SubscribeEndCallback func()

type MqAdapter interface {
	Publisher
	Subscriber
}

type Publisher interface {
	CreateTopic(ctx *dgctx.DgContext, topic string) error
	Publish(ctx *dgctx.DgContext, topic string, message any) error
	PublishWithTag(ctx *dgctx.DgContext, topic, tag string, message any) error
	PublishDelay(ctx *dgctx.DgContext, topic string, message any, delay time.Duration) error
	Destroy(ctx *dgctx.DgContext, topic string) error
	Close()
}

type Subscriber interface {
	Subscribe(ctx *dgctx.DgContext, topic string, handler SubscribeHandler) (SubscribeEndCallback, error)
	SubscribeWithTag(ctx *dgctx.DgContext, topic, tag string, handler SubscribeHandler) (SubscribeEndCallback, error)
	DynamicSubscribe(ctx *dgctx.DgContext, closeCh chan struct{}, topic string, handler SubscribeHandler) error
	SubscribeDelay(ctx *dgctx.DgContext, topic string, sleepDuration time.Duration, handler SubscribeHandler) (SubscribeEndCallback, error)
	Unsubscribe(ctx *dgctx.DgContext, topic string) error
	UnsubscribeWithTag(ctx *dgctx.DgContext, topic, tag string) error
}

const (
	MqAdapterRedisList   = "redis_list"
	MqAdapterRedisStream = "redis_stream"
	MqAdapterNats        = "nats"
)

type MqAdapterConfig struct {
	Type      string        `json:"type" mapstructure:"type"`
	Host      string        `json:"host" mapstructure:"host"`
	Port      int           `json:"port" mapstructure:"port"`
	Username  string        `json:"username" mapstructure:"username"`
	Password  string        `json:"password" mapstructure:"password"`
	Timeout   time.Duration `json:"timeout" mapstructure:"timeout"`
	PoolSize  int           `json:"poolSize" mapstructure:"poolSize"`
	Group     string        `json:"group" mapstructure:"group"`
	Consumer  string        `json:"consumer" mapstructure:"consumer"`
	BatchSize int64         `json:"batchSize" mapstructure:"batchSize"`
}

func NewMqAdapter(config *MqAdapterConfig) (MqAdapter, error) {
	var mqAdapter MqAdapter
	switch config.Type {
	case MqAdapterRedisList:
		mqAdapter = NewRedisListAdapter(config)
	case MqAdapterRedisStream:
		mqAdapter = NewRedisStreamAdapter(config)
	case MqAdapterNats:
		mqAdapter = NewNatsAdapter(config)
	default:
		return nil, errors.New("错误的类型")
	}

	return mqAdapter, nil
}