mq

Documentation

Index

• Constants
• func ConsumeMQ(mqCategory string, consumeProxy *mqenv.MQConsumerProxy) error
• func FindOneCategoryNameByInstance(instanceName string) string
• func GetAllCategoryNamesByInstance(instanceName string) []string
• func GetAllMQDriverConfigs() map[string]mqenv.MQConnectorConfig
• func GetKafka(name string) (*kafka.KafkaWorker, error)
• func GetMQRoutes() map[string]Config
• func GetPulsar(name string) (*pulsar.PulsarMQ, error)
• func GetRabbitMQ(name string) (*rabbitmq.RabbitMQ, error)
• func HasConnections() bool
• func Init(mqConfigFile string, mqDriverConfigs map[string]mqenv.MQConnectorConfig) error
• func InitMQTopic(topicCategory string, topicConfig *Config, ...) error
• func InitMQWithRPC(topicCategory string, rpcType int, connCfg *mqenv.MQConnectorConfig, ...) error
• func InitMockMQTopic(mqCategory string, topic string)
• func InitWithMQRoutes(mqRoutesEnv *RoutesEnv, mqDriverConfigs map[string]mqenv.MQConnectorConfig) error
• func NewMQResponseMessage(body []byte, cm *mqenv.MQConsumerMessage) *mqenv.MQPublishMessage
• func PublishMQ(mqCategory string, publishMsg *mqenv.MQPublishMessage) error
• func QueryMQ(mqCategory string, pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)
• func QueryMQRPC(mqCategory string, pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)
• func SetMQConfig(category string, cnf Config)
• func SetupTrackerQueue(queueName string)
• type Config
  • func GetMQConfig(category string) *Config
• type Exchange
• type RoutesEnv
  • func InitMQRoutesEnv(configFile string) (*RoutesEnv, error)

Constants

const (
	DriverTypeAMQP  = mqenv.DriverTypeAMQP
	DriverTypeKafka = mqenv.DriverTypeKafka

	MQTypeConsumer  = mqenv.MQTypeConsumer
	MQTypePublisher = mqenv.MQTypePublisher

	MQEventCodeOk     = mqenv.MQEventCodeOk
	MQEventCodeFailed = mqenv.MQEventCodeFailed
)

Constants

Functions

func ConsumeMQ

func ConsumeMQ(mqCategory string, consumeProxy *mqenv.MQConsumerProxy) error

ConsumeMQ consume

func FindOneCategoryNameByInstance

func FindOneCategoryNameByInstance(instanceName string) string

FindOneCategoryNameByInstance first hit category

func GetAllCategoryNamesByInstance

func GetAllCategoryNamesByInstance(instanceName string) []string

GetAllCategoryNamesByInstance by instancename

func GetAllMQDriverConfigs

func GetAllMQDriverConfigs() map[string]mqenv.MQConnectorConfig

GetAllMQDriverConfigs configs

func GetKafka

func GetKafka(name string) (*kafka.KafkaWorker, error)

GetKafka get kafka instance

func GetMQRoutes

func GetMQRoutes() map[string]Config

GetMQRoutes config map

func GetPulsar

func GetPulsar(name string) (*pulsar.PulsarMQ, error)

GetPulsar get pulsar instance

func GetRabbitMQ

func GetRabbitMQ(name string) (*rabbitmq.RabbitMQ, error)

GetRabbitMQ get rabbitmq instance

func HasConnections

func HasConnections() bool

HasConnections returns true if has any mq connection

func Init

func Init(mqConfigFile string, mqDriverConfigs map[string]mqenv.MQConnectorConfig) error

Init initializer

func InitMQTopic

func InitMQTopic(topicCategory string, topicConfig *Config, mqDriverConfigs map[string]mqenv.MQConnectorConfig) error

InitMQTopic initialize sigle mq topic with drivers

func InitMQWithRPC

func InitMQWithRPC(topicCategory string, rpcType int, connCfg *mqenv.MQConnectorConfig, mqCfg *Config) error

InitMQWithRPC init mq with RPC

func InitMockMQTopic

func InitMockMQTopic(mqCategory string, topic string)

InitMockMQTopic for testing

func InitWithMQRoutes

func InitWithMQRoutes(mqRoutesEnv *RoutesEnv, mqDriverConfigs map[string]mqenv.MQConnectorConfig) error

InitWithMQRoutes patitionally init with RoutesEnv

func NewMQResponseMessage

func NewMQResponseMessage(body []byte, cm *mqenv.MQConsumerMessage) *mqenv.MQPublishMessage

NewMQResponseMessage new mq response publish messge depends on mq consumer message

func PublishMQ

func PublishMQ(mqCategory string, publishMsg *mqenv.MQPublishMessage) error

PublishMQ publish

func QueryMQ

func QueryMQ(mqCategory string, pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)

QueryMQ publishes a message and waiting the response

func QueryMQRPC

func QueryMQRPC(mqCategory string, pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)

QueryMQRPC publishes a message and waiting the response

func SetMQConfig

func SetMQConfig(category string, cnf Config)

SetMQConfig with category

func SetupTrackerQueue

func SetupTrackerQueue(queueName string)

SetupTrackerQueue name

Types

type Config

type Config struct {
	Instance string `yaml:"instance" json:"instance"`
	// RabbitMQ parameters
	Queue       string            `yaml:"queue" json:"queue"`
	Exchange    Exchange          `yaml:"exchange" json:"exchange"`
	BindingKey  string            `yaml:"bindingKey" json:"bindingKey"`
	RoutingKeys map[string]string `yaml:"routingKeys" json:"routingKeys"`
	Durable     bool              `yaml:"durable" json:"durable"`
	AutoDelete  bool              `yaml:"autoDelete" json:"autoDelete"`
	RPCEnabled  bool              `yaml:"rpcEnabled"`
	// Kafka parameters
	Topic             string `yaml:"topic" json:"topic"`
	GroupID           string `yaml:"groupId" json:"groupId"`
	Partition         int    `yaml:"partition" json:"partition"`
	MaxPollIntervalMS int    `yaml:"maxPollIntervalMs" json:"maxPollIntervalMs"`
	// 消息类型:
	//direct:组播,订阅同一个topic，消费者组会相同，一条消息只会被组内一个消费者接收
	//fanout:广播,订阅同一个topic，但是消费者组会使用uuid，所有组都会收到信息
	MessageType        string `yaml:"messageType" json:"messageType"`
	UseOriginalContent bool   `yaml:"useOriginalContent" json:"useOriginalContent"`
}

Config struct

func GetMQConfig

func GetMQConfig(category string) *Config

GetMQConfig config

type Exchange

type Exchange struct {
	Type    string `yaml:"type" json:"type"`
	Name    string `yaml:"name" json:"name"`
	Durable bool   `yaml:"durable" json:"durable"`
}

Exchange struct

type RoutesEnv

type RoutesEnv struct {
	MQs map[string]Config `yaml:"mq"`
	// contains filtered or unexported fields
}

RoutesEnv struct

func InitMQRoutesEnv

func InitMQRoutesEnv(configFile string) (*RoutesEnv, error)

InitMQRoutesEnv initialize with configure file