consumer

Documentation

Index

• type Cfg
• type Kafka
  • func NewKafka(ctx context.Context, cfg Cfg, decoder protoparser.Decoder, ...) (*Kafka, error)
  • func (k *Kafka) Run() <-chan error

Types

type Cfg

type Cfg struct {
	URL        string
	Topic      string
	Start, End int64
	Verbose    bool
	KeyGrep    string
}

Cfg is the configuration of this consumer.

type Kafka

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

Kafka is the consumer itself.

func NewKafka

func NewKafka(ctx context.Context, cfg Cfg, decoder protoparser.Decoder, printer output.Printer) (*Kafka, error)

NewKafka returns a new instance of this consumer or an error if something isn't right.

func (*Kafka) Run

func (k *Kafka) Run() <-chan error

Run runs the consumer and consumes everything according to its configuration. If any [infra] error happens before we even started, it gets written to the output error channel. If any [parsing] error happens during the consumption, it's given to a printer. When consumer reaches the configured end offset, it stops. Otherwise, it keeps waiting for new messages. All consumers will stop if the consumer context is cancelled.