mq

Documentation

Index

• Constants
• func ConsumeMQ(mqCategory string, consumeProxy *mqenv.MQConsumerProxy) error
• func GetAllMQDriverConfigs() map[string]mqenv.MQConnectorConfig
• func GetKafka(name string) (*kafka.Kafka, error)
• func GetMQRoutes() map[string]Config
• func GetRabbitMQ(name string) (*rabbitmq.RabbitMQ, error)
• 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 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 QueryMQRPC(mqCategory string, pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)
• 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 GetAllMQDriverConfigs

func GetAllMQDriverConfigs() map[string]mqenv.MQConnectorConfig

GetAllMQDriverConfigs configs

func GetKafka

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

GetKafka get kafka

func GetMQRoutes

func GetMQRoutes() map[string]Config

GetMQRoutes config map

func GetRabbitMQ

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

GetRabbitMQ get rabbitmq

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 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 QueryMQRPC

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

QueryMQRPC publishes a message and waiting the response

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"`
}

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"`
}

RoutesEnv struct

func InitMQRoutesEnv

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

InitMQRoutesEnv initialize with configure file