jobrunner

Documentation

Index

• Constants
• type Handler
• type Helper
  • func HelperFor(ctx context.Context) Helper
• type JobRunner
  • func NewJobRunner(mgr manager.Manager, opts Options) *JobRunner
  • func (jr *JobRunner) Push(ctx context.Context, job *client.Job) error
  • func (jr *JobRunner) Register(jobtype string, fn sparq.PerformFunc)
  • func (jr *JobRunner) Run(ctx context.Context) error
  • func (jr *JobRunner) Shutdown(ctx context.Context) error
• type LifecycleEventHandler
• type MiddlewareFunc
• type NoHandlerError
  • func (s *NoHandlerError) Error() string
• type Options
• type Runner
  • func NewRunner(mgr faktory.Manager) *Runner
  • func (mgr *Runner) On(event lifecycleEventType, fn LifecycleEventHandler)
  • func (mgr *Runner) Quiet()
  • func (mgr *Runner) Register(name string, fn sparq.PerformFunc)
  • func (mgr *Runner) Run(ctx context.Context) error
  • func (mgr *Runner) Terminate(shutdownCtx context.Context)
  • func (mgr *Runner) Use(middleware ...MiddlewareFunc)

Constants

const (
	Startup  lifecycleEventType = 1
	Quiet    lifecycleEventType = 2
	Shutdown lifecycleEventType = 3
)

Types

type Handler

type Handler func(ctx context.Context, job *client.Job) error

type Helper

type Helper interface {
	Jid() string
	JobType() string

	// Custom provides access to the job custom hash.
	// Returns the value and `ok=true` if the key was found.
	// If not, returns `nil` and `ok=false`.
	//
	// No type checking is performed, please use with caution.
	Custom(key string) (value interface{}, ok bool)

	// allows direct access to the Faktory server from the job
	With(func(sparq.Pusher) error) error
}

The Helper provides access to valuable data and APIs within an executing job.

We're pretty strict about what's exposed in the Helper because execution should be orthogonal to most of the Job payload contents.

	func myJob(ctx context.Context, args ...interface{}) error {
	  helper := worker.HelperFor(ctx)
	  jid := helper.Jid()

	  helper.With(func(mgr faktory.Manager) error {
      job := client.NewJob("JobType", 1, 2, 3)
	    mgr.Push(job)
		})

func HelperFor

func HelperFor(ctx context.Context) Helper

Caution: this method must only be called within the context of an executing job. It will panic if it cannot create a Helper due to missing context values.

type JobRunner

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

func NewJobRunner

func NewJobRunner(mgr manager.Manager, opts Options) *JobRunner

func (*JobRunner) Push

func (jr *JobRunner) Push(ctx context.Context, job *client.Job) error

func (*JobRunner) Register

func (jr *JobRunner) Register(jobtype string, fn sparq.PerformFunc)

func (*JobRunner) Run

func (jr *JobRunner) Run(ctx context.Context) error

func (*JobRunner) Shutdown

func (jr *JobRunner) Shutdown(ctx context.Context) error

type LifecycleEventHandler

type LifecycleEventHandler func(*Runner) error

type MiddlewareFunc

type MiddlewareFunc func(ctx context.Context, job *client.Job, next func(ctx context.Context) error) error

type NoHandlerError

type NoHandlerError struct {
	JobType string
}

func (*NoHandlerError) Error

func (s *NoHandlerError) Error() string

type Options

type Options struct {
	Concurrency int
	Queues      []string
}

type Runner

type Runner struct {
	Concurrency int
	Labels      []string
	Queues      []string
	// contains filtered or unexported fields
}

Runner coordinates the processes for the worker. It is responsible for starting and stopping goroutines to perform work at the desired concurrency level

func NewRunner

func NewRunner(mgr faktory.Manager) *Runner

NewManager returns a new manager with default values.

func (*Runner) On

func (mgr *Runner) On(event lifecycleEventType, fn LifecycleEventHandler)

Register a callback to be fired when a process lifecycle event occurs. These are useful for hooking into process startup or shutdown.

func (*Runner) Quiet

func (mgr *Runner) Quiet()

After calling Quiet(), no more jobs will be pulled from Faktory by this process.

func (*Runner) Register

func (mgr *Runner) Register(name string, fn sparq.PerformFunc)

Register a handler for the given jobtype. It is expected that all jobtypes are registered upon process startup.

mgr.Register("ImportantJob", ImportantFunc)

func (*Runner) Run

func (mgr *Runner) Run(ctx context.Context) error

RunWithContext starts processing jobs. The method will return if an error is encountered while starting. If the context is present then os signals will be ignored, the context must be canceled for the method to return after running.

func (*Runner) Terminate

func (mgr *Runner) Terminate(shutdownCtx context.Context)

Terminate signals that the various components should shutdown. Blocks on the shutdownWaiter until all components have finished.

func (*Runner) Use

func (mgr *Runner) Use(middleware ...MiddlewareFunc)

Use(...) adds middleware to the chain.