worker

Documentation

Index

• type Pool
  • func CreateWorkerPool(size int, eventsChannel <-chan collector.CollectRequest, producer Producer) *Pool
  • func (w *Pool) FlushWithTimeOut(timeout time.Duration) bool
  • func (w *Pool) StartWorkers()
• type Producer

Types

type Pool

type Pool struct {
	Size          int
	EventsChannel <-chan collector.CollectRequest
	// contains filtered or unexported fields
}

Pool spawn goroutine as much as Size that will listen to EventsChannel. On Close, wait for all data in EventsChannel to be processed.

func CreateWorkerPool

func CreateWorkerPool(size int, eventsChannel <-chan collector.CollectRequest, producer Producer) *Pool

CreateWorkerPool create new Pool struct given size and EventsChannel worker.

func (*Pool) FlushWithTimeOut

func (w *Pool) FlushWithTimeOut(timeout time.Duration) bool

FlushWithTimeOut waits for the workers to complete the pending the messages to be flushed to the publisher within a timeout. Returns true if waiting timed out, meaning not all the events could be processed before this timeout.

func (*Pool) StartWorkers

func (w *Pool) StartWorkers()

StartWorkers initialize worker pool as much as Pool.Size

type Producer

type Producer interface {
	// ProduceBulk message to a sink. Blocks until all messages are sent. Returns slice of error.
	ProduceBulk(events []*pb.Event, connGroup string) error

	// Name returns the name of the producer
	Name() string
}

Producer produces data to sink