pipeline

README

pkg/sync/pipeline

提供内存批量聚合工具

Documentation

Index

• Variables
• type Aggregation
  • func NewPipeline(config *Config) (res *Aggregation)
  • func (p *Aggregation) Add(c context.Context, key string, value interface{}) (err error)
  • func (p *Aggregation) Close() (err error)
  • func (p *Aggregation) Start()
  • func (p *Aggregation) SyncAdd(c context.Context, key string, value interface{}) (err error)
• type Config

Variables

var ErrFull = errors.New("channel full")

ErrFull channel full error

Types

type Aggregation

type Aggregation struct {
	Do    func(c context.Context, index int, values map[string][]interface{})
	Split func(key string) int
	// contains filtered or unexported fields
}

Aggregation pipeline struct

func NewPipeline

func NewPipeline(config *Config) (res *Aggregation)

NewPipeline new pipline

func (*Aggregation) Add

func (p *Aggregation) Add(c context.Context, key string, value interface{}) (err error)

Add async add a value to channal, channel shard in split method

func (*Aggregation) Close

func (p *Aggregation) Close() (err error)

Close all goroutinue

func (*Aggregation) Start

func (p *Aggregation) Start()

Start start all mergeproc

func (*Aggregation) SyncAdd

func (p *Aggregation) SyncAdd(c context.Context, key string, value interface{}) (err error)

SyncAdd sync add a value to channal, channel shard in split method

type Config

type Config struct {
	// MaxSize merge size
	MaxSize int
	// Interval merge interval
	Interval xtime.Duration
	// Buffer channel size
	Buffer int
	// Worker channel number
	Worker int
	// Name use for metrics
	Name string
}

Config Aggregation config