sm

Documentation

Index

• Constants
• Variables
• func NewNonBlockingQueue(queueSize int, batchSize int, onItem OnItemsCB) *safeQueue
• func NewSafeQueue(queueSize, batchSize int, onItem OnItemsCB) *safeQueue
• func NewStateMachine(wg *sync.WaitGroup, closed Closable, rQueue, ckpQueue Queue) *stateMachine
• type Closable
• type ClosedState
  • func (c *ClosedState) IsClosed() bool
  • func (c *ClosedState) TryClose() bool
• type EnqueueOp
• type Loop
  • func NewLoop(queue, nextQueue chan any, fn func([]any, chan any), maxBatchSize int) *Loop
  • func (l *Loop) Start()
  • func (l *Loop) Stop()
• type OnFinCB
• type OnItemsCB
• type Queue
• type StateMachine

Constants

const (
	Created int32 = iota
	Running
	ReceiverStopped
	PrepareStop
	Stopped
)

Variables

var (
	ErrClose = moerr.NewInternalErrorNoCtx("closed")
	ErrFull  = moerr.NewInternalErrorNoCtx("nonblocking queue is full, dropped item")
)

Functions

func NewNonBlockingQueue

func NewNonBlockingQueue(queueSize int, batchSize int, onItem OnItemsCB) *safeQueue

func NewSafeQueue

func NewSafeQueue(queueSize, batchSize int, onItem OnItemsCB) *safeQueue

NewSafeQueue is blocking queue by default

func NewStateMachine

func NewStateMachine(wg *sync.WaitGroup, closed Closable, rQueue, ckpQueue Queue) *stateMachine

Types

type Closable

type Closable interface {
	IsClosed() bool
	TryClose() bool
}

type ClosedState

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

func (*ClosedState) IsClosed

func (c *ClosedState) IsClosed() bool

func (*ClosedState) TryClose

func (c *ClosedState) TryClose() bool

type EnqueueOp

type EnqueueOp = func(any) any

type Loop

type Loop struct {
	Itemcount int
	Itemtimes int
	// contains filtered or unexported fields
}

func NewLoop

func NewLoop(queue, nextQueue chan any, fn func([]any, chan any), maxBatchSize int) *Loop

func (*Loop) Start

func (l *Loop) Start()

func (*Loop) Stop

func (l *Loop) Stop()

type OnFinCB

type OnFinCB = func()

type OnItemsCB

type OnItemsCB = func(...any)

type Queue

type Queue interface {
	Start()
	Stop()

	// Enqueue puts an item into this queue
	// it will return directly when if it is an unblocking queue and there has no more free space,
	// and ErrFull will be return to notify the producer.
	Enqueue(any) (any, error)
}

type StateMachine

type StateMachine interface {
	Start()
	Stop()
	EnqueueRecevied(any) (any, error)
	EnqueueCheckpoint(any) (any, error)
}