batcher

README

Batcher

Batcher service batches individual client insert calls, into centralized bulk insert.

Client1: INSERT INTO X
Client2: INSERT INTO X     ->  Batch(by time or max elements) -> Execute BULK INSERT (updated autoincrement back to clients)

ClientN: INSERT INTO X

Motivation

Usage

Documentation

Index

• type Batch
  • func NewBatch(rType reflect.Type, maxElements int, maxDurationMs int) *Batch
  • func (s *Batch) CanFlush() bool
  • func (b *Batch) HasExpired() bool
  • func (b *Batch) TryAcquire() bool
• type Collection
  • func NewCollection(rType reflect.Type) *Collection
  • func (c *Collection) Append(value interface{})
  • func (c *Collection) Len() int
  • func (c *Collection) Reset()
  • func (c *Collection) Unwrap() interface{}
  • func (c *Collection) ValuePointerAt(index int) interface{}
• type Config
• type Service
  • func New(ctx context.Context, inserter *insert.Service, rType reflect.Type, ...) (*Service, error)
  • func (s *Service) Collect(recPtr interface{}) (*State, error)
• type State
  • func (s *State) Wait() error

Types

type Batch

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

Batch represents batch for collecting data before inserting into db

func NewBatch

func NewBatch(rType reflect.Type, maxElements int, maxDurationMs int) *Batch

NewBatch creates a new batch

func (*Batch) CanFlush

func (s *Batch) CanFlush() bool

CanFlush checks possibility of flushing batch

func (*Batch) HasExpired

func (b *Batch) HasExpired() bool

HasExpired checks if the batch exceeds its max duration time

func (*Batch) TryAcquire

func (b *Batch) TryAcquire() bool

TryAcquire checks if batch can be used for collecting data

type Collection

type Collection struct {
	RWMutex sync.RWMutex
	// contains filtered or unexported fields
}

Collection represents slice to collecting data

func NewCollection

func NewCollection(rType reflect.Type) *Collection

NewCollection creates a new collection

func (*Collection) Append

func (c *Collection) Append(value interface{})

Append adds item into collection

func (*Collection) Len

func (c *Collection) Len() int

Len returns count of items in collection

func (*Collection) Reset

func (c *Collection) Reset()

Reset set length collection to 0 (doesn't change capacity)

func (*Collection) Unwrap

func (c *Collection) Unwrap() interface{}

Unwrap unwrap collection

func (*Collection) ValuePointerAt

func (c *Collection) ValuePointerAt(index int) interface{}

ValuePointerAt returns item at given index

type Config

type Config struct {
	MaxElements   int
	MaxDurationMs int
	BatchSize     int
}

Config represents batcher's config

type Service

type Service struct {
	RecordType reflect.Type
	// contains filtered or unexported fields
}

Service represents batcher service

func New

func New(ctx context.Context, inserter *insert.Service, rType reflect.Type, config *Config) (*Service, error)

New creates a batcher service

func (*Service) Collect

func (s *Service) Collect(recPtr interface{}) (*State, error)

Collect puts data into a batch

type State

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

State represents state of batch

func (*State) Wait

func (s *State) Wait() error

Wait returns state's error