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(parallel int, worker Worker, tasksBuffer 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
  • func (r Result) Value() <-chan interface{}
• type Task
  • func (t Task) SetResult(v interface{})
• type Worker

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{"agent_id"})
	DispatcherTasksFailed = prometheus.NewCounterVec(prometheus.CounterOpts{
		Name: "dispatcher_tasks_failed",
	}, []string{"agent_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(parallel int, worker Worker, tasksBuffer int) Dispatcher

Start spawns a pool of workers. tasksBuffer sets how many tasks should be pre-emptively put into each worker's incoming queue. Set to 0 for prevent greedy tasks assignment (this will make `Dispatch` blocking).

func (*Dispatcher) Dispatch

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

Dispatch takes `payload`, wraps it into a `Task` and dispatches to the first available `Worker`.

func (Dispatcher) Stop

func (d Dispatcher) Stop()

type Result

type Result struct {
	Status int
	Error  error
	// contains filtered or unexported fields
}

Result is a result of Task execution. TODO: setting/returning this needs to be implemented better using channels.

func (Result) Done

func (r Result) Done() bool

func (Result) Failed

func (r Result) Failed() bool

func (Result) Value

func (r Result) Value() <-chan interface{}

type Task

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

Task represents a unit of work. Each worker should accept it as an argument. Example:

 func (w encoderWorker) Work(t dispatcher.Task) error {
		r := t.Payload.(*resolve.TranscodingRequest)
 ...

func (Task) SetResult

func (t Task) SetResult(v interface{})

type Worker

type Worker interface {
	Work(Task) error
}

Worker can be any object that is capable to do `Work()`.