relay

README

Package relay

This package is a Go client for the Relay Job Runner found here.

Usage

See example here.

Requirements

• Go 1.18+

How to Contribute

Make a pull request...

License

Distributed under MIT License, please see license file within the code for more details.

Documentation

Index

• type Client
  • func New[P any, S any](cfg Config) (*Client[P, S], error)
  • func (r *Client[P, S]) Enqueue(ctx context.Context, job Job[P, S]) error
  • func (r *Client[P, S]) EnqueueBatch(ctx context.Context, jobs []Job[P, S]) error
  • func (r *Client[P, S]) Exists(ctx context.Context, queue, jobID string) (bool, error)
  • func (r *Client[P, S]) ExistsWithRetry(ctx context.Context, queue, jobID string) (bool, error)
  • func (r *Client[P, S]) Get(ctx context.Context, queue, jobID string) (*Job[P, S], error)
  • func (r *Client[P, S]) GetWithRetry(ctx context.Context, queue, jobID string) (*Job[P, S], error)
  • func (r *Client[P, S]) Next(ctx context.Context, queue string, num_jobs uint32) ([]*JobHelper[P, S], error)
  • func (r *Client[P, S]) Remove(ctx context.Context, queue, jobID string) error
• type Config
• type ErrJobExits
  • func (e ErrJobExits) Error() string
• type ErrNotFound
  • func (e ErrNotFound) Error() string
• type Job
• type JobHelper
  • func (j *JobHelper[P, S]) Complete(ctx context.Context) error
  • func (j *JobHelper[P, S]) CompleteWithRetry(ctx context.Context) error
  • func (j *JobHelper[P, S]) Heartbeat(ctx context.Context, state *S) error
  • func (j *JobHelper[P, S]) HeartbeatAuto(ctx context.Context, interval time.Duration)
  • func (j *JobHelper[P, S]) Job() *Job[P, S]
  • func (j *JobHelper[P, S]) Reschedule(ctx context.Context, job Job[P, S]) error
  • func (j *JobHelper[P, S]) RescheduleWithRetry(ctx context.Context, job Job[P, S]) error

Types

type Client

type Client[P any, S any] struct {
	// contains filtered or unexported fields
}

Client is used to interact with the Client Job Server.

func New

func New[P any, S any](cfg Config) (*Client[P, S], error)

New creates a new Client instance for use.

func (*Client[P, S]) Enqueue

func (r *Client[P, S]) Enqueue(ctx context.Context, job Job[P, S]) error

Enqueue submits the provided Job for processing to the Job Server.

func (*Client[P, S]) EnqueueBatch

func (r *Client[P, S]) EnqueueBatch(ctx context.Context, jobs []Job[P, S]) error

EnqueueBatch submits one or more Jobs for processing to the Job Server in one call.

func (*Client[P, S]) Exists

func (r *Client[P, S]) Exists(ctx context.Context, queue, jobID string) (bool, error)

Exists checks if a Job exists.

func (*Client[P, S]) ExistsWithRetry

func (r *Client[P, S]) ExistsWithRetry(ctx context.Context, queue, jobID string) (bool, error)

ExistsWithRetry is the same as Exits only automatically handles retryable errors.

func (*Client[P, S]) Get

func (r *Client[P, S]) Get(ctx context.Context, queue, jobID string) (*Job[P, S], error)

Get retrieves a Job from the database for debugging or usage of state data.

func (*Client[P, S]) GetWithRetry

func (r *Client[P, S]) GetWithRetry(ctx context.Context, queue, jobID string) (*Job[P, S], error)

GetWithRetry is the same as Get only automatically handles retryable errors.

func (*Client[P, S]) Next

func (r *Client[P, S]) Next(ctx context.Context, queue string, num_jobs uint32) ([]*JobHelper[P, S], error)

Next attempts to retrieve the next Job in the `queue` requested. It will retry and backoff attempting to retrieve a Job and will block until retrieving a Job or the Context is cancelled.

func (*Client[P, S]) Remove

func (r *Client[P, S]) Remove(ctx context.Context, queue, jobID string) error

Remove removes the Job from the DB for processing. In fact this function makes a call to the complete endpoint.

NOTE: It does not matter if the Job is in-flight or not it will be removed. All relevant code paths return an

ErrNotFound to handle such events within Job Workers so that they can bail gracefully if desired.

type Config

type Config struct {
	// BasURL of the HTTP server
	BaseURL string

	// NextBackoff if the backoff used when calling the `next` endpoint and there is no data yet available.
	// Optional: If not set a default backoff is used.
	NextBackoff backoff.Exponential

	// RetryBackoff is the backoff used when calling any of the retryable functions.
	RetryBackoff backoff.Exponential

	// Client is the HTTP Client to use if using a custom one is desired.
	// Optional: If not set it will create a new one cloning the `http.DefaultTransport` and tweaking the settings
	//           for use with sane limits & Defaults.
	Client *http.Client
}

Config contains all information to create a new Relay instance fo use.

type ErrJobExits

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

ErrJobExits denotes that the Job that was attempted to be submitted/enqueued on the Job Server already exists and the Job was not accepted because of this.

func (ErrJobExits) Error

func (e ErrJobExits) Error() string

Error returns the error in string form.

type ErrNotFound

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

ErrNotFound indicates that the queue and/or Job you specified could not be found on the Job Server.

func (ErrNotFound) Error

func (e ErrNotFound) Error() string

Error returns the error in string form.

type Job

type Job[P any, S any] struct {

	// ID is the unique Job ID which is also CAN be used to ensure the Job is a singleton.
	ID string `json:"id"`

	// Queue is used to differentiate different job types that can be picked up by job runners.
	Queue string `json:"queue"`

	// Timeout denotes the duration, in seconds, after a Job has started processing or since the last
	// heartbeat request occurred before considering the Job failed and being put back into the
	// queue.
	Timeout int32 `json:"timeout"`

	// MaxRetries determines how many times the Job can be retried, due to timeouts, before being considered
	// permanently failed.
	MaxRetries int32 `json:"max_retries,omitempty"`

	// Payload is the raw JSON payload that the job runner will receive.
	Payload P `json:"payload"`

	// State is the raw JSON payload that the job runner will receive.
	State *S `json:"state,omitempty"`

	// RunAt can optionally schedule/set a Job to be run only at a specific time in the
	// future. This option should mainly be used for one-time jobs and scheduled jobs that have
	// the option of being self-perpetuated in combination with the rescheduling endpoint.
	RunAt *time.Time `json:"run_at,omitempty"`

	// UpdatedAt indicates last time the Job was updated either through enqueue, reschedule or heartbeat.
	// This value is for reporting purposes only and will be ignored when enqueuing and rescheduling Jobs.
	UpdatedAt *time.Time `json:"updated_at"`
}

Job defines all information needed to process a job.

type JobHelper

type JobHelper[P any, S any] struct {
	// contains filtered or unexported fields
}

JobHelper is used to process an individual Job retrieved from the Job Server. It contains a number of helper methods to `Heartbeat` and `Complete` Jobs.

func (*JobHelper[P, S]) Complete

func (j *JobHelper[P, S]) Complete(ctx context.Context) error

Complete marks the Job as complete. It does NOT matter to the Job Runner if the job was successful or not.

func (*JobHelper[P, S]) CompleteWithRetry

func (j *JobHelper[P, S]) CompleteWithRetry(ctx context.Context) error

CompleteWithRetry is the same as Complete but also automatically retries on transient errors.

func (*JobHelper[P, S]) Heartbeat

func (j *JobHelper[P, S]) Heartbeat(ctx context.Context, state *S) error

Heartbeat calls the Job Runners heartbeat endpoint to keep the job alive. Optional: It optionally accepts a state payload if desired to be used in case of failure for

point-in-time restarting.

func (*JobHelper[P, S]) HeartbeatAuto

func (j *JobHelper[P, S]) HeartbeatAuto(ctx context.Context, interval time.Duration)

HeartbeatAuto automatically calls the Job Runners heartbeat endpoint in a separate goroutine on the provided interval. It is convenience to use this when no state needs to be saved but Job kept alive.

func (*JobHelper[P, S]) Job

func (j *JobHelper[P, S]) Job() *Job[P, S]

Job returns the Job to process

func (*JobHelper[P, S]) Reschedule

func (j *JobHelper[P, S]) Reschedule(ctx context.Context, job Job[P, S]) error

Reschedule submits the provided Job for processing by rescheduling an existing Job for another iteration.

func (*JobHelper[P, S]) RescheduleWithRetry

func (j *JobHelper[P, S]) RescheduleWithRetry(ctx context.Context, job Job[P, S]) error

RescheduleWithRetry is the same as Reschedule but automatically retries on transient errors.