consumer

Documentation

Index

• func NewFactory(bootstrapServers []string, groupID string, topic string, ...) *factory
• type ClosableConsumer
• type GroupConsumer
  • func NewGroupConsumer(opts GroupConsumerOpts) (*GroupConsumer, error)
  • func (gc *GroupConsumer) Close() error
  • func (gc *GroupConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result
• type GroupConsumerOpts
• type MultiPartitionConsumer
  • func NewMultiPartitionConsumer(ctx context.Context, opts MultiPartitionConsumerOpts) (*MultiPartitionConsumer, error)
  • func (mpc *MultiPartitionConsumer) Close() error
  • func (mpc *MultiPartitionConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result
• type MultiPartitionConsumerOpts
  • func (opts *MultiPartitionConsumerOpts) Paritions(ctx context.Context) ([]kafka.Partition, error)
• type PartitionConsumer
  • func NewPartitionConsumer(ctx context.Context, opts PartitionConsumerOpts) (*PartitionConsumer, error)
  • func (pc *PartitionConsumer) Close() error
  • func (pc *PartitionConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result
• type PartitionConsumerOpts
• type Result
• type WireDecoder

Functions

func NewFactory

func NewFactory(
	bootstrapServers []string,
	groupID string,
	topic string,
	partition mo.Either[int, string],
	offset int64,
	fromDateTime time.Time,
	wireFormat wire.Format,
	protoFile string,
	messageType string,
	includePaths []string,
) *factory

Types

type ClosableConsumer

type ClosableConsumer interface {
	io.Closer
	Consume(ctx context.Context, decoder WireDecoder) <-chan Result
}

type GroupConsumer

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

func NewGroupConsumer

func NewGroupConsumer(opts GroupConsumerOpts) (*GroupConsumer, error)

func (*GroupConsumer) Close

func (gc *GroupConsumer) Close() error

func (*GroupConsumer) Consume

func (gc *GroupConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result

type GroupConsumerOpts

type GroupConsumerOpts struct {
	BootstrapServers []string `validate:"required"`
	GroupID          string   `validate:"required"`
	Topic            string   `validate:"required"`
}

type MultiPartitionConsumer

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

func NewMultiPartitionConsumer

func NewMultiPartitionConsumer(ctx context.Context, opts MultiPartitionConsumerOpts) (*MultiPartitionConsumer, error)

func (*MultiPartitionConsumer) Close

func (mpc *MultiPartitionConsumer) Close() error

func (*MultiPartitionConsumer) Consume

func (mpc *MultiPartitionConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result

type MultiPartitionConsumerOpts

type MultiPartitionConsumerOpts struct {
	BootstrapServers []string `validate:"required"`
	Topic            string   `validate:"required"`
	Offset           int64
	FromDateTime     time.Time
}

func (*MultiPartitionConsumerOpts) Paritions

func (opts *MultiPartitionConsumerOpts) Paritions(ctx context.Context) ([]kafka.Partition, error)

type PartitionConsumer

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

func NewPartitionConsumer

func NewPartitionConsumer(ctx context.Context, opts PartitionConsumerOpts) (*PartitionConsumer, error)

func (*PartitionConsumer) Close

func (pc *PartitionConsumer) Close() error

func (*PartitionConsumer) Consume

func (pc *PartitionConsumer) Consume(ctx context.Context, decoder WireDecoder) <-chan Result

type PartitionConsumerOpts

type PartitionConsumerOpts struct {
	BootstrapServers []string `validate:"required"`
	Topic            string   `validate:"required"`
	Partition        int      `validate:"gte=0"`
	Offset           int64
	FromDateTime     time.Time
}

type Result

type Result struct {
	Message string
	Err     error
}

type WireDecoder

type WireDecoder interface {
	Decode([]byte) (string, error)
}