runners

README

go-runners

Provides a means a simple means of managing long-running functions and services

Documentation

Index

• func Closed() context.Context
• func CtxWithCancel() (context.Context, context.CancelFunc)
• type CancelCtx
  • func (CancelCtx) Deadline() (time.Time, bool)
  • func (ctx CancelCtx) Done() <-chan struct{}
  • func (ctx CancelCtx) Err() error
  • func (ctx CancelCtx) GoString() string
  • func (ctx CancelCtx) String() string
  • func (CancelCtx) Value(key interface{}) interface{}
• type Processable
• type Processor
  • func (p *Processor) Process(proc Processable) (err error)
• type Service
  • func (svc *Service) Done() <-chan struct{}
  • func (svc *Service) GoRun(fn func(context.Context)) bool
  • func (svc *Service) Run(fn func(context.Context)) bool
  • func (svc *Service) RunWait(fn func(context.Context)) bool
  • func (svc *Service) Running() bool
  • func (svc *Service) Stop() bool
  • func (svc *Service) While(fn func())
• type WorkerFunc
• type WorkerPool
  • func (pool *WorkerPool) Done() <-chan struct{}
  • func (pool *WorkerPool) Enqueue(fn WorkerFunc)
  • func (pool *WorkerPool) EnqueueCtx(ctx context.Context, fn WorkerFunc) bool
  • func (pool *WorkerPool) EnqueueNow(fn WorkerFunc) bool
  • func (pool *WorkerPool) MustEnqueueCtx(ctx context.Context, fn WorkerFunc) (ok bool)
  • func (pool *WorkerPool) Queue() int
  • func (pool *WorkerPool) Running() bool
  • func (pool *WorkerPool) Start(workers int, queue int) bool
  • func (pool *WorkerPool) Stop() bool

Functions

func Closed

func Closed() context.Context

Closed returns an always closed context.

func CtxWithCancel

func CtxWithCancel() (context.Context, context.CancelFunc)

CtxWithCancel returns a new context.Context impl with cancel.

Types

type CancelCtx

type CancelCtx (<-chan struct{})

CancelCtx is the simplest possible cancellable context.

func (CancelCtx) Deadline

func (CancelCtx) Deadline() (time.Time, bool)

func (CancelCtx) Done

func (ctx CancelCtx) Done() <-chan struct{}

func (CancelCtx) Err

func (ctx CancelCtx) Err() error

func (CancelCtx) GoString

func (ctx CancelCtx) GoString() string

func (CancelCtx) String

func (ctx CancelCtx) String() string

func (CancelCtx) Value

func (CancelCtx) Value(key interface{}) interface{}

type Processable

type Processable func() error

Processable defines a runnable process with error return that can be passed to a Processor instance for managed running.

type Processor

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

Processor acts similarly to a sync.Once object, except that it is reusable. After the first call to Process(), any further calls before this first has returned will block until the first call has returned, and return the same error. This ensures that only a single instance of it is ever running at any one time.

func (*Processor) Process

func (p *Processor) Process(proc Processable) (err error)

Process will process the given function if first-call, else blocking until the first function has returned, returning the same error result.

type Service

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

Service provides a means of tracking a single long-running service, provided protected state changes and preventing multiple instances running. Also providing service state information.

func (*Service) Done

func (svc *Service) Done() <-chan struct{}

Done returns a channel that's closed when Service.Stop() is called. It is the same channel provided to the currently running service function.

func (*Service) GoRun

func (svc *Service) GoRun(fn func(context.Context)) bool

GoRun will run the supplied function until completion in a goroutine, using given context to propagate cancel. Immediately returns boolean indicating success, or that service is already running.

func (*Service) Run

func (svc *Service) Run(fn func(context.Context)) bool

Run will run the supplied function until completion, using given context to propagate cancel. Immediately returns false if the Service is already running, and true after completed run.

func (*Service) RunWait

func (svc *Service) RunWait(fn func(context.Context)) bool

RunWait is functionally the same as .Run(), but blocks until the first instance of .Run() returns.

func (*Service) Running

func (svc *Service) Running() bool

Running returns if Service is running (i.e. state NOT stopped / stopping).

func (*Service) Stop

func (svc *Service) Stop() bool

Stop will attempt to stop the service, cancelling the running function's context. Immediately returns false if not running, and true only after Service is fully stopped.

func (*Service) While

func (svc *Service) While(fn func())

While allows you to execute given function guaranteed within current service state. Please note that this will hold the underlying service state change mutex open while executing the function.

type WorkerFunc

type WorkerFunc func(context.Context)

WorkerFunc represents a function processable by a worker in WorkerPool. Note that implementations absolutely MUST check whether passed context is <-ctx.Done() otherwise stopping the pool may block indefinitely.

type WorkerPool

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

WorkerPool provides a means of enqueuing asynchronous work.

func (*WorkerPool) Done

func (pool *WorkerPool) Done() <-chan struct{}

Done returns a channel that's closed when WorkerPool.Stop() is called. It is the same channel provided to the currently running worker functions.

func (*WorkerPool) Enqueue

func (pool *WorkerPool) Enqueue(fn WorkerFunc)

Enqueue will add provided WorkerFunc to the queue to be performed when there is a free worker. This will block until function is queued or pool is stopped. In all cases, the WorkerFunc will be executed, with the state of the pool being indicated by <-ctx.Done() of the passed ctx. WorkerFuncs MUST respect the passed context.

func (*WorkerPool) EnqueueCtx

func (pool *WorkerPool) EnqueueCtx(ctx context.Context, fn WorkerFunc) bool

EnqueueCtx is functionally identical to WorkerPool.Enqueue() but returns early in the case that caller provided <-ctx.Done() is closed, WITHOUT running the WorkerFunc.

func (*WorkerPool) EnqueueNow

func (pool *WorkerPool) EnqueueNow(fn WorkerFunc) bool

EnqueueNow attempts Enqueue but returns false if not executed.

func (*WorkerPool) MustEnqueueCtx

func (pool *WorkerPool) MustEnqueueCtx(ctx context.Context, fn WorkerFunc) (ok bool)

MustEnqueueCtx functionally performs similarly to WorkerPool.EnqueueCtx(), but in the case that the provided <-ctx.Done() is closed, it is passed asynchronously to WorkerPool.Enqueue(). Return boolean indicates whether function was executed in time before <-ctx.Done() is closed.

func (*WorkerPool) Queue

func (pool *WorkerPool) Queue() int

Queue returns the number of currently queued WorkerFuncs.

func (*WorkerPool) Running

func (pool *WorkerPool) Running() bool

Running returns if WorkerPool management loop is running (i.e. NOT stopped / stopping).

func (*WorkerPool) Start

func (pool *WorkerPool) Start(workers int, queue int) bool

Start will start the main WorkerPool management loop in a new goroutine, along with requested number of child worker goroutines. Returns false if already running.

func (*WorkerPool) Stop

func (pool *WorkerPool) Stop() bool

Stop will stop the WorkerPool management loop, blocking until stopped.