backpressure

Documentation

Index

• Variables
• type Backpressure
• type Option
  • func BatchMaxSize(batchMaxSize int) Option
  • func BatchTimeMs(batchTimeMs float64) Option
  • func Log(log mclog.Logger) Option
  • func MaxBatchesInQueue(maxBatchesInQueue int) Option
  • func Workers(maxWorkers int) Option
• type Options
• type PressureContext
  • func NewBackpressureContext(backpressurePutMulti Backpressure, opts ...Option) (*PressureContext, error)
  • func (rc *PressureContext) Add(value interface{}) error
  • func (rc *PressureContext) Close()

Variables

var (
	// ErrBackPressureInit in case initialization fails
	ErrBackPressureInit = errors.New("backpressure run context failed to initialize")
)

Types

type Backpressure

type Backpressure interface {

	// PutMulti handles 0, 1 or more events sent
	PutMulti(events []interface{}) error
}

Backpressure interface (required implementation)

type Option

type Option func(*Options)

Option a single option

func BatchMaxSize

func BatchMaxSize(batchMaxSize int) Option

BatchMaxSize maximum number of

func BatchTimeMs

func BatchTimeMs(batchTimeMs float64) Option

BatchTimeMs how long a wait before sending list of tasks to our worker

func Log

func Log(log mclog.Logger) Option

Log - if log present it means debug enabled

func MaxBatchesInQueue

func MaxBatchesInQueue(maxBatchesInQueue int) Option

MaxBatchesInQueue - a queue with batches. Defines maximum number of batches help in the batch queue

func Workers

func Workers(maxWorkers int) Option

Workers - number of workers processing jobs

type Options

type Options struct {
	BatchTimeMs       float64
	BatchMaxSize      int
	MaxWorkers        int
	MaxBatchesInQueue int
	Log               mclog.Logger
}

Options - settings options insted of defaults

type PressureContext

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

PressureContext which combines all the channels

func NewBackpressureContext

func NewBackpressureContext(backpressurePutMulti Backpressure, opts ...Option) (*PressureContext, error)

NewBackpressureContext creates a backpressure run context and kicks off 2 go routinges (consumer and collector)

func (*PressureContext) Add

func (rc *PressureContext) Add(value interface{}) error

Add event to be handled by backpressure mechanism

func (*PressureContext) Close

func (rc *PressureContext) Close()

Close channels