batch

Documentation

Index

• type DispatchHandler
• type DispatchState
• type DispatcherOptions
• type FlushStatus
• type Manager
  • func NewBatchManager(ctx context.Context, ns string, di database.Plugin, dm data.Manager, ...) (Manager, error)
• type ManagerStatus
• type ProcessorStatus

Types

type DispatchHandler

type DispatchHandler func(context.Context, *DispatchState) error

type DispatchState

type DispatchState struct {
	Persisted core.BatchPersisted
	Messages  []*core.Message
	Data      core.DataArray
	Pins      []*fftypes.Bytes32
	// contains filtered or unexported fields
}

type DispatcherOptions

type DispatcherOptions struct {
	BatchType      core.BatchType
	BatchMaxSize   uint
	BatchMaxBytes  int64
	BatchTimeout   time.Duration
	DisposeTimeout time.Duration
}

type FlushStatus

type FlushStatus struct {
	LastFlushTime        *fftypes.FFTime `ffstruct:"BatchFlushStatus" json:"lastFlushStartTime"`
	Flushing             *fftypes.UUID   `ffstruct:"BatchFlushStatus" json:"flushing,omitempty"`
	Blocked              bool            `ffstruct:"BatchFlushStatus" json:"blocked"`
	LastFlushError       string          `ffstruct:"BatchFlushStatus" json:"lastFlushError,omitempty"`
	LastFlushErrorTime   *fftypes.FFTime `ffstruct:"BatchFlushStatus" json:"lastFlushErrorTime,omitempty"`
	AverageBatchBytes    int64           `ffstruct:"BatchFlushStatus" json:"averageBatchBytes"`
	AverageBatchMessages float64         `ffstruct:"BatchFlushStatus" json:"averageBatchMessages"`
	AverageBatchData     float64         `ffstruct:"BatchFlushStatus" json:"averageBatchData"`
	AverageFlushTimeMS   int64           `ffstruct:"BatchFlushStatus" json:"averageFlushTimeMS"`
	TotalBatches         int64           `ffstruct:"BatchFlushStatus" json:"totalBatches"`
	TotalErrors          int64           `ffstruct:"BatchFlushStatus" json:"totalErrors"`
	// contains filtered or unexported fields
}

FlushStatus is an object that can be returned on REST queries to understand the status of the batch processor

type Manager

type Manager interface {
	RegisterDispatcher(name string, txType core.TransactionType, msgTypes []core.MessageType, handler DispatchHandler, batchOptions DispatcherOptions)
	NewMessages() chan<- int64
	Start() error
	Close()
	WaitStop()
	Status() *ManagerStatus
}

func NewBatchManager

func NewBatchManager(ctx context.Context, ns string, di database.Plugin, dm data.Manager, im identity.Manager, txHelper txcommon.Helper) (Manager, error)

type ManagerStatus

type ManagerStatus struct {
	Processors []*ProcessorStatus `ffstruct:"BatchManagerStatus" json:"processors"`
}

type ProcessorStatus

type ProcessorStatus struct {
	Dispatcher string      `ffstruct:"BatchProcessorStatus" json:"dispatcher"`
	Name       string      `ffstruct:"BatchProcessorStatus" json:"name"`
	Status     FlushStatus `ffstruct:"BatchProcessorStatus" json:"status"`
}