aggregator

Documentation

Index

• type Aggregator
  • func (aggregator *Aggregator) AddHashItem(item types.HashItem)
  • func (aggregator *Aggregator) AggregateAndReset() []types.Aggregation
  • func (aggregator *Aggregator) HeadHashItem() types.HashItem
  • func (aggregator *Aggregator) ProcessAggregation(msgStructSlice []types.HashItem, drand string) types.Aggregation
  • func (aggregator *Aggregator) StartAggregation() error

Types

type Aggregator

type Aggregator struct {
	HashItems    goconcurrentqueue.Queue //In
	Logger       log.Logger
	LatestTime   string
	Aggregations goconcurrentqueue.Queue //Out
	AggMutex     sync.Mutex
	RestartMutex sync.Mutex
	QueueMutex   sync.Mutex
	TempStop     chan struct{}
	WaitGroup    sync.WaitGroup
}

Aggregator : object includes rabbitURI and Logger

func (*Aggregator) AddHashItem

func (aggregator *Aggregator) AddHashItem(item types.HashItem)

func (*Aggregator) AggregateAndReset

func (aggregator *Aggregator) AggregateAndReset() []types.Aggregation

func (*Aggregator) HeadHashItem

func (aggregator *Aggregator) HeadHashItem() types.HashItem

func (*Aggregator) ProcessAggregation

func (aggregator *Aggregator) ProcessAggregation(msgStructSlice []types.HashItem, drand string) types.Aggregation

ProcessAggregation creates merkle trees of received hashes a la https://github.com/chainpoint/chainpoint-services/blob/develop/node-aggregator-service/server.js#L66

func (*Aggregator) StartAggregation

func (aggregator *Aggregator) StartAggregation() error

ReceiveCalRMQ : Continually consume the calendar work queue and process any resulting messages from the tx and monitor services