msghelper

package

v0.0.1 Latest Latest Go to latest Published: May 25, 2022 License: MIT Imports: 2 Imported by: 0

Details

This section is empty.

This section is empty.

func AddHeader(key string, value []byte) optparams.Option[kafka.Message]

AddHeader 添加一个header @params key string header键 @params value []byte header值

func ExtractHeaders(msg *kafka.Message) (map[string][]byte, error)

ExtractValue 从消息中获取headers字典 @params msg *kafka.Message 消息指针

func ExtractKey(msg *kafka.Message) ([]byte, error)

ExtractValue 从消息中获取key @params msg *kafka.Message 消息指针

func ExtractTopic(msg *kafka.Message) string

ExtractTopic 从消息中获取来源的topic @params msg *kafka.Message 消息指针

func ExtractValue(msg *kafka.Message) ([]byte, error)

ExtractValue 从消息中获取value @params msg *kafka.Message 消息指针

func NewMsg(topic string, value []byte, opts ...optparams.Option[kafka.Message]) *kafka.Message

NewMsg 构造消息用于发送 @params topic string 消息要发送去的topic @params value []byte 消息的值 @params opts ...optparams.Option[kafka.Message] 消息的其他设置

func WithHeaders(headers map[string][]byte) optparams.Option[kafka.Message]

WithHeaders 设置headers @params headers map[string][]byte header键值对

func WithKey(key []byte) optparams.Option[kafka.Message]

WithKey 设置key @params key []byte key的值

func WithPartition(partition int32) optparams.Option[kafka.Message]

WithPartition 设置发送去的Partition @params partition int32 分区号,默认为kafka.PartitionAny,即-1

type ConciseMsg struct {
	Topic   string
	Value   []byte
	Key     []byte
	Headers map[string][]byte
}

ConciseMsg 简化版本的消息对象,*kafka.Message信息过全,结构略复杂并不太利于利用

func Extract(msg *kafka.Message) (*ConciseMsg, error)

Extract 从消息中提取精简信息

func (c *ConciseMsg) AsMessage(opts ...optparams.Option[kafka.Message]) *kafka.Message

AsMessage 将精简消息转化为kafka消息 @params opts ...optparams.Option[kafka.Message] 只可以使用参数WithPartition用于指定希望发送去的分区