sm

package

v0.8.1 Latest Latest Go to latest Published: Oct 9, 2023 License: Apache-2.0 Imports: 5 Imported by: 0

Details

View Source

const (
	Created int32 = iota
	Running
	ReceiverStopped
	PrepareStop
	Stopped
)

View Source

var (
	ErrClose = moerr.NewInternalErrorNoCtx("closed")
)

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

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

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

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

func (c *ClosedState) IsClosed() bool

func (c *ClosedState) TryClose() bool

type EnqueueOp = func(any) any

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

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

func (l *Loop) Start()

func (l *Loop) Stop()

type OnFinCB = func()

type OnItemsCB = func(...any)

type Queue interface {
	Start()
	Stop()
	Enqueue(any) (any, error)
}

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