dispatcher

Documentation

Index

• Constants
• Variables
• func RegisterMetrics()
• func SetLogger(l *zap.SugaredLogger)
• func WaitUntilTrue(ctx context.Context, between time.Duration, f func() bool) error
• type Dispatcher
  • func Start(workers int, wl Workload, tasksLen int) Dispatcher
  • func (d *Dispatcher) Dispatch(payload interface{}) *Result
  • func (d Dispatcher) Stop()
• type Result
  • func (r Result) Done() bool
  • func (r Result) Failed() bool
• type Task
• type Worker
  • func NewWorker(id int, workerPool chan chan Task, wl Workload, gwait *sync.WaitGroup) Worker
  • func (w *Worker) Start()
  • func (w *Worker) Stop()
• type Workload

Constants

const (
	TaskFailed = iota
	TaskDone
	TaskActive
	TaskPending
	TaskDropped
)

Variables

var (
	DispatcherQueueLength = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "dispatcher_queue_length",
	})
	DispatcherTasksActive = prometheus.NewGauge(prometheus.GaugeOpts{
		Name: "dispatcher_tasks_active",
	})
	DispatcherTasksQueued = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "dispatcher_tasks_queued",
	})
	DispatcherTasksDropped = prometheus.NewCounter(prometheus.CounterOpts{
		Name: "dispatcher_tasks_dropped",
	})
	DispatcherTasksDone = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "dispatcher_tasks_done",
	}, []string{"worker_id"})
	DispatcherTasksFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "dispatcher_tasks_failed",
	}, []string{"worker_id"})
)

var ErrInvalidPayload = errors.New("invalid payload")

Functions

func RegisterMetrics

func RegisterMetrics()

func SetLogger

func SetLogger(l *zap.SugaredLogger)

func WaitUntilTrue

func WaitUntilTrue(ctx context.Context, between time.Duration, f func() bool) error

Types

type Dispatcher

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

func Start

func Start(workers int, wl Workload, tasksLen int) Dispatcher

func (*Dispatcher) Dispatch

func (d *Dispatcher) Dispatch(payload interface{}) *Result

func (Dispatcher) Stop

func (d Dispatcher) Stop()

type Result

type Result struct {
	Status int
	Error  error
}

func (Result) Done

func (r Result) Done() bool

func (Result) Failed

func (r Result) Failed() bool

type Task

type Task struct {
	Payload    interface{}
	Dispatcher *Dispatcher
	// contains filtered or unexported fields
}

type Worker

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

func NewWorker

func NewWorker(id int, workerPool chan chan Task, wl Workload, gwait *sync.WaitGroup) Worker

func (*Worker) Start

func (w *Worker) Start()

Start starts reading from tasks channel

func (*Worker) Stop

func (w *Worker) Stop()

Stop stops the wl invocation cycle (it will finish the current wl).

type Workload

type Workload interface {
	Do(Task) error
}