batcher

Documentation

Index

• Constants
• Variables
• func NoOpProcessor[T any]([]T) error
• func ProvideBatcherInFX[T any](processorFactory any, batchSize int, batchInterval time.Duration) fx.Option
• type AtomicCounter
  • func NewAtomicCounter() *AtomicCounter
  • func (c *AtomicCounter) Add(num int64)
  • func (c *AtomicCounter) Read() int64
  • func (c *AtomicCounter) Reset()
• type Batcher
  • func New[T any](options ...Option[T]) *Batcher[T]
  • func (b *Batcher[T]) Add(item T)
  • func (b *Batcher[T]) Close() error
  • func (b *Batcher[T]) Config() *Config[T]
  • func (b *Batcher[T]) Errors() <-chan error
  • func (b *Batcher[T]) IsClosed() bool
  • func (b *Batcher[T]) Join(timeout time.Duration) error
  • func (b *Batcher[T]) Len() int
  • func (b *Batcher[T]) Start()
• type Config
• type Option
  • func WithBatchInterval[T any](batchInterval time.Duration) Option[T]
  • func WithBatchSize[T any](batchSize int) Option[T]
  • func WithProcessor[T any](fn Processor[T]) Option[T]
  • func WithSkipAutoStart[T any]() Option[T]
• type Processor

Constants

const (
	// DefaultBatchSize is the default batch size.
	DefaultBatchSize = 1000

	// DefaultBatchInterval is the default batch interval.
	DefaultBatchInterval = 1 * time.Second

	// DefaultConcurrency is the default concurrency.
	DefaultConcurrency = 3
)

Variables

var ErrTimeout = fmt.Errorf("timeout waiting for batches to complete")

Functions

func NoOpProcessor

func NoOpProcessor[T any]([]T) error

func ProvideBatcherInFX

func ProvideBatcherInFX[T any](
	processorFactory any,
	batchSize int,
	batchInterval time.Duration,
) fx.Option

Types

type AtomicCounter

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

func NewAtomicCounter

func NewAtomicCounter() *AtomicCounter

func (*AtomicCounter) Add

func (c *AtomicCounter) Add(num int64)

func (*AtomicCounter) Read

func (c *AtomicCounter) Read() int64

func (*AtomicCounter) Reset

func (c *AtomicCounter) Reset()

type Batcher

type Batcher[T any] struct {
	// contains filtered or unexported fields
}

func New

func New[T any](options ...Option[T]) *Batcher[T]

New creates a new Batcher with the given options.

func (*Batcher[T]) Add

func (b *Batcher[T]) Add(item T)

func (*Batcher[T]) Close

func (b *Batcher[T]) Close() error

func (*Batcher[T]) Config

func (b *Batcher[T]) Config() *Config[T]

func (*Batcher[T]) Errors

func (b *Batcher[T]) Errors() <-chan error

func (*Batcher[T]) IsClosed

func (b *Batcher[T]) IsClosed() bool

func (*Batcher[T]) Join

func (b *Batcher[T]) Join(timeout time.Duration) error

func (*Batcher[T]) Len

func (b *Batcher[T]) Len() int

func (*Batcher[T]) Start

func (b *Batcher[T]) Start()

type Config

type Config[T any] struct {
	SkipAutoStart bool
	BatchSize     int
	BatchInterval time.Duration
	Concurrency   int
	ProcessorFunc Processor[T]
}

type Option

type Option[T any] func(*Batcher[T])

func WithBatchInterval

func WithBatchInterval[T any](batchInterval time.Duration) Option[T]

WithBatchInterval sets the batch interval.

func WithBatchSize

func WithBatchSize[T any](batchSize int) Option[T]

WithBatchSize sets the batch size.

func WithProcessor

func WithProcessor[T any](fn Processor[T]) Option[T]

WithProcessor sets the processor function to be called for each batch.

func WithSkipAutoStart

func WithSkipAutoStart[T any]() Option[T]

WithSkipAutoStart skips the automatic start of the batcher.

type Processor

type Processor[T any] func([]T) error