mqenv

Documentation

Index

• Constants
• func GetPublishMessageChannelSize() int
• func SetPublishMessageChannelSize(value int) int
• type MQConnectorConfig
• type MQConsumerCallback
• type MQConsumerMessage
  • func NewConsumerMessageFromPublishMessage(pm *MQPublishMessage) MQConsumerMessage
  • func (m *MQConsumerMessage) GetHeader(name string) string
  • func (m *MQConsumerMessage) SetHeader(name string, value string)
• type MQConsumerProxy
• type MQEvent
• type MQPublishMessage
  • func NewMQResponseMessage(body []byte, cm *MQConsumerMessage) *MQPublishMessage
  • func (m *MQPublishMessage) CallbackEnabled() bool
  • func (m *MQPublishMessage) OnClosed()

Constants

const (
	DriverTypeAMQP   = "rabbitmq"
	DriverTypeKafka  = "kafka"
	DriverTypeMock   = "mock"
	DriverTypePulsar = "pulsar"

	MQTypeConsumer  = 1
	MQTypePublisher = 2

	MQEventCodeOk     = 0
	MQEventCodeFailed = -1
	MQEventCodeClosed = -9

	MQReconnectSeconds        = 1
	MQQueueStatusFreshSeconds = 30
)

Constants

Functions

func GetPublishMessageChannelSize

func GetPublishMessageChannelSize() int

GetPublishMessageChannelSize get publishing message channel size for initializing mq publish channel

func SetPublishMessageChannelSize

func SetPublishMessageChannelSize(value int) int

SetPublishMessageChannelSize set publishing message channel size for initializing mq publish channel

Types

type MQConnectorConfig

type MQConnectorConfig struct {
	Driver       string `yaml:"driver" json:"driver"`
	Host         string `yaml:"host" json:"host"`
	Port         int    `yaml:"port" json:"port"`
	Path         string `yaml:"virtualHost" json:"virtualHost"`
	User         string `yaml:"username" json:"username"`
	Password     string `yaml:"password" json:"password"`
	Timeout      int    `yaml:"timeout" json:"timeout"`
	Heartbeat    int    `yaml:"heartbeat" json:"heartbeat"`
	Disabled     bool   `yaml:"disabled" json:"disabled"`
	SSHTunnelDSN string `yaml:"sshTunnel" json:"sshTunnel"`
}

MQConnectorConfig connector config

type MQConsumerCallback

type MQConsumerCallback func(MQConsumerMessage) *MQPublishMessage

MQConsumerCallback callback

type MQConsumerMessage

type MQConsumerMessage struct {
	Driver        string            `json:"driver"`
	Queue         string            `json:"queue"`
	CorrelationID string            `json:"correlationId"`
	ConsumerTag   string            `json:"consumerTag"`
	ReplyTo       string            `json:"replyTo"`
	MessageID     string            `json:"messageId"`
	AppID         string            `json:"appId"`
	UserID        string            `json:"userId"`
	ContentType   string            `json:"contentType"`
	Exchange      string            `json:"exchange"`
	RoutingKey    string            `json:"routingKey"`
	Timestamp     time.Time         `json:"-"`
	Body          []byte            `json:"body"`
	Headers       map[string]string `json:"headers"`
	BindData      interface{}       `json:"-"`
}

MQConsumerMessage consumer message

func NewConsumerMessageFromPublishMessage

func NewConsumerMessageFromPublishMessage(pm *MQPublishMessage) MQConsumerMessage

NewConsumerMessageFromPublishMessage new consumer message from publish message

func (*MQConsumerMessage) GetHeader

func (m *MQConsumerMessage) GetHeader(name string) string

GetHeader by key

func (*MQConsumerMessage) SetHeader

func (m *MQConsumerMessage) SetHeader(name string, value string)

SetHeader value by key

type MQConsumerProxy

type MQConsumerProxy struct {
	Queue       string
	Callback    MQConsumerCallback
	ConsumerTag string
	AutoAck     bool
	Exclusive   bool
	NoLocal     bool
	NoWait      bool
	Ready       chan bool // notifies if consumer subscribes ready
}

MQConsumerProxy consumer proxy

type MQEvent

type MQEvent struct {
	Code    int    `json:"code"`
	Label   string `json:"label"`
	Message string `json:"message"`
}

MQEvent event

type MQPublishMessage

type MQPublishMessage struct {
	Body           []byte                 `json:"body"`
	Exchange       string                 `json:"exchange"`
	RoutingKey     string                 `json:"routingKey"`
	CorrelationID  string                 `json:"correlationId"`
	ReplyTo        string                 `json:"replyTo"`
	MessageID      string                 `json:"messageId"`
	AppID          string                 `json:"appId"`
	UserID         string                 `json:"userId"`
	ContentType    string                 `json:"contentType"`
	PublishStatus  chan MQEvent           `json:"-"`
	EventLabel     string                 `json:"eventLabel"`
	Headers        map[string]string      `json:"headers"`
	Response       chan MQConsumerMessage `json:"-"`
	TimeoutSeconds int

	SkipExchange bool // if publish a message only to a queue, not bind to exchange
	// contains filtered or unexported fields
}

MQPublishMessage publish message

func NewMQResponseMessage

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

NewMQResponseMessage new mq response publish messge depends on mq consumer message

func (*MQPublishMessage) CallbackEnabled

func (m *MQPublishMessage) CallbackEnabled() bool

CallbackEnabled is callback enabled

func (*MQPublishMessage) OnClosed

func (m *MQPublishMessage) OnClosed()

OnClosed on close event