batch

Documentation

Index

• type Consumer
• type KeyBatchProcessor
  • func NewBatchProcessorWithKey(s int, c KeyConsumer, opts ...OptionFunc) KeyBatchProcessor
• type KeyConsumer
• type OptionFunc
  • func WithBatchThresholdStrategy(n int) OptionFunc
  • func WithItemsWeightStrategy(calculator func(i interface{}) int, threshold int) OptionFunc
  • func WithMaxWaitStrategy(i time.Duration) OptionFunc
• type Processor
  • func NewBatchProcessor(s int, c Consumer, opts ...OptionFunc) Processor

Types

type Consumer

type Consumer interface {
	// must be reentrancy
	Consume([]interface{})
}

type KeyBatchProcessor

type KeyBatchProcessor interface {
	TryPut(string, interface{}) bool

	Run()
	Shutdown()

	KeyNum() int
	BufferNum() int

	AdjustBatchMaxSize(int)
	AdjustBatchMaxWait(duration int)
	// contains filtered or unexported methods
}

func NewBatchProcessorWithKey

func NewBatchProcessorWithKey(s int, c KeyConsumer, opts ...OptionFunc) KeyBatchProcessor

type KeyConsumer

type KeyConsumer interface {
	// must be reentrancy
	Consume(string, []interface{})
}

type OptionFunc

type OptionFunc func(*options)

func WithBatchThresholdStrategy

func WithBatchThresholdStrategy(n int) OptionFunc

func WithItemsWeightStrategy

func WithItemsWeightStrategy(calculator func(i interface{}) int, threshold int) OptionFunc

func WithMaxWaitStrategy

func WithMaxWaitStrategy(i time.Duration) OptionFunc

type Processor

type Processor interface {
	Put(interface{})
	TryPut(interface{}) bool

	Run()
	Shutdown()

	Num() int

	AdjustBatchMaxSize(int)
	AdjustBatchMaxWait(duration int)
	// contains filtered or unexported methods
}

func NewBatchProcessor

func NewBatchProcessor(s int, c Consumer, opts ...OptionFunc) Processor