task

Feb 5, 2025

Changes in this version

+ const MaxCountInBuf

+ const MaxParallelism

+ type Commit struct

+ type Consumer struct

type Service

+ func (service *Service) Put(msg *model.InputMessage, traceId string, flushFn func(traceId, with string)) error

type Sharder

+ func (sh *Sharder) Flush(c context.Context, wg *sync.WaitGroup, rmap map[int32]*model.BatchRange, ...)

+ func (sh *Sharder) PutElement(msgRow *model.MsgRow)

+ type Sinker struct

+ func NewSinker(rcm cm.RemoteConfManager, http string, cmd *util.CmdOptions) *Sinker

+ func (s *Sinker) Close()

+ func (s *Sinker) GetCurrentConfig() *config.Config

+ func (s *Sinker) Init() (err error)

+ func (s *Sinker) Run()

Oct 12, 2022

Changes in this version

+ type Ring struct

+ func (ring *Ring) ForceBatchOrShard(_ interface{})

+ func (ring *Ring) MakeRoom(newMsg *model.InputMessage)

+ func (ring *Ring) PutElem(msgRow model.MsgRow)

+ func (ring *Ring) PutMsgNolock(msg *model.InputMessage)

+ func (ring *Ring) QuitIdle()

+ type Service struct

+ func NewTaskService(cfg *config.Config, taskCfg *config.TaskConfig) (service *Service)

+ func (service *Service) Flush(batch *model.Batch) (err error)

+ func (service *Service) Init() (err error)

+ func (service *Service) Run()

+ func (service *Service) Stop()

+ type Sharder struct

+ func NewSharder(service *Service) (sh *Sharder, err error)

+ func (sh *Sharder) Calc(row *model.Row) (int, error)

+ func (sh *Sharder) ForceFlush(arg interface{})

+ func (sh *Sharder) PutElems(partition int, ringBuf []model.MsgRow, begOff, endOff, ringCapMask int64)

+ type ShardingPolicy struct

+ func NewShardingPolicy(shardingKey string, shardingStripe uint64, dims []*model.ColumnWithType, ...) (policy *ShardingPolicy, err error)

+ func (policy *ShardingPolicy) Calc(row *model.Row) (shard int, err error)