job

package
v1.2.3 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 8, 2025 License: Apache-2.0 Imports: 24 Imported by: 0

Documentation

Overview

go-job is a flexible and extensible job scheduling and execution library for Go. It enables you to register, schedule, and manage jobs with arbitrary function signatures, supporting custom executors, priorities, and advanced scheduling options such as cron expressions and delayed execution.

The library provides robust job observation features, including state and log history tracking, as well as customizable response and error handlers. With support for distributed storage backends, `go-job` is suitable for both local and distributed environments, making it ideal for building scalable, reliable job processing systems in Go applications.

Example 
// Create a job manager
mgr, _ := NewManager()

// Register a job with a custom executor
sumJob, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithScheduleAt(time.Now()), // immediate scheduling is the default, so this option is redundant
	WithCompleteProcessor(func(ji Instance, res []any) {
		ji.Infof("Result: %v", res)
	}),
	WithTerminateProcessor(func(ji Instance, err error) error {
		ji.Errorf("Error executing job: %v", err)
		return err
	}),
)
mgr.RegisterJob(sumJob)

// Schedule the registered job
ji, _ := mgr.ScheduleRegisteredJob("sum", WithArguments(1, 2))

// Start the job manager
mgr.Start()

// Wait waits for all jobs to complete or terminate.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
mgr.Wait(ctx)

// Retrieve and print the job instance state history
query := NewQuery(
	WithQueryInstance(ji), // filter by specific job instance
)
history, _ := mgr.LookupInstanceHistory(query)
for _, record := range history {
	fmt.Println(record.State())
}

// Retrieve and print the job instance logs
query = NewQuery(
	WithQueryInstance(ji), // filter by specific job instance
)
logs, _ := mgr.LookupInstanceLogs(query)
for _, log := range logs {
	fmt.Println(log.Message())
}

// Stop the job manager
mgr.Stop()

Output:
Created
Scheduled
Processing
Completed
Result: [3]

Index

Examples

Constants

View Source 
const (
	// DefaultBindAddr is the default address of the gRPC and Prometheus server.
	DefaultBindAddr = ""
	// DefaultGRPCPort is the default gRPC port for the job server.
	DefaultGRPCPort = 59051
	// DefaultPrometheusPort is the default Prometheus port for the job server.
	DefaultPrometheusPort = 9090
	// DefaultAPIVersion is the default API version.
	DefaultAPIVersion = "v1"
)
View Source 
const (
	// NoRetry is a constant indicating that a job should not be retried.
	NoRetry = 0
	// RetryForever is a constant indicating that a job should retry indefinitely.
	RetryForever = -1
)
View Source 
const (
	// NoTimeout indicates no timeout limit.
	NoTimeout = 0
	// DefaultTimeout is the default timeout for jobs.
	DefaultTimeout = 0
)
View Source 
const (
	// HighPriority is the high priority for jobs.
	HighPriority = Priority(0)
	// MediumPriority is the medium priority for jobs.
	MediumPriority = Priority(5)
	// DefaultPriority is the default priority for jobs.
	DefaultPriority = MediumPriority
	// LowPriority is the low priority for jobs.
	LowPriority = Priority(10)
)
View Source 
const (
	// JobStateInitial represents the initial state of a job (created but not yet scheduled).
	JobStateInitial = JobCreated
	// JobStateActive represents the active states of a job (scheduled or processing).
	JobStateActive = JobScheduled | JobProcessing
	// JobStateFinal represents the final states of a job (canceled, timed out, completed, or terminated).
	JobStateFinal = JobCanceled | JobTimedOut | JobCompleted | JobTerminated
	// JobStateError represents the error states of a job (canceled, timed out, or terminated).
	JobStateError = JobCanceled | JobTimedOut | JobTerminated
	// JobStateSuccess represents the successful completion of a job.
	JobStateSuccess = JobCompleted
	// JobStateAll represents all possible states of a job.
	JobStateAll = JobCreated | JobScheduled | JobProcessing | JobCanceled | JobTimedOut | JobCompleted | JobTerminated
)
View Source 
const (
	// DefaultWorkerNum is the default number of workers in the group.
	DefaultWorkerNum = 1
)
View Source 
const (
	// ProductName is the product name.
	ProductName = "go-job"
)
View Source 
const (
	Version = "v1.2.3"
)

Variables

View Source 
var ErrExists = errors.New("exists")

ErrExists is an exists error.

View Source 
var ErrInvalid = errors.New("invalid")

ErrInvalid is an invalid error.

View Source 
var ErrNil = errors.New("nil")

ErrNil is a nil error.

View Source 
var ErrNotFound = errors.New("not found")

ErrNotFound is an not found error.

View Source 
var ErrNotProcessing = errors.New("not processing")

ErrNotProcessing is a not processing error.

View Source 
var Placeholder string = "?"

Placeholder is a placeholder for the job arguments.

Functions

This section is empty.

Types

type Arguments 

type Arguments interface {
	// Arguments returns the underlying arguments.
	Arguments() []any
	// Map returns the arguments as a map.
	Map() map[string]any
	// String returns a string representation of the arguments.
	String() string
}

type ArgumentsOption 

type ArgumentsOption func(*argumentsImpl)

ArgumentsOption defines a function that configures the arguments for a job.

func WithArguments 

func WithArguments(args ...any) ArgumentsOption

WithArguments sets the arguments for a job.

type BackoffStrategy added in v0.9.1 

type BackoffStrategy func(ji Instance) time.Duration

BackoffStrategy is a function type that defines how long to wait before retrying a job.

type CLIClient added in v1.0.0 

type CLIClient interface {
	Client
	// SetCommandExecutor sets the command executor.
	SetCommandExecutor(executor CommandExecutor)
}

func NewCliClient added in v1.0.0 

func NewCliClient(args ...string) CLIClient

NewCliClient returns a new cli client.

type Client added in v1.0.0 

type Client interface {
	// Name returns the name of the client.
	Name() string
	// SetHost sets a host name.
	SetHost(host string)
	// SetPort sets a port number.
	SetPort(port int)
	// Open opens a connection.
	Open() error
	// Close closes the connection.
	Close() error
	// GetVersion retrieves the version of the service.
	GetVersion() (string, error)
	// ScheduleJob schedules a job with the given kind and arguments.
	ScheduleJob(kind string, args ...any) (Instance, error)
	// ListRegisteredJobs lists all registered jobs.
	ListRegisteredJobs() ([]Job, error)
	// LookupInstances looks up job instances based on the provided query.
	LookupInstances(query Query) ([]Instance, error)
	// CancelInstances cancels job instances based on the provided query.
	CancelInstances(query Query) ([]Instance, error)
}

Client represents a gRPC client.

func NewClient added in v1.0.0 

func NewClient() Client

NewClient returns a new default gRPC client.

func NewGrpcClient added in v1.0.0 

func NewGrpcClient() Client

NewClient returns a new gRPC client.

type CommandExecutor added in v1.0.0 

type CommandExecutor func(name string, args ...string) ([]byte, error)

CommandExecutor defines a function type for executing commands.

type CompleteProcessor added in v1.0.0 

type CompleteProcessor func(job Instance, responses []any)

CompleteProcessor is called when a job reaches the completed state (successful completion). It allows users to handle the results of the job execution. Users can process the results and perform any necessary actions.

type Config added in v1.2.0 

type Config interface {
	// SetGRPCPort sets the gRPC port for the job server.
	SetGRPCPort(port int)
	// GRPCPort returns the gRPC port for the job server.
	GRPCPort() int
	// SetPrometheusPort sets the Prometheus port for the job server.
	SetPrometheusPort(port int)
	// PrometheusPort returns the Prometheus port for the job server.
	PrometheusPort() int
}

Config is the interface for the job server configuration.

type Executor 

type Executor any

Executor is a type that represents a function that executes a job. It can be any function type, allowing for flexible job execution.

type Filter added in v1.0.0 

type Filter interface {
	// Before returns the time before which job instances should be filtered.
	Before() (time.Time, bool)
	// After returns the time after which job instances should be filtered.
	After() (time.Time, bool)
	// IsUnset returns true if no filter criteria are configured.
	IsUnset() bool
	// Matches checks if the specified object matches the filter criteria.
	Matches(v any) bool
}

Filter is an interface that defines methods for filtering job instances.

func NewFilter added in v1.0.0 

func NewFilter(opts ...FilterOption) Filter

NewFilter creates a new instance of Filter with the given options.

type FilterOption added in v1.0.0 

type FilterOption func(*filter)

FilterOption is a function that configures a job filter.

func WithFilterAfter added in v1.1.0 

func WithFilterAfter(after time.Time) FilterOption

WithFilterAfter sets the time after which job instances should be filtered.

func WithFilterBefore added in v1.0.0 

func WithFilterBefore(before time.Time) FilterOption

WithFilterBefore sets the time before which job instances should be filtered.

type Handler 

type Handler interface {
	// Executor returns the executor function set for the job handler.
	Executor() Executor
	// StateChangeProcessor returns the state change handler function set for the job handler.
	StateChangeProcessor() StateChangeProcessor
	// CompleteProcessor returns the completion handler function set for the job handler.
	CompleteProcessor() CompleteProcessor
	// TerminateProcessor returns the error processor function set for the job handler.
	TerminateProcessor() TerminateProcessor
	// Execute runs the job with the provided parameters.
	Execute(ctx context.Context, args []any, opts ...any) ([]any, error)
	// HandleTerminated processes errors that occur during job execution.
	HandleTerminated(job Instance, err error) error
	// HandleCompleted processes the responses from a job execution.
	HandleCompleted(job Instance, responses []any)
}

Handler is an interface that defines methods for executing jobs and handling errors.

type HandlerOption 

type HandlerOption func(*handler)

HandlerOption is a function type that applies options to a job handler.

func WithCompleteProcessor added in v1.0.0 

func WithCompleteProcessor(fn CompleteProcessor) HandlerOption

WithCompleteProcessor sets a handler function that is called when a job instance completes successfully during execution by the local worker. NOTE: In a distributed environment with multiple worker groups, the worker that schedules a job instance may be different from the worker that actually executes it.

Example 
// Create a job with a specific complete processor
job, _ := NewJob(
	WithKind("complete"),
	WithExecutor(func() {}),
	WithCompleteProcessor(func(ji Instance, res []any) {
		ji.Infof("Job completed with result: %v", res)
	}),
)
fmt.Printf("%T\n", job.Handler().CompleteProcessor())

Output:
job.CompleteProcessor

func WithExecutor 

func WithExecutor(executor Executor) HandlerOption

WithExecutor sets the executor function for the job handler.

Example (Concat) 
// Create a job with a specific executor
job, _ := NewJob(
	WithKind("concat"),
	WithExecutor(func(a string, b string) string {
		return a + ", " + b
	}),
)
fmt.Printf("%T\n", job.Handler().Executor())

Output:
func(string, string) string
Example (Hello) 
// Create a job with a specific executor
job, _ := NewJob(
	WithKind("hello"),
	WithExecutor(func() {
		fmt.Println("Hello, world!")
	}),
)
fmt.Printf("%T\n", job.Handler().Executor())

Output:
func()
Example (Struct) 
type ConcatString struct {
	A string
	B string
	S string
}
job, _ := NewJob(
	WithKind("struct"),
	WithExecutor(func(param *ConcatString) *ConcatString {
		param.S = param.A + ", " + param.B
		return param
	}),
)
fmt.Printf("%T\n", job.Handler().Executor())

Output:
func(*job.ConcatString) *job.ConcatString
Example (Sum) 
// Create a job with a specific executor
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int {
		return a + b
	}),
)
fmt.Printf("%T\n", job.Handler().Executor())

Output:
func(int, int) int

func WithStateChangeProcessor added in v1.0.0 

func WithStateChangeProcessor(fn StateChangeProcessor) HandlerOption

WithStateChangeProcessor sets a handler function that is invoked each time the state of a job instance changes while being processed by the local worker. NOTE: In a distributed environment with multiple worker groups, the worker that schedules a job instance may not receive all status updates for that instance.

Example 
// Create a job with a specific state change processor
job, _ := NewJob(
	WithKind("state"),
	WithExecutor(func() {}),
	WithStateChangeProcessor(func(ji Instance, state JobState) {
		ji.Infof("State changed to: %v", state)
	}),
)
fmt.Printf("%T\n", job.Handler().StateChangeProcessor())

Output:
func(job.Instance, job.JobState)

func WithTerminateProcessor added in v1.0.0 

func WithTerminateProcessor(fn TerminateProcessor) HandlerOption

WithTerminateProcessor sets a handler function that is called if a job instance ends with an error during execution by the local worker. NOTE: In a distributed environment with multiple worker groups, the worker that schedules a job instance may be different from the worker that actually executes it.

Example 
// Create a job with a specific terminate processor
job, _ := NewJob(
	WithKind("terminate"),
	WithExecutor(func() {}),
	WithTerminateProcessor(func(ji Instance, err error) error {
		if errors.Is(err, context.DeadlineExceeded) {
			// Do not retry if the job was terminated due to a deadline being exceeded
			ji.Infof("Job (%s) terminated due to deadline exceeded: %v", ji.Kind(), err)
			return nil
		}
		// Retry for all other errors
		return err
	}),
)
fmt.Printf("%T\n", job.Handler().TerminateProcessor())

Output:
func(job.Instance, error) error

type History 

type History interface {
	// StateHistory provides methods for managing the state history of job instances.
	StateHistory
	// LogHistory provides methods for logging messages related to job instances.
	LogHistory
}

History is an interface that defines methods for managing the history of job instance state changes.

type HistoryStore 

type HistoryStore interface {
	// StateStore provides methods for managing job instance state history.
	StateStore
	// LogStore provides methods for logging job instance messages.
	LogStore
}

HistoryStore is an interface that defines methods for managing job instance state history.

type Instance 

type Instance interface {
	// Schedule defines the scheduling interface for the job instance.
	Schedule
	// Handler defines the handler interface for the job instance.
	Handler
	// Policy defines the policy interface for the job instance.
	Policy
	// Arguments defines the arguments interface for the job instance.
	Arguments

	// Job returns the job associated with this job instance.
	Job() Job
	// Kind returns the kind of job this instance represents.
	Kind() Kind
	// UUID returns the unique identifier of the job instance.
	UUID() uuid.UUID
	// Context returns the context associated with this job instance, typically passed from the worker.
	Context() context.Context
	// CreatedAt returns the time when the job instance was created.
	CreatedAt() time.Time
	// ScheduledAt returns the time when the job instance was scheduled.
	ScheduledAt() time.Time
	// ProcessedAt returns the time when the job instance started processing.
	ProcessedAt() time.Time
	// CompletedAt returns the completed time when the job instance was completed.
	CompletedAt() time.Time
	// TerminatedAt returns the terminated time when the job instance was terminated.
	TerminatedAt() time.Time
	// CanceledAt returns the time when the job instance was canceled.
	CanceledAt() time.Time
	// TimeoutedAt returns the time when the job instance timed out.
	TimeoutedAt() time.Time
	// Arguments returns the arguments for the job instance.
	Arguments() []any
	// Policy returns the policy associated with the job instance.
	Policy() Policy
	// UpdateState updates the state of the job instance and records the state change.
	UpdateState(state JobState, opts ...any) error
	// Process executes the job instance executor with the arguments provided in the context.
	Process(ctx context.Context, opts ...any) ([]any, error)
	// Result returns the processed result set of the executor when the job instance is completed or terminated.
	// If the job instance is not completed or terminated, it returns an error.
	ResultSet() (ResultSet, error)
	// History returns the history of state changes for the job instance.
	History() (InstanceHistory, error)
	// Logs returns the logs for the job instance.
	Logs() ([]Log, error)
	// State returns the current state of the job instance.
	State() JobState
	// Attempts returns the number of attempts made to process this job instance.
	Attempts() int
	// IsRecurring checks if the job instance is recurring.
	IsRecurring() bool
	// IsRetriable checks if the job instance can be retried.
	IsRetriable() bool
	// Equal checks if two job instances are equal.
	Equal(other Instance) bool
	// Map returns a map representation of the job instance.
	Map() map[string]any
	// JSONString returns a JSON string representation of the job instance.
	JSONString() (string, error)
	// String returns a string representation of the job instance.
	String() string
	// contains filtered or unexported methods
}

Instance represents a specific instance of a job that has been scheduled or executed.

func NewInstance 

func NewInstance(opts ...any) (Instance, error)

NewInstance creates a new JobInstance with a unique identifier and initial state.

func NewInstanceFromMap added in v1.0.0 

func NewInstanceFromMap(m map[string]any, opts ...any) (Instance, error)

NewInstanceFromMap creates a new job instance from the provided map and options.

type InstanceHistory 

type InstanceHistory []InstanceState

InstanceHistory represents a history of job instance executions.

func (InstanceHistory) LastState 

func (history InstanceHistory) LastState() InstanceState

LastState returns the last state of the job instance from its history.

type InstanceOption 

type InstanceOption func(*jobInstance) error

InstanceOption defines a function that configures a job instance.

func WithAttempts added in v1.1.0 

func WithAttempts(attempt int) InstanceOption

WithAttempts sets the number of attempts made to process the job instance.

func WithCanceledAt added in v1.2.0 

func WithCanceledAt(t time.Time) InstanceOption

WithCanceledAt sets the time when the job instance was canceled.

func WithCompletedAt added in v0.9.2 

func WithCompletedAt(t time.Time) InstanceOption

WithCompletedAt sets the time when the job instance was completed.

func WithCreatedAt added in v0.9.2 

func WithCreatedAt(t time.Time) InstanceOption

WithCreatedAt sets the time when the job instance was created.

func WithInstanceHistory 

func WithInstanceHistory(history History) InstanceOption

WithInstanceHistory sets the history for the job instance.

func WithJob added in v1.0.0 

func WithJob(job Job) InstanceOption

WithJob sets the job for the job instance.

func WithProcessingAt added in v0.9.2 

func WithProcessingAt(t time.Time) InstanceOption

WithProcessingAt sets the time when the job instance started processing.

func WithResultError added in v0.9.2 

func WithResultError(err error) InstanceOption

WithResultError sets the error for the job instance result.

func WithResultSet added in v0.9.2 

func WithResultSet(rs ResultSet) InstanceOption

WithResultSet sets the result set for the job instance.

func WithState added in v0.9.2 

func WithState(state JobState) InstanceOption

WithState sets the state of the job instance.

func WithTerminatedAt added in v0.9.2 

func WithTerminatedAt(t time.Time) InstanceOption

WithTerminatedAt sets the time when the job instance was terminated.

func WithTimedOutAt added in v0.9.2 

func WithTimedOutAt(t time.Time) InstanceOption

WithTimedOutAt sets the time when the job instance timed out.

func WithUUID added in v0.9.2 

func WithUUID(uuid uuid.UUID) InstanceOption

WithUUID sets the unique identifier for the job instance.

type InstanceQueue added in v1.0.0 

type InstanceQueue interface {
	// Enqueue adds a job to the queue.
	Enqueue(ctx context.Context, job Instance) error
	// Dequeue removes and returns a job from the queue.
	Dequeue(ctx context.Context) (Instance, error)
	// Remove removes a job from the queue.
	Remove(ctx context.Context, job Instance) error
	// List returns a list of all jobs in the queue.
	List(ctx context.Context) ([]Instance, error)
	// Size returns the number of jobs in the queue.
	Size(ctx context.Context) (int, error)
	// Empty checks if the queue is empty.
	Empty(ctx context.Context) (bool, error)
	// Clear clears all jobs in the queue.
	Clear(ctx context.Context) error
}

InstanceQueue is an interface that defines methods for managing a job scheduled instance queue.

func NewInstanceQueue added in v1.0.0 

func NewInstanceQueue(opts ...InstanceQueueOption) InstanceQueue

NewInstanceQueue creates a new instance of the job queue.

type InstanceQueueOption added in v1.0.0 

type InstanceQueueOption func(*queueImpl)

InstanceQueueOption is a function that configures a job queue.

func WithInstanceQueueStore added in v1.0.0 

func WithInstanceQueueStore(store Store) InstanceQueueOption

WithInstanceQueueStore sets the store for the job queue.

type InstanceState 

type InstanceState interface {
	// Kind returns the kind of the job instance.
	Kind() string
	// UUID returns the unique identifier of the job instance.
	UUID() uuid.UUID
	// Timestamp returns the timestamp of when the state history was created.
	Timestamp() time.Time
	// State returns the state of the job instance.
	State() JobState
	// Options returns the additional options associated with the instance record.
	Options() map[string]any
	// Map returns a map representation of the instance record.
	Map() map[string]any
	// JSONString returns a JSON string representation of the instance record.
	JSONString() (string, error)
	// String returns a string representation of the instance record.
	String() string
}

InstanceState represents the state of a job instance at a specific point in time.

func NewInstanceStateFromMap added in v1.0.0 

func NewInstanceStateFromMap(m map[string]any) (InstanceState, error)

NewInstanceStateFromMap creates a new instance state from a map representation.

type JitterGenerator added in v1.2.2 

type JitterGenerator func() time.Duration

JitterGenerator defines a function that returns a random jitter duration for job scheduling.

type Job 

type Job interface {
	// Kind returns the name of the job.
	Kind() string
	// Description returns a description of the job.
	Description() string
	// Handler returns the job handler for the job.
	Handler() Handler
	// Schedule returns the schedule for the job.
	Schedule() Schedule
	// Policy returns the policy for the job.
	Policy() Policy
	// RegisteredAt returns the time when the job was registered.
	RegisteredAt() time.Time
	// Map returns a map representation of the job.
	Map() map[string]any
	// String returns a string representation of the job.
	String() string
}

Job represents a job that can be scheduled to run at a specific time or interval.

func NewJob 

func NewJob(opts ...any) (Job, error)

NewJob creates a new job with the given name and options.

Example (Abs) 
job, err := NewJob(
	WithKind("abs (one arg and one return)"),
	WithDescription("Returns the absolute value of an integer"),
	WithExecutor(func(a int) int { return int(math.Abs(float64(a))) }),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: abs (one arg and one return)
Example (Concat) 
job, err := NewJob(
	WithKind("concat (two args and one return)"),
	WithDescription("Concatenates two strings"),
	WithExecutor(func(a, b string) string { return a + ", " + b }),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: concat (two args and one return)
Example (MutatingStruct) 
type ConcatString struct {
	A string
	B string
	S string
}
job, err := NewJob(
	WithKind("concat (one struct input and one struct output)"),
	WithDescription("Concatenates two strings from a struct"),
	WithExecutor(func(param *ConcatString) *ConcatString {
		// Store the concatenated string result in the input struct
		param.S = param.A + " " + param.B
		return param
	}),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: concat (one struct input and one struct output)
Example (Simple) 
job, err := NewJob(
	WithKind("no args and no return"),
	WithDescription("A simple job that prints a message"),
	WithExecutor(func() { fmt.Println("Hello, World!") }),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: no args and no return
Example (Split) 
job, err := NewJob(
	WithKind("split (one arg and two return)"),
	WithDescription("Splits a string into two parts"),
	WithExecutor(func(s string) (string, string) {
		parts := strings.Split(s, ",")
		return parts[0], parts[1]
	}),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0], res[1])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: split (one arg and two return)
Example (Struct) 
type SumOpt struct {
	A int
	B int
}
job, err := NewJob(
	WithKind("sum (struct arg and one return)"),
	WithDescription("Returns the sum of two integers from a struct"),
	WithExecutor(func(opt SumOpt) int { return opt.A + opt.B }),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: sum (struct arg and one return)
Example (Sum) 
job, err := NewJob(
	WithKind("sum (two args and one return)"),
	WithDescription("Returns the sum of two integers"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithCompleteProcessor(func(ji Instance, res []any) {
		// In this case, log the result to the go-job manager
		ji.Infof("%v", res[0])
	}),
)
if err != nil {
	fmt.Printf("Error creating job: %v\n", err)
	return
}
fmt.Printf("Created job: %s\n", job.Kind())

Output:
Created job: sum (two args and one return)

type JobOption 

type JobOption func(*job)

JobOption is a function that configures a job.

func WithDescription added in v0.9.2 

func WithDescription(desc string) JobOption

WithDescription sets the description of the job.

Example 
// Create a job with a specific description
job, _ := NewJob(
	WithKind("description"),
	WithExecutor(func() {}),
	WithDescription("This job sums two numbers"),
)
fmt.Printf("%s\n", job.Description())

Output:
This job sums two numbers

func WithKind 

func WithKind(kind string) JobOption

WithKind sets the name of the job.

Example 
// Create a job with a specific kind
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func() {}),
)
fmt.Printf("%s\n", job.Kind())

Output:
sum

type JobState 

type JobState int

JobState represents the state of a job as an integer. 

const (
	// JobStateUnset indicates that the job state is not set.
	JobStateUnset JobState = 0 // 0 indicates an unset state
	// JobCreated indicates the job has been created but not yet started.
	JobCreated JobState = 1 << iota
	// JobScheduled indicates the job has been scheduled for execution.
	JobScheduled
	// JobProcessing indicates the job is currently being processed.
	JobProcessing
	// JobCanceled indicates the job was canceled before completion.
	JobCanceled
	// JobTimedOut indicates the job has exceeded its allowed execution time.
	JobTimedOut
	// JobCompleted indicates the job has completed (either successfully or unsuccessfully).
	JobCompleted
	// JobTerminated indicates the job has been terminated.
	JobTerminated
)

func (JobState) Is 

func (s JobState) Is(state JobState) bool

Is checks if the current JobState is equal to the provided state.

func (JobState) Matches added in v1.1.0 

func (s JobState) Matches(state JobState) bool

Matches checks if the current JobState matches the provided state.

func (JobState) String 

func (s JobState) String() string

String returns the string representation of the JobState.

type Kind 

type Kind = string

Kind is a type that represents the kind of a job.

type Log 

type Log interface {
	// Kind returns the type of the log entry.
	Kind() string
	// UUID returns the unique identifier of the log entry.
	UUID() uuid.UUID
	// Timestamp returns the timestamp of the log entry.
	Timestamp() time.Time
	// Level returns the log level of the log entry.
	Level() LogLevel
	// Message returns the message of the log entry.
	Message() string
	// Equal checks if two log entries are equal.
	Equal(other Log) bool
	// Map returns a map representation of the log entry.
	Map() map[string]any
	// String returns the string representation of the log entry.
	String() string
}

Log represents a log entry associated with a job.

func NewLog 

func NewLog(opts ...LogOption) Log

NewLog creates a new log entry with the specified options.

func NewLogFromMap added in v1.0.0 

func NewLogFromMap(m map[string]any) (Log, error)

NewLogFromMap creates a new log entry from a map representation.

type LogHistory 

type LogHistory interface {
	// Infof logs an informational message for a job instance.
	Infof(job Instance, format string, args ...any) error
	// Warnf logs a warning message for a job instance.
	Warnf(job Instance, format string, args ...any) error
	// Errorf logs an error message for a job instance.
	Errorf(job Instance, format string, args ...any) error
	// Debugf logs a debug message for a job instance.
	Debugf(job Instance, format string, args ...any) error
	// LookupLogs lists all log entries for a job instance that match the specified query. The returned logs are sorted by their timestamp.
	LookupLogs(query Query) ([]Log, error)
	// ClearLogs clears all log entries for a job instance that match the specified filter.
	ClearLogs(filter Filter) error
}

LogHistory is an interface that defines methods for logging messages related to job instances.

type LogLevel 

type LogLevel int

LogLevel represents the level of log messages. It is used to categorize log messages for better organization and filtering. 

const (
	// LogInfo represents informational log messages.
	LogInfo LogLevel = 1 << iota // 1
	// LogError represents error log messages.
	LogError // 2
	// LogWarn represents warning log messages.
	LogWarn // 4
	// LogDebug represents debug log messages.
	LogDebug // 8
	// LogNone represents no log messages.
	LogNone LogLevel = 0 // 0
	// LogAll represents all log levels combined.
	LogAll LogLevel = LogInfo | LogError | LogWarn | LogDebug // 15
)

func NewLogLevelFromString added in v1.0.0 

func NewLogLevelFromString(s string) (LogLevel, error)

NewLogLevelFromString returns the LogLevel corresponding to the given string.

func (LogLevel) Contains 

func (l LogLevel) Contains(other LogLevel) bool

Contains checks if the LogLevel contains another LogLevel.

func (LogLevel) String 

func (l LogLevel) String() string

String returns the string representation of the Log.

type LogOption 

type LogOption func(*log)

LogOption defines a function that configures a log entry.

func WithLogKind added in v0.9.1 

func WithLogKind(kind string) LogOption

WithLogKind sets the type of the log entry.

func WithLogLevel 

func WithLogLevel(level LogLevel) LogOption

WithLogLevel sets the log level of the log entry.

func WithLogMessage 

func WithLogMessage(msg string) LogOption

WithLogMessage sets the message of the log entry.

func WithLogTimestamp 

func WithLogTimestamp(ts time.Time) LogOption

WithLogTimestamp sets the timestamp of the log entry.

func WithLogUUID 

func WithLogUUID(uuid uuid.UUID) LogOption

WithLogUUID sets the unique identifier of the log entry.

type LogStore 

type LogStore interface {
	// Infof logs an informational message for a job instance.
	Infof(ctx context.Context, job Instance, format string, args ...any) error
	// Warnf logs a warning message for a job instance.
	Warnf(ctx context.Context, job Instance, format string, args ...any) error
	// Errorf logs an error message for a job instance.
	Errorf(ctx context.Context, job Instance, format string, args ...any) error
	// Debugf logs a debug message for a job instance.
	Debugf(ctx context.Context, job Instance, format string, args ...any) error
	// LookupInstanceLogs lists all log entries for a job instance that match the specified query. The returned logs are sorted by their timestamp.
	LookupInstanceLogs(ctx context.Context, query Query) ([]Log, error)
	// ClearInstanceLogs clears all log entries for a job instance that match the specified filter.
	ClearInstanceLogs(ctx context.Context, filter Filter) error
}

LogStore is an interface that defines methods for logging job instance messages.

type Manager 

type Manager interface {
	// Store returns the job store.
	Store() Store

	// RegisterJob registers a job in the registry. If a job with the same kind is already registered,
	// it will be overwritten with the new job.
	RegisterJob(job Job) error
	// UnregisterJob removes a job from the registry by its kind.
	UnregisterJob(kind Kind) error
	// ListJobs returns a slice of all registered jobs.
	ListJobs() ([]Job, error)
	// LookupJob looks up a job by its kind in the registry.
	LookupJob(kind Kind) (Job, bool)

	// ScheduleJob schedules a job instance with the given job and options.
	// It creates a new job instance and enqueues it in the job queue.
	// If no schedule option is set, the job instance will be scheduled to run immediately by default.
	// If the specified job is not registered, the manager will register the job automatically.
	ScheduleJob(job Job, opts ...any) (Instance, error)
	// ScheduleRegisteredJob schedules a registered job by its kind with the given options.
	// If the job is not registered, an error will be returned.
	// It creates a new job instance and enqueues it in the job queue.
	// If the schedule option is not set, the job instance will be scheduled to run immediately as default.
	ScheduleRegisteredJob(kind Kind, opts ...any) (Instance, error)
	// EnqueueInstance enqueues a job instance in the job queue.
	EnqueueInstance(job Instance) error
	// DequeueNextInstance returns the next scheduled job instance and dequeues it from the job queue.
	DequeueNextInstance() (Instance, error)
	// LookupInstances looks up all job instances which match the specified query.
	LookupInstances(query Query) ([]Instance, error)
	// CancelInstances cancels all job instances which match the specified query.
	CancelInstances(query Query) ([]Instance, error)
	// ListInstances returns all job instances which are currently scheduled, processing, completed, or terminated after the manager started.
	ListInstances() ([]Instance, error)

	// LookupHistory retrieves all state records for a job instance, sorted by timestamp.
	LookupInstanceHistory(query Query) (InstanceHistory, error)
	// ClearInstanceHistory clears all state records for a job instance that match the specified filter.
	ClearInstanceHistory(filter Filter) error

	// LookupLogs retrieves all logs for a job instance.
	LookupInstanceLogs(query Query) ([]Log, error)
	// ClearInstanceLogs clears all log entries for a job instance that match the specified filter.
	ClearInstanceLogs(filter Filter) error

	// Workers returns a list of all workers in the group.
	Workers() []Worker
	// ResizeWorkers scales the number of workers in the group.
	ResizeWorkers(ctx context.Context, num int) error
	// NumWorkers returns the number of workers in the group.
	NumWorkers() int

	// Start starts the job manager.
	Start() error
	// Stop stops the job manager.
	Stop() error
	// Wait waits for all scheduled jobs to complete or terminate.
	Wait(ctx context.Context) error
	// Clear clears all jobs and history from the job manager without registered jobs.
	Clear() error
}

Manager is an interface that defines methods for managing jobs.

Example (ResizeWorkers) 
// Create a manager with 1 worker (default)
mgr, _ := NewManager(WithNumWorkers(1))
mgr.Start()
defer mgr.Stop()

// Print the current number of workers
fmt.Println("Initial workers:", mgr.NumWorkers())

// Monitor queue size and scale accordingly
query := NewQuery(
	WithQueryState(JobScheduled), // filter by scheduled state
)
jobs, _ := mgr.LookupInstances(query)
queueSize := len(jobs)
currentWorkers := mgr.NumWorkers()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if queueSize > currentWorkers*2 {
	// Scale up if queue is getting too long
	mgr.ResizeWorkers(ctx, currentWorkers+2)
} else if queueSize == 0 && currentWorkers > 2 {
	// Scale down if no jobs queued
	mgr.ResizeWorkers(ctx, currentWorkers-1)
}

// Print the current number of workers
fmt.Println("Active workers:", mgr.NumWorkers())

Output:
Initial workers: 1
Active workers: 1
Example (ScheduleJob) 
// Create a job manager
mgr, _ := NewManager()
// Create a job with a custom executor
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
)
// Schedule the job with the manager
mgr.ScheduleJob(
	job,
	WithScheduleAt(time.Now()), // Immediate scheduling is the default, so this option is redundant
	WithCompleteProcessor(func(inst Instance, res []any) {
		fmt.Printf("Result: %v\n", res)
	}),
	WithArguments(1, 2),
)
// Start the job manager
mgr.Start()

// Wait waits for all jobs to complete or terminate.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
mgr.Wait(ctx)

// Stop the job manager
mgr.Stop()

Output:
Result: [3]
Example (ScheduleRegisteredJob) 
// Create a job manager
mgr, _ := NewManager()
// Create a job with a custom executor
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
)
// Register the job with the manager
mgr.RegisterJob(job)
// Schedule the registered job
mgr.ScheduleRegisteredJob(
	job.Kind(),
	WithScheduleAt(time.Now()), // Immediate scheduling is the default, so this option is redundant
	WithCompleteProcessor(func(inst Instance, res []any) {
		fmt.Printf("Result: %v\n", res)
	}),
	WithArguments(1, 2),
)
// Start the job manager
mgr.Start()

// Wait waits for all jobs to complete or terminate.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
mgr.Wait(ctx)

// Stop the job manager
mgr.Stop()

Output:
Result: [3]

func NewManager 

func NewManager(opts ...any) (Manager, error)

NewManager creates a new instance of the job manager.

type ManagerOption 

type ManagerOption func(*manager)

ManagerOption is a function that configures a job manager.

func WithStore 

func WithStore(store Store) ManagerOption

WithManagerQueue sets the queue for the job manager.

type Policy 

type Policy interface {
	// MaxRetries returns the maximum number of retries allowed for the job.
	MaxRetries() int
	// Priority returns the priority of the job.
	Priority() Priority
	// Timeout returns the timeout duration for the job.
	Timeout() time.Duration
	// BackoffStrategy returns the backoff strategy for the job.
	BackoffStrategy() BackoffStrategy
	// Map returns a map representation of the job instance.
	Map() map[string]any
	// String returns a string representation of the job instance.
	String() string
}

Policy defines the interface for job scheduling, supporting crontab expressions.

type PolicyOption 

type PolicyOption func(*policy)

PolicyOption defines a function that configures a job policy.

func WithBackoffDuration added in v0.9.1 

func WithBackoffDuration(duration time.Duration) PolicyOption

WithBackoffDuration sets a fixed backoff duration with random jitter for the job policy.

func WithBackoffStrategy added in v0.9.1 

func WithBackoffStrategy(fn BackoffStrategy) PolicyOption

WithBackoffStrategy sets the function to determine the delay before retrying a job.

Example 
// Create and register a job with a specific backoff strategy
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithBackoffStrategy(func(ji Instance) time.Duration {
		// Exponential backoff
		return time.Duration(float64(ji.Attempts()) * float64(time.Second) * (0.8 + 0.4*rand.Float64()))
	}))

fmt.Printf("BackoffStrategy: %T\n", job.Policy().BackoffStrategy())

Output:
BackoffStrategy: job.BackoffStrategy

func WithHighPriority 

func WithHighPriority() PolicyOption

WithHighPriority sets the job policy to high priority.

func WithInfiniteRetries 

func WithInfiniteRetries() PolicyOption

WithInfiniteRetries sets the job policy to retry indefinitely.

func WithLowPriority 

func WithLowPriority() PolicyOption

WithLowPriority sets the job policy to low priority.

func WithMaxRetries 

func WithMaxRetries(count int) PolicyOption

WithMaxRetries sets the maximum number of retries for the job policy.

Example 
// Create a job manager
mgr, _ := NewManager()

// Create and register a job with a specific max retries
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithMaxRetries(3),
)
mgr.RegisterJob(job)
fmt.Printf("MaxRetries: %d\n", job.Policy().MaxRetries())

// Start the job manager
mgr.Start()
defer mgr.Stop()

// Schedule the registered job with default job max retries
ji, _ := mgr.ScheduleRegisteredJob("sum")
fmt.Printf("MaxRetries: %d\n", ji.MaxRetries())

// Schedule the registered job with an overridden max retries
ji, _ = mgr.ScheduleRegisteredJob("sum", WithMaxRetries(5))
fmt.Printf("MaxRetries: %d\n", ji.MaxRetries())

Output:
MaxRetries: 3
MaxRetries: 3
MaxRetries: 5

func WithNoTimeout 

func WithNoTimeout() PolicyOption

WithNoTimeout sets the job policy to have no timeout limit.

func WithPriority 

func WithPriority(priority Priority) PolicyOption

WithPriority sets the priority for the job policy.

Example 
// Create a job manager
mgr, _ := NewManager()

// Create and register a job with a specific priority
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithPriority(0),
)
mgr.RegisterJob(job)
fmt.Printf("Priority: %d\n", job.Policy().Priority())

// Start the job manager
mgr.Start()
defer mgr.Stop()

// Schedule the registered job with default job priority
ji, _ := mgr.ScheduleRegisteredJob("sum")
fmt.Printf("Priority: %d\n", ji.Priority())

// Schedule the registered job with an overridden priority
ji, _ = mgr.ScheduleRegisteredJob("sum", WithPriority(1))
fmt.Printf("Priority: %d\n", ji.Priority())

Output:
Priority: 0
Priority: 0
Priority: 1

func WithTimeout 

func WithTimeout(duration time.Duration) PolicyOption

WithTimeout sets the timeout duration for the job policy.

Example 
// Create a job manager
mgr, _ := NewManager()

// Create and register a job with a specific timeout
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithTimeout(5*time.Second),
)
mgr.RegisterJob(job)
fmt.Printf("Timeout: %v\n", job.Policy().Timeout())

// Start the job manager
mgr.Start()
defer mgr.Stop()

// Schedule the registered job with default job timeout
ji, _ := mgr.ScheduleRegisteredJob("sum")
fmt.Printf("Timeout: %v\n", ji.Timeout())

// Schedule the registered job with an overridden timeout
ji, _ = mgr.ScheduleRegisteredJob("sum", WithTimeout(10*time.Second))
fmt.Printf("Timeout: %v\n", ji.Timeout())

Output:
Timeout: 5s
Timeout: 5s
Timeout: 10s

type Priority 

type Priority int

Priority represents the priority of a job. A lower value means a higher priority, similar to the Unix nice value.

func NewPriorityFrom added in v1.0.0 

func NewPriorityFrom(a any) (Priority, error)

NewPriorityFrom creates a Priority from various input types.

func (Priority) Equal 

func (p Priority) Equal(other Priority) bool

Equal checks if the priority is equal to another priority.

func (Priority) Higher 

func (p Priority) Higher(other Priority) bool

Higher checks if the priority is higher than another priority.

func (Priority) Lower 

func (p Priority) Lower(other Priority) bool

Lower checks if the priority is lower than another priority.

func (Priority) String 

func (p Priority) String() string

String returns the string representation of the priority.

type Query added in v0.9.2 

type Query interface {
	// Filter provides additional filtering methods for time-based or custom criteria.
	Filter

	// UUID returns the UUID criterion for the query, if set.
	UUID() (uuid.UUID, bool)
	// Kind returns the kind criterion for the query, if set.
	Kind() (string, bool)
	// State returns the job state criterion for the query, if set.
	State() (JobState, bool)
	// LogLevel returns the log level criterion for the query, if set.
	LogLevel() (LogLevel, bool)

	// IsUnset returns true if no query criteria are set.
	IsUnset() bool

	// Matches returns true if the specified object satisfies all query criteria.
	Matches(v any) bool
}

Query defines an interface for specifying and evaluating query criteria for jobs, instances, and logs.

func NewQuery 

func NewQuery(opts ...QueryOption) Query

type QueryOption 

type QueryOption func(*query)

QueryOption is a function that configures a job query.

func WithQueryAfter added in v1.1.0 

func WithQueryAfter(after time.Time) QueryOption

WithQueryAfter sets the time after which target should be filtered.

func WithQueryBefore added in v1.1.0 

func WithQueryBefore(before time.Time) QueryOption

WithQueryBefore sets the time before which target should be filtered.

func WithQueryInstance added in v1.0.0 

func WithQueryInstance(instance Instance) QueryOption

WithQueryInstance sets the query UUID and kind based on an existing job instance.

func WithQueryKind added in v0.9.2 

func WithQueryKind(kind string) QueryOption

WithQueryKind sets the kind for the query.

func WithQueryLogLevel added in v1.1.0 

func WithQueryLogLevel(level LogLevel) QueryOption

WithQueryLogLevel sets the level for the query.

func WithQueryState added in v0.9.2 

func WithQueryState(state JobState) QueryOption

WithQueryState sets the state for the query.

func WithQueryUUID added in v0.9.2 

func WithQueryUUID(uuid uuid.UUID) QueryOption

WithQueryUUID sets the UUID for the query.

type QueueStore 

type QueueStore interface {
	// EnqueueInstance stores a job instance in the store.
	EnqueueInstance(ctx context.Context, job Instance) error
	// DequeueInstance removes a specific job instance from the store.
	DequeueInstance(ctx context.Context, job Instance) error
	// DequeueNextInstance retrieves and removes the highest priority job instance from the store. If no job instance is available, it returns nil.
	DequeueNextInstance(ctx context.Context) (Instance, error)
	// ListInstances lists all job instances in the store.
	ListInstances(ctx context.Context) ([]Instance, error)
	// ClearInstances clears all job instances in the store.
	ClearInstances(ctx context.Context) error
}

QueueStore is an interface that defines methods for managing job instances in a pending state.

type ResultSet added in v0.9.2 

type ResultSet []any

ResultSet represents the result of a job execution.

func Execute 

func Execute(fn any, args []any, opts ...any) (ResultSet, error)

Execute calls the given function with the provided parameters and returns results as []any.

func (ResultSet) String added in v0.9.2 

func (r ResultSet) String() string

String returns a string representation of the Result.

type Schedule 

type Schedule interface {
	// CrontabSpec returns the crontab spec string.
	CrontabSpec() string
	// IsScheduled returns true if the schedule has timing configuration.
	IsScheduled() bool
	// IsRecurring checks if the job is recurring.
	IsRecurring() bool
	// Next returns the next scheduled time.
	Next() time.Time
	// Jitter returns the jitter duration for the job.
	Jitter() JitterGenerator
	// Map returns a map representation of the job.
	Map() map[string]any
	// String returns a string representation of the job.
	String() string
}

Schedule defines the interface for job scheduling, supporting crontab expressions.

func NewSchedule 

func NewSchedule(opts ...ScheduleOption) (Schedule, error)

NewSchedule creates a new schedule instance with the provided options. Available options include WithCrontabSpec() for cron-based scheduling and WithScheduleAt() for one-time scheduling. If no options are provided, the current time is used as the default schedule.

type ScheduleOption 

type ScheduleOption func(*schedule) error

ScheduleOption defines a function that configures a job schedule.

func WithCrontabSpec 

func WithCrontabSpec(spec string) ScheduleOption

WithCrontabSpec sets the crontab spec string for the job schedule.

func WithJitter added in v1.2.1 

func WithJitter(jitterFunc JitterGenerator) ScheduleOption

WithJitter sets the job schedule jitter function.

Example 
// Create a job manager
mgr, _ := NewManager()

// Create and register a job with a specific jitter
job, _ := NewJob(
	WithKind("sum"),
	WithExecutor(func(a, b int) int { return a + b }),
	WithJitter(
		func() time.Duration {
			return 100 * time.Millisecond
		},
	),
)
mgr.RegisterJob(job)
fmt.Printf("Jitter: %v\n", job.Schedule().Jitter()())

// Start the job manager
mgr.Start()
defer mgr.Stop()

// Schedule the registered job with default job jitter
ji, _ := mgr.ScheduleRegisteredJob("sum")
fmt.Printf("Jitter: %v\n", ji.Jitter()())

// Schedule the registered job with an overridden jitter
ji, _ = mgr.ScheduleRegisteredJob(
	"sum",
	WithJitter(
		func() time.Duration {
			return 200 * time.Millisecond
		},
	))
fmt.Printf("Jitter: %v\n", ji.Jitter()())

Output:
Jitter: 100ms
Jitter: 100ms
Jitter: 200ms

func WithScheduleAfter 

func WithScheduleAfter(d time.Duration) ScheduleOption

WithScheduleNow sets the job schedule to the current time.

func WithScheduleAt 

func WithScheduleAt(t time.Time) ScheduleOption

WithSchedule sets the cron.Schedule for the job schedule.

type Server added in v1.0.0 

type Server interface {
	// Config is the interface for the job server configuration.
	Config
	// Manager returns the job manager associated with the server.
	Manager() Manager
	// Start starts the job server.
	Start() error
	// Stop stops the job server.
	Stop() error
	// Restart restarts the job server.
	Restart() error
}

Server is an interface that defines methods for managing the job server.

func NewServer added in v1.0.0 

func NewServer(opts ...any) (Server, error)

NewServer returns a new job server instance.

type StateChangeProcessor added in v1.0.0 

type StateChangeProcessor = func(job Instance, state JobState)

StateChangeProcessor is called when a job's state changes.

type StateHistory 

type StateHistory interface {
	// LogProcessState logs a state change for a job instance.
	LogProcessState(job Instance, state JobState, opts ...instanceStateOption) error
	// LookupHistory lists all state records for a job instance that match the specified query. The returned history is sorted by their timestamp.
	LookupHistory(query Query) (InstanceHistory, error)
	// ClearHistory clears all state records for a job instance that match the specified filter.
	ClearHistory(filter Filter) error
}

StateHistory is an interface that defines methods for managing the state history of job instances.

type StateStore 

type StateStore interface {
	// LogInstanceState adds a new state record for a job instance.
	LogInstanceState(ctx context.Context, state InstanceState) error
	// LookupInstanceHistory lists all state records for a job instance that match the specified query. The returned history is sorted by their timestamp.
	LookupInstanceHistory(ctx context.Context, query Query) (InstanceHistory, error)
	// ClearInstanceHistory clears all state records for a job instance that match the specified filter.
	ClearInstanceHistory(ctx context.Context, filter Filter) error
}

StateStore is an interface that defines methods for managing job instance state history.

type Store 

type Store interface {
	// Name returns the name of the store.
	Name() string
	// PendingStore provides methods for managing job instances.
	QueueStore
	// HistoryStore provides methods for managing job instance state history.
	HistoryStore
	// Start starts the store.
	Start() error
	// Stop stops the store.
	Stop() error
	// Clear clears all data in the store.
	Clear() error
}

Store defines the interface for job queue, history, and logging.

func NewLocalStore 

func NewLocalStore() Store

NewLocalStore creates a new in-memory job store.

type TerminateProcessor added in v1.0.0 

type TerminateProcessor = func(job Instance, err error) error

TerminateProcessor is called when a job reaches the terminated state. Users can:

  • Return nil to resolve the error (mark job as successful)
  • Return a modified error to transform the error
  • Return the original error to keep it unchanged

type Timestamp added in v1.0.0 

type Timestamp time.Time

Timestamp represents a point in time.

func NewTimestamp added in v1.0.0 

func NewTimestamp() Timestamp

NewTimestamp creates a new Timestamp from the current time.

func NewTimestampFrom added in v1.0.0 

func NewTimestampFrom(a any) (Timestamp, error)

NewTimestampFrom creates a new Timestamp from a given value.

func NewTimestampFromString added in v1.0.0 

func NewTimestampFromString(s string) (Timestamp, error)

NewTimestampFromString creates a new Timestamp from a string representation of time.

func NewTimestampFromTime added in v1.0.0 

func NewTimestampFromTime(t time.Time) Timestamp

NewTimestampFromTime creates a new Timestamp from a time.Time value.

func (Timestamp) Equal added in v1.0.0 

func (t Timestamp) Equal(other Timestamp) bool

Equal checks if two Timestamps are equal.

func (Timestamp) String added in v1.0.0 

func (t Timestamp) String() string

String returns the string representation of the Timestamp in RFC3339 format.

func (Timestamp) Time added in v1.0.0 

func (t Timestamp) Time() time.Time

Time returns the time.Time representation of the Timestamp.

type UUID added in v1.0.0 

type UUID = uuid.UUID

UUID is a type alias for uuid.UUID to represent job instance UUIDs.

func NewUUID added in v1.0.0 

func NewUUID() UUID

NewUUID generates a new UUID for a job instance.

func NewUUIDFrom added in v1.0.0 

func NewUUIDFrom(a any) (UUID, error)

NewUUIDFrom creates a new UUID from a given value.

func NewUUIDFromString added in v1.0.0 

func NewUUIDFromString(s string) (UUID, error)

NewUUIDFromString creates a UUID from a string representation.

type Worker 

type Worker interface {
	// Start starts the worker to process jobs.
	Start() error
	// Cancel cancels the currently processing job. Returns an error if no job is being processed.
	Cancel() error
	// Wait waits for the worker to finish processing jobs.
	Wait(ctx context.Context) error
	// Stop cancels the worker from processing jobs.
	Stop() error
	// IsProcessing returns true if the worker is currently processing a job.
	IsProcessing() bool
	// ProcessingInstance returns the job instance being processed, if any.
	ProcessingInstance() (Instance, bool)
}

Worker is an interface that defines methods for processing jobs.

type WorkerGroup 

type WorkerGroup interface {
	// Start starts all workers in the group.
	Start() error
	// Stop stops all workers in the group.
	Stop() error
	// Wait waits for all workers in the group to finish processing.
	Wait(ctx context.Context) error
	// Workers returns a list of all workers in the group.
	Workers() []Worker
	// ResizeWorkers scales the number of workers in the group.
	ResizeWorkers(ctx context.Context, num int) error
	// NumWorkers returns the number of workers in the group.
	NumWorkers() int
}

WorkerGroup is an interface that defines methods for managing a group of workers.

type WorkerGroupOption 

type WorkerGroupOption func(*workerGroup)

WorkerGroupOption defines a function that configures a worker group.

func WithNumWorkers 

func WithNumWorkers(number int) WorkerGroupOption

WithWorkerGroupNumber sets the number of workers in the group.

Source Files

View all Source files

Directories

Path Synopsis
api
gen/go/v1
cmd
cli
server
Package encoding provides utilities for encoding and decoding data in various formats used within the go-job project.
 Package encoding provides utilities for encoding and decoding data in various formats used within the go-job project.
Package plugin provides interfaces and utilities for implementing plugins within the job system.
 Package plugin provides interfaces and utilities for implementing plugins within the job system.
job
Package job provides pluggable system job backends for go-job.
 Package job provides pluggable system job backends for go-job.
job/system
store
Package store provides pluggable storage backends for go-job.
 Package store provides pluggable storage backends for go-job.
store/kv
Package kv provides a generic key-value store interface and utilities for go-job.
 Package kv provides a generic key-value store interface and utilities for go-job.
store/kv/etcd
Package etcd provides an etcd-based key-value store implementation for go-job.
 Package etcd provides an etcd-based key-value store implementation for go-job.
store/kv/memdb
Package memdb provides an in-memory key-value store implementation for go-job.
 Package memdb provides an in-memory key-value store implementation for go-job.
store/kv/redis
Package redis provides a Redis-based key-value store implementation for go-job.
 Package redis provides a Redis-based key-value store implementation for go-job.
store/kv/valkey
Package valkey provides a Valkey-based key-value store implementation for go-job.
 Package valkey provides a Valkey-based key-value store implementation for go-job.
store/kvutil
Package kvutil provides utility functions for key-value store plugins in go-job.
 Package kvutil provides utility functions for key-value store plugins in go-job.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.