queue

README

Go-Queue

A simple generic chan based queue worker

License

Please refer to LICENSE.md

Examples

Simple example

package main 

import (
	queue "github.com/tcfw/go-queue"
)

type Processor struct {}

func (p *Processor) Handle(job interface{}) {
	//Handle job...
}

func main() {
	processor := &Processor{}

	dispatcher := queue.NewDispatcher(processor)
	dispatcher.Run()
}

Specify number of workers

package main 

import (
	queue "github.com/tcfw/go-queue"
)

type Processor struct {}

func (p *Processor) Handle(job interface{}) {
	//Handle job...
}

func main() {
	processor := &Processor{}

	dispatcher := queue.NewDispatcher(processor)
	//20 workers will be created 
	dispatcher.MaxWorkers = 20
	dispatcher.Run()
}

Documentation

Index

• type Dispatcher
  • func NewDispatcher(processor WorkProcessor) *Dispatcher
  • func (d *Dispatcher) Queue(job interface{})
  • func (d *Dispatcher) QueueAfter(job interface{}, t time.Duration)
  • func (d *Dispatcher) Run()
  • func (d *Dispatcher) Stop()
• type WorkProcessor
• type Worker
  • func NewWorker(workerPool chan chan interface{}) Worker
  • func (w Worker) Ready()
  • func (w Worker) Start()
  • func (w Worker) Stop()

Types

type Dispatcher

type Dispatcher struct {
	WorkerPool chan chan interface{}
	MaxWorkers int
	Processor  WorkProcessor
	Workers    []Worker
	// contains filtered or unexported fields
}

Dispatcher holds the worker pool and delegates jobs to those workers

func NewDispatcher

func NewDispatcher(processor WorkProcessor) *Dispatcher

NewDispatcher creates a new dispatcher with respective processor

func (*Dispatcher) Queue

func (d *Dispatcher) Queue(job interface{})

Queue sends a job to be processed

func (*Dispatcher) QueueAfter

func (d *Dispatcher) QueueAfter(job interface{}, t time.Duration)

QueueAfter sends a job to be processed after x amount of time

func (*Dispatcher) Run

func (d *Dispatcher) Run()

Run registers the workers and allocation of jobs

func (*Dispatcher) Stop

func (d *Dispatcher) Stop()

Stop signals all queue workers to quit

type WorkProcessor

type WorkProcessor interface {
	Handle(job interface{})
}

WorkProcessor should be able to process jobs that come in

type Worker

type Worker struct {
	WorkerPool chan chan interface{}
	JobChannel chan interface{}

	Processor WorkProcessor
	// contains filtered or unexported fields
}

Worker proccesses jobs via the processor interface

func NewWorker

func NewWorker(workerPool chan chan interface{}) Worker

NewWorker bootstraps the job and quit channels

func (Worker) Ready

func (w Worker) Ready()

Ready attaches the worker to the pool

func (Worker) Start

func (w Worker) Start()

Start beings processing jobs from the worker's job queue

func (Worker) Stop

func (w Worker) Stop()

Stop signals the worker to stop listening for work requests.