stream

Documentation

Index

• Constants
• func FromOffset(o int64) consumerOpts
• func WithEOFBehaviour(v eofBehaviour) consumerOpts
• func WithMaxBatchSize(v int) consumerOpts
• func WithMaxRecordCount(o int64) consumerOpts
• func WithMinBatchSize(v int) consumerOpts
• func WithName(v string) consumerOpts
• func WithOffsetIterator(v OffsetIterator) consumerOpts
• type Batch
• type Consumer
  • func NewConsumer(opts ...consumerOpts) Consumer
• type ConsumerOpts
• type OffsetIterator
• type Poller
• type Processor

Constants

const (
	// EOFBehaviourPoll will make the session poll for new records after an EOF error is received
	EOFBehaviourPoll eofBehaviour = 1 << iota
	// EOFBehaviourExit wil make the session exit when EOF is received
	EOFBehaviourExit eofBehaviour = 1 << iota
)

Functions

func FromOffset

func FromOffset(o int64) consumerOpts

func WithEOFBehaviour

func WithEOFBehaviour(v eofBehaviour) consumerOpts

func WithMaxBatchSize

func WithMaxBatchSize(v int) consumerOpts

func WithMaxRecordCount

func WithMaxRecordCount(o int64) consumerOpts

func WithMinBatchSize

func WithMinBatchSize(v int) consumerOpts

func WithName

func WithName(v string) consumerOpts

func WithOffsetIterator

func WithOffsetIterator(v OffsetIterator) consumerOpts

Types

type Batch

type Batch struct {
	Records []commitlog.Entry
}

type Consumer

type Consumer interface {
	Consume(ctx context.Context, r io.ReadSeeker, processor Processor) error
}

func NewConsumer

func NewConsumer(opts ...consumerOpts) Consumer

type ConsumerOpts

type ConsumerOpts struct {
	Name                      string
	MaxBatchSize              int
	MinBatchSize              int
	MaxBatchMemorySizeInBytes int
	MaxRecordCount            int64
	FromOffset                int64
	EOFBehaviour              eofBehaviour
	OffsetProvider            OffsetIterator
	Middleware                []func(Processor, ConsumerOpts) Processor
}

ConsumerOpts describes stream session preferences

type OffsetIterator

type OffsetIterator interface {
	Next() (uint64, error)
	AdvanceTo(uint64)
}

type Poller

type Poller interface {
	Ready() <-chan Batch
	Error() error
}

type Processor

type Processor func(context.Context, Batch) error

Processor is a function that will process stream records