dagprocessor

Documentation

Index

• Variables
• type Callback
• type Config
  • func DefaultConfig(scale cachescale.Func) Config
• type EventCallback
• type Processor
  • func New(eventsSemaphore *datasemaphore.DataSemaphore, cfg Config, callback Callback) *Processor
  • func (f *Processor) Clear()
  • func (f *Processor) Enqueue(peer string, events dag.Events, ordered bool, ...) error
  • func (f *Processor) IsBuffered(id hash.Event) bool
  • func (f *Processor) Overloaded() bool
  • func (f *Processor) Start()
  • func (f *Processor) Stop()
  • func (f *Processor) TasksCount() int
  • func (f *Processor) TotalBuffered() dag.Metric

Variables

var (
	ErrBusy = errors.New("failed to acquire events semaphore")
)

Types

type Callback

type Callback struct {
	Event          EventCallback
	HighestLamport func() idx.Lamport
}

type Config

type Config struct {
	EventsBufferLimit dag.Metric

	EventsSemaphoreTimeout time.Duration

	MaxTasks int
}

func DefaultConfig

func DefaultConfig(scale cachescale.Func) Config

type EventCallback

type EventCallback struct {
	Process         func(e dag.Event) error
	Released        func(e dag.Event, peer string, err error)
	Get             func(hash.Event) dag.Event
	Exists          func(hash.Event) bool
	CheckParents    func(e dag.Event, parents dag.Events) error
	CheckParentless func(e dag.Event, checked func(error))
}

type Processor

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

Processor is responsible for processing incoming events

func New

func New(eventsSemaphore *datasemaphore.DataSemaphore, cfg Config, callback Callback) *Processor

New creates an event processor

func (*Processor) Clear

func (f *Processor) Clear()

func (*Processor) Enqueue

func (f *Processor) Enqueue(peer string, events dag.Events, ordered bool, notifyAnnounces func(hash.Events), done func()) error

func (*Processor) IsBuffered

func (f *Processor) IsBuffered(id hash.Event) bool

func (*Processor) Overloaded

func (f *Processor) Overloaded() bool

Overloaded returns true if too much events are being processed or requested

func (*Processor) Start

func (f *Processor) Start()

Start boots up the events processor.

func (*Processor) Stop

func (f *Processor) Stop()

Stop interrupts the processor, canceling all the pending operations. Stop waits until all the internal goroutines have finished.

func (*Processor) TasksCount

func (f *Processor) TasksCount() int

func (*Processor) TotalBuffered

func (f *Processor) TotalBuffered() dag.Metric