internal

Documentation

Index

• type BatchProcessor
  • func NewBatchProcessor[T any](batchSize int, flushInterval time.Duration, fc func(items []T) error) *BatchProcessor[T]
  • func (bp *BatchProcessor[T]) Add(item T)
  • func (bp *BatchProcessor[T]) Start()
  • func (bp *BatchProcessor[T]) Stop()
• type Message
  • func NewMessage(msg *kafka.Message) *Message
  • func (m *Message) GetHeader(key string) string
  • func (m *Message) SetHeader(key, val string)
• type MessageCarrier
  • func NewMessageCarrier(msg *Message) MessageCarrier
  • func (m MessageCarrier) Get(key string) string
  • func (m MessageCarrier) Keys() []string
  • func (m MessageCarrier) Set(key string, value string)

Types

type BatchProcessor

type BatchProcessor[T any] struct {
	// contains filtered or unexported fields
}

BatchProcessor 批处理模块

func NewBatchProcessor

func NewBatchProcessor[T any](batchSize int, flushInterval time.Duration, fc func(items []T) error) *BatchProcessor[T]

NewBatchProcessor 创建新的批处理模块

func (*BatchProcessor[T]) Add

func (bp *BatchProcessor[T]) Add(item T)

Add 添加数据到队列

func (*BatchProcessor[T]) Start

func (bp *BatchProcessor[T]) Start()

Start 启动批处理模块

func (*BatchProcessor[T]) Stop

func (bp *BatchProcessor[T]) Stop()

Stop 停止批处理模块并处理剩余数据

type Message

type Message struct {
	*kafka.Message
}

func NewMessage

func NewMessage(msg *kafka.Message) *Message

func (*Message) GetHeader

func (m *Message) GetHeader(key string) string

func (*Message) SetHeader

func (m *Message) SetHeader(key, val string)

type MessageCarrier

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

MessageCarrier injects and extracts traces from a types.Message.

func NewMessageCarrier

func NewMessageCarrier(msg *Message) MessageCarrier

NewMessageCarrier returns a new MessageCarrier.

func (MessageCarrier) Get

func (m MessageCarrier) Get(key string) string

Get returns the value associated with the passed key.

func (MessageCarrier) Keys

func (m MessageCarrier) Keys() []string

Keys lists the keys stored in this carrier.

func (MessageCarrier) Set

func (m MessageCarrier) Set(key string, value string)

Set stores the key-value pair.