utils

Documentation

Index

• func CalculateMsgLength(message pulsar.Message) int
• func ConvertMap2Headers(m map[string]string) []*codec.Header
• func DebugKafkaTopicMatch(s set.Set[string], topic string) bool
• func DebugPulsarPartitionTopicMatch(s set.Set[string], partitionTopic string) bool
• func DebugTopicMatch(kSet, pSet set.Set[string], kTopic, pTopic string) bool
• func GetLatestMsgId(partitionedTopic string, client *padmin.PulsarAdmin) (*padmin.MessageId, error)
• func PartitionedTopic(topic string, partition int) string
• func ReadEarliestMsg(partitionedTopic string, maxWaitMs int, pulsarClient pulsar.Client) (pulsar.Message, error)
• func ReadLatestMsg(partitionedTopic string, maxWaitMs int, messageId *padmin.MessageId, ...) (pulsar.Message, error)
• type KeyBasedRateLimiter
  • func NewKeyBasedRateLimiter(seconds, times int) *KeyBasedRateLimiter
  • func (k *KeyBasedRateLimiter) Acquire(key string) bool
  • func (k *KeyBasedRateLimiter) Clean(key string)
• type PartitionedTopicInfo
  • func GetTenantNamespaceTopicFromPartitionedPrefix(str string) (info PartitionedTopicInfo, err error)

Functions

func CalculateMsgLength

func CalculateMsgLength(message pulsar.Message) int

func ConvertMap2Headers

func ConvertMap2Headers(m map[string]string) []*codec.Header

func DebugKafkaTopicMatch

func DebugKafkaTopicMatch(s set.Set[string], topic string) bool

func DebugPulsarPartitionTopicMatch

func DebugPulsarPartitionTopicMatch(s set.Set[string], partitionTopic string) bool

func DebugTopicMatch

func DebugTopicMatch(kSet, pSet set.Set[string], kTopic, pTopic string) bool

func GetLatestMsgId

func GetLatestMsgId(partitionedTopic string, client *padmin.PulsarAdmin) (*padmin.MessageId, error)

func PartitionedTopic

func PartitionedTopic(topic string, partition int) string

func ReadEarliestMsg

func ReadEarliestMsg(partitionedTopic string, maxWaitMs int, pulsarClient pulsar.Client) (pulsar.Message, error)

func ReadLatestMsg

func ReadLatestMsg(partitionedTopic string, maxWaitMs int, messageId *padmin.MessageId, pulsarClient pulsar.Client) (pulsar.Message, error)

Types

type KeyBasedRateLimiter

type KeyBasedRateLimiter struct {
	// contains filtered or unexported fields
}

func NewKeyBasedRateLimiter

func NewKeyBasedRateLimiter(seconds, times int) *KeyBasedRateLimiter

func (*KeyBasedRateLimiter) Acquire

func (k *KeyBasedRateLimiter) Acquire(key string) bool

func (*KeyBasedRateLimiter) Clean

func (k *KeyBasedRateLimiter) Clean(key string)

type PartitionedTopicInfo

type PartitionedTopicInfo struct {
	Tenant    string
	Namespace string
	Topic     string
	Partition int
}

func GetTenantNamespaceTopicFromPartitionedPrefix

func GetTenantNamespaceTopicFromPartitionedPrefix(str string) (info PartitionedTopicInfo, err error)