async

Documentation

Index

• Constants
• type Daemon
  • func NewDaemon(name string, runnable Runnable) Daemon
• type Job
• type JobFunc
  • func (j JobFunc) Run(ctx context.Context)
• type Pool
  • func NewPool(o PoolOptions, queue Queue) *Pool
  • func (p *Pool) Enqueue(job Job)
  • func (p *Pool) Start()
  • func (p *Pool) Stop()
  • func (p *Pool) WaitUntilProcessed()
• type PoolOptions
• type Queue
• type Runnable
  • func NewRunnable(runFunc func(context.Context) error) Runnable

Constants

const (
	// DefaultMaxWorkers of a Pool. See Pool.SetMaxWorkers for more info.
	DefaultMaxWorkers = 4
)

Types

type Daemon

type Daemon interface {
	// Start starts the daemon. The daemon is running when the underlying
	// runnable is started. Start blocks until the runnable is in the running
	// state. Otherwise it returns and does not block.
	Start()

	// Stop stops the daemon. The daemon is running until the underlying
	// runnable returns. Stop blocks until the runnable is in state stopped.
	// Otherwise it returns and does not block.
	Stop()
}

Daemon represents a function that we want to start and run continuously until stopped.

func NewDaemon

func NewDaemon(name string, runnable Runnable) Daemon

NewDaemon will create a new daemon.

type Job

type Job interface {
	// Run the Job, provided the given context.
	// TODO: Error result?
	Run(ctx context.Context)
}

Job to be inserted to a Pool or Queue.

type JobFunc

type JobFunc func(context.Context)

JobFunc is an convenience type for easily converting function literals to a Job compatible object.

func (JobFunc) Run

func (j JobFunc) Run(ctx context.Context)

Run the JobFunc by invoking itself.

type Pool

type Pool struct {
	sync.Mutex
	// contains filtered or unexported fields
}

Pool structure for running up to a maximum number of jobs concurrently. The pool as an internal queue, such that all jobs added will be accepted but not run until it reached the front of the queue and a worker is free.

func NewPool

func NewPool(o PoolOptions, queue Queue) *Pool

NewPool returns a new pool, provided the PoolOptions and the queue.

func (*Pool) Enqueue

func (p *Pool) Enqueue(job Job)

Enqueue a job in the pool. TODO: Take an context argument that will be associated to the job. That way deadlines can easily be propagated.

func (*Pool) Start

func (p *Pool) Start()

Start the worker pool by initializing the stop channel and starting all the workers

func (*Pool) Stop

func (p *Pool) Stop()

Stop sets the assigned workers (goal state) to zero, and then stopWorkers terminates running workers (actual state) to 0 value amd finally cleans up the stop channel

func (*Pool) WaitUntilProcessed

func (p *Pool) WaitUntilProcessed()

WaitUntilProcessed will block until both the queue is empty and all workers are idle. This is useful for per-request Pools and in testing.

type PoolOptions

type PoolOptions struct {
	MaxWorkers int
}

PoolOptions for constructing a new Pool.

type Queue

type Queue interface {
	// Run runs the Queue and will stop the Queue if the stopChan provided
	// is closed
	Run(stopChan chan struct{})
	// Enqueue is used to enqueue a job
	Enqueue(job Job)
	// Dequeue is used to fetch an enqueued job when a worker is available
	Dequeue() Job
}

Queue defines the interface of a queue used by the async pool to enqueue jobs and then dequeue the job when a worker becomes available

type Runnable

type Runnable interface {
	// Run will run the runnable with a context and return any errors that
	// might occur.
	Run(ctx context.Context) (err error)
}

Runnable represents a runnable function that can return an error.

func NewRunnable

func NewRunnable(runFunc func(context.Context) error) Runnable

NewRunnable creates a new runnable from a function type.