pulsar

Documentation

Index

• func DeserializePulsarMsgID(messageID []byte) (pulsar.MessageID, error)
• func GetFullTopicName(tenant string, namespace string, topic string) (string, error)
• func NewAdminClient(address, authPlugin, authParams string) (pulsarctl.Client, error)
• func NewClient(tenant string, namespace string, opts pulsar.ClientOptions) (*pulsarClient, error)
• func SerializePulsarMsgID(messageID pulsar.MessageID) []byte
• type Consumer
  • func (pc *Consumer) Ack(message mqwrapper.Message)
  • func (pc *Consumer) Chan() <-chan mqwrapper.Message
  • func (pc *Consumer) CheckTopicValid(topic string) error
  • func (pc *Consumer) Close()
  • func (pc *Consumer) GetLatestMsgID() (mqwrapper.MessageID, error)
  • func (pc *Consumer) Seek(id mqwrapper.MessageID, inclusive bool) error
  • func (pc *Consumer) Subscription() string

Functions

func DeserializePulsarMsgID

func DeserializePulsarMsgID(messageID []byte) (pulsar.MessageID, error)

DeserializePulsarMsgID returns the deserialized message ID

func GetFullTopicName

func GetFullTopicName(tenant string, namespace string, topic string) (string, error)

func NewAdminClient

func NewAdminClient(address, authPlugin, authParams string) (pulsarctl.Client, error)

func NewClient

func NewClient(tenant string, namespace string, opts pulsar.ClientOptions) (*pulsarClient, error)

NewClient creates a pulsarClient object according to the parameter opts of type pulsar.ClientOptions

func SerializePulsarMsgID

func SerializePulsarMsgID(messageID pulsar.MessageID) []byte

SerializePulsarMsgID returns the serialized message ID

Types

type Consumer

type Consumer struct {
	AtLatest bool
	// contains filtered or unexported fields
}

Consumer consumes from pulsar

func (*Consumer) Ack

func (pc *Consumer) Ack(message mqwrapper.Message)

Ack the consumption of a single message

func (*Consumer) Chan

func (pc *Consumer) Chan() <-chan mqwrapper.Message

Chan returns a message channel

func (*Consumer) CheckTopicValid

func (pc *Consumer) CheckTopicValid(topic string) error

func (*Consumer) Close

func (pc *Consumer) Close()

Close the consumer and stop the broker to push more messages

func (*Consumer) GetLatestMsgID

func (pc *Consumer) GetLatestMsgID() (mqwrapper.MessageID, error)

func (*Consumer) Seek

func (pc *Consumer) Seek(id mqwrapper.MessageID, inclusive bool) error

Seek seek consume position to the pointed messageID, the pointed messageID will be consumed after the seek in pulsar

func (*Consumer) Subscription

func (pc *Consumer) Subscription() string

Subscription get a subscription for the consumer