mqenv

Documentation

Index

• Constants
• 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"
	DriverTypePulsar = "pulsar"
	DriverTypeMock   = "mock"

	MQTypeConsumer  = 1
	MQTypePublisher = 2

	MQEventCodeOk     = 0
	MQEventCodeFailed = -1
	MQEventCodeClosed = -9

	MQReconnectSeconds        = 1
	MQQueueStatusFreshSeconds = 30
)

Constants

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