pulsar

Documentation

Index

• Constants
• func NewLogger(enableDebugLogger bool) pulsarlog.Logger
• func SetupTrackerQueue(queueName string)
• type Config
  • func (me *Config) Equals(to *Config) bool
• type PulsarMQ
  • func GetPulsarMQ(name string) (*PulsarMQ, error)
  • func InitPulsarMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, pulsarCfg *Config) (*PulsarMQ, error)
  • func NewPulsarMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, pulsarCfg *Config) *PulsarMQ
  • func (r *PulsarMQ) QueryRPC(pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)
  • func (r *PulsarMQ) Run()

Constants

const (
	PropertyCorrelationID = "CorrelationId"
	PropertyReplyTo       = "ReplyTo"
	PropertyMessageID     = "MessageId"
	PropertyAppID         = "AppId"
	PropertyUserID        = "UserId"
	PropertyContentType   = "ContentType"
)

Constants

Functions

func NewLogger

func NewLogger(enableDebugLogger bool) pulsarlog.Logger

NewLogger creates a new logger which wraps

func SetupTrackerQueue

func SetupTrackerQueue(queueName string)

SetupTrackerQueue name

Types

type Config

type Config struct {
	Topic          string
	ConnConfigName string
	// 消息类型:
	//direct:组播,订阅同一个topic，消费者组会相同，一条消息只会被组内一个消费者接收
	//fanout:广播,订阅同一个topic，但是消费者组会使用uuid，所有组都会收到信息
	MessageType string `yaml:"messageType" json:"messageType"`
}

Config Pulsar MQ configuration

func (*Config) Equals

func (me *Config) Equals(to *Config) bool

Equals check if equals

type PulsarMQ

type PulsarMQ struct {
	Name    string
	Publish chan *mqenv.MQPublishMessage
	Consume chan *mqenv.MQConsumerProxy
	Done    chan error
	Close   chan interface{}
	// contains filtered or unexported fields
}

PulsarMQ instance

func GetPulsarMQ

func GetPulsarMQ(name string) (*PulsarMQ, error)

GetPulsarMQ get

func InitPulsarMQ

func InitPulsarMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, pulsarCfg *Config) (*PulsarMQ, error)

InitPulsarMQ init

func NewPulsarMQ

func NewPulsarMQ(mqConnName string, connCfg *mqenv.MQConnectorConfig, pulsarCfg *Config) *PulsarMQ

NewPulsarMQ with parameters

func (*PulsarMQ) QueryRPC

func (r *PulsarMQ) QueryRPC(pm *mqenv.MQPublishMessage) (*mqenv.MQConsumerMessage, error)

QueryRPC publishes a message and waiting the response

func (*PulsarMQ) Run

func (r *PulsarMQ) Run()

Run start 1. init the pulsar conneciton 2. expect messages from the message hub on the Publish channel 3. if the connection is closed, try to restart it