convert

Documentation

Index

• func Convert(ctx context.Context, msgs chan []*pulsarutils.ConsumerMessage, bufferSize int, ...) chan *model.BatchUpdate
• type MessageRowConverter
  • func (rc *MessageRowConverter) ConvertBatch(ctx context.Context, batch []*pulsarutils.ConsumerMessage) *model.BatchUpdate

Functions

func Convert

func Convert(ctx context.Context, msgs chan []*pulsarutils.ConsumerMessage, bufferSize int, converter *MessageRowConverter) chan *model.BatchUpdate

Convert takes a channel of pulsar message batches and outputs a channel of batched events that we store in Redis

Types

type MessageRowConverter

type MessageRowConverter struct {
	Compressor          compress.Compressor
	MaxMessageBatchSize int
}

MessageRowConverter converts raw pulsar messages into events that we can store in Redis

func (*MessageRowConverter) ConvertBatch

func (rc *MessageRowConverter) ConvertBatch(ctx context.Context, batch []*pulsarutils.ConsumerMessage) *model.BatchUpdate