pipeline

README

pkg/sync/pipeline

提供内存批量聚合工具

Documentation

Index

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

Variables

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

ErrFull channel full error

Types

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 Pipeline config

type Pipeline

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

Pipeline pipeline struct

func NewPipeline

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

NewPipeline new pipline

func (*Pipeline) Add

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

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

func (*Pipeline) Close

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

Close all goroutinue

func (*Pipeline) Start

func (p *Pipeline) Start()

Start start all mergeproc

func (*Pipeline) SyncAdd

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

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