activejob

package module
v0.0.0-...-62fd193 Latest Latest
Warning

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

 Go to latest Published: Jul 17, 2026 License: BSD-3-Clause Imports: 13 Imported by: 0

README

go-ruby-activejob/activejob

activejob — go-ruby-activejob

Docs License Go Coverage

A pure-Go (no cgo) reimplementation of the foundation of Rails' ActiveJob — the job model, the argument-serialization core, and the queue-adapter framework — faithful to MRI 4.0.5's activejob gem. It mirrors ActiveJob's observable behaviour — perform_later / perform_now / set, queue_as, retry_on / discard_on, enqueue/perform callbacks, and the exact ActiveJob::Arguments wire format (_aj_serialized, _aj_symbol_keys, _aj_hash_with_indifferent_access, _aj_globalid, …) — without any Ruby runtime, so its payloads interoperate byte-for-byte with a real Rails queue.

It is the ActiveJob backend for go-embedded-ruby, but is a standalone, reusable module — a sibling of go-ruby-activesupport and go-ruby-set. The Ruby perform method body, the Ruby class dispatch, and the GlobalID conversion/location are all injectable seams, so a host runtime plugs in its own semantics.

MRI-faithful, not Composition-Oriented. This is the Rails ActiveJob wire format and job lifecycle — reach for it when you want ActiveJob semantics and payload interop.

The seams

Everything Ruby-specific is a plug, so this v0.1 foundation runs without a Ruby VM and a host (go-embedded-ruby) can wire in the real behaviour later:

Seam Type Role
Perform func(jobClass string, args []any) error the Ruby perform method body
Class dispatch *Registry (Register / Lookup) maps a job_class name back to its *Base when deserializing a payload
GlobalID (serialize) Arguments.ToGlobalID func(any) (uri string, ok bool, err error) convert a host object to gid://…
GlobalID (deserialize) Arguments.LocateGlobalID func(uri string) (any, error) resolve gid://… back to a host object (defaults to a GlobalID value)
Queue adapter Adapter interface where enqueued jobs go

Install

go get github.com/go-ruby-activejob/activejob

Usage

package main

import (
	"errors"
	"fmt"
	"time"

	aj "github.com/go-ruby-activejob/activejob"
)

func main() {
	// Define a job class: queue, adapter, the perform seam, and a retry rule.
	var errTransient = errors.New("transient")
	greet := aj.NewBase("GreetJob").
		QueueAs("mailers").
		WithAdapter(aj.InlineAdapter{}).
		WithPerform(func(class string, args []any) error {
			fmt.Printf("%s says hi to %v\n", class, args[0])
			return nil
		}).
		RetryOn(aj.MatchError(errTransient), aj.RetryOptions{Attempts: 3})

	// perform_later — serialize the arguments and enqueue through the adapter.
	_ = greet.New("Ada").PerformLater()

	// set(...).perform_later — configure a single enqueue.
	_ = greet.New("Grace").
		Set(aj.SetOptions{Queue: "urgent", Wait: 5 * 60}).
		PerformLater()

	// The ActiveJob argument wire format, byte-compatible with Rails:
	args := aj.NewArguments()
	b, _ := args.SerializeJSON([]any{
		aj.Symbol("mode"),
		aj.NewHash().Set(aj.Symbol("id"), int64(7)),
	})
	fmt.Println(string(b))
	// [{"_aj_serialized":"ActiveJob::Serializers::SymbolSerializer","value":"mode"},
	//  {"id":7,"_aj_symbol_keys":["id"]}]
}
Testing jobs
ta := &aj.TestAdapter{}
job := aj.NewBase("MyJob").WithAdapter(ta).
	WithPerform(perform).New(1, 2, 3)

_ = job.PerformLater()          // records, does not run
len(ta.EnqueuedJobs())          // 1
_ = ta.PerformEnqueuedJobs()    // drain + run
len(ta.PerformedJobs())         // 1

Argument serialization fidelity

ActiveJob::Arguments.serialize / deserialize are reproduced exactly. Ruby argument types map onto Go values and dedicated wrappers:

Ruby Go Serialized form
nil / true / Integer / Float / String nil / bool / int64 / float64 / string as-is
Symbol Symbol {"_aj_serialized":"…SymbolSerializer","value":…}
BigDecimal BigDecimal {"_aj_serialized":"…BigDecimalSerializer","value":…}
Array []any element-wise
Hash (String/Symbol keys) *Hash values + _aj_symbol_keys
HashWithIndifferentAccess *IndifferentHash values + _aj_hash_with_indifferent_access
Time / Date / DateTime / TimeWithZone Time / Date / DateTime / TimeWithZone ISO-8601(9) wrappers
ActiveSupport::Duration Duration value + serialized parts
Range Range begin / end / exclude_end
Module / Class Module {"_aj_serialized":"…ModuleSerializer","value":…}
GlobalID::Identification GlobalID / seam {"_aj_globalid":"gid://…"}

An *Object is an insertion-ordered, string-keyed map whose MarshalJSON preserves key order, so the emitted bytes match MRI's ActiveJob::Arguments.serialize(args).to_json exactly (Ruby hashes are insertion-ordered; Go maps are not).

Float note. Whole-valued floats render differently in Go and Ruby (3.0 ↔ 3). The oracle uses fractional floats; integral values should be passed as integers, exactly as ActiveJob expects.

v0.1 scope

  • Base / Job — perform_later, perform_now, set(queue:, wait:, wait_until:, priority:), queue_as (static and per-job), and the job_id / queue_name / arguments / executions / exception_executions / enqueued_at / … state.
  • Exceptions — rescue_from (the primitive), plus retry_on / discard_on built on it: attempts (incl. :unlimited), a custom wait proc, a constant WaitSeconds, the :polynomially_longer backoff ((executions⁴) + jitter + 2), class-level and per-rule jitter, per-retry queue / priority overrides, and the faithful exception_executions bucket. Handlers are searched bottom-to-top.
  • Arguments — MRI-exact serialize / deserialize for every argument type above.
  • Adapters — the Adapter interface plus InlineAdapter, TestAdapter (record + PerformEnqueuedJobs), AsyncAdapter (goroutine pool + Drain), a BulkAdapter capability, and the named-adapter registry (RegisterAdapter / LookupAdapter).
  • TestHelper — ActiveJob::TestHelper-style assertions over a TestAdapter: AssertEnqueuedJobs / AssertNoEnqueuedJobs, AssertPerformedJobs / AssertNoPerformedJobs, and AssertEnqueuedWith / AssertPerformedWith (matching job / args / queue / priority / at).
  • Callbacks — before/after/around for enqueue and perform.
  • Registry — job-class dispatch; PerformAllLater bulk enqueue.

Roadmap (deferred)

  • Real Sidekiq / Resque adapter wiring (bridging the payload to go-ruby-sidekiq / go-ruby-resque). The Job.Serialize payload and the BulkAdapter capability are the intended integration points; a future adapter maps our payload onto the backend's job hash.
  • i18n locale propagation beyond the recorded locale field.
  • Instrumentation / Notifications (ActiveSupport::Notifications events), including the after_discard hook and the report: error-reporter option.
  • Block-scoped TestHelper forms (perform_enqueued_jobs / assert_enqueued_with with a block that scopes the assertion to jobs enqueued inside it).
  • Continuations and ruby2_keywords fidelity beyond symbol-key restoration.

Tests & coverage

The suite pairs deterministic, ruby-free tests — which alone hold coverage at 100%, so the qemu cross-arch and Windows lanes pass the gate — with a differential MRI oracle that serializes each argument type here and diffs the bytes against the real activejob gem (installed on the ubuntu/macos CI lanes; the oracle skips itself where ruby or the gem is absent).

COVERPKG=$(go list ./... | paste -sd, -)
go test -race -coverpkg="$COVERPKG" -coverprofile=cover.out ./...
go tool cover -func=cover.out | tail -1   # 100.0%

CGO-free, dependency-free, gofmt + go vet clean, and green across the six 64-bit Go targets (amd64, arm64, riscv64, loong64, ppc64le, s390x) and three OSes (Linux, macOS, Windows).

License

BSD-3-Clause — see LICENSE. Copyright the go-ruby-activejob/activejob authors.

WebAssembly

Being pure Go (CGO=0), this library also compiles to WebAssembly — both GOOS=js GOARCH=wasm (browser / Node.js) and GOOS=wasip1 GOARCH=wasm (WASI). CI builds both targets on every push, alongside the six 64-bit native/qemu arches.

GOOS=js     GOARCH=wasm go build ./...   # browser / Node
GOOS=wasip1 GOARCH=wasm go build ./...   # WASI (wasmtime, wasmer, wasmedge, …)

Documentation

Overview

Package activejob is a pure-Go (no cgo) reimplementation of the foundation of Rails' ActiveJob: the job model, the argument-serialization core, and the queue-adapter framework, faithful to MRI 4.0.5's `activejob` gem.

It mirrors ActiveJob's observable behaviour without any Ruby runtime:

  • Arguments serializes/deserializes job arguments in exactly the wire format the `activejob` gem produces (`_aj_serialized`, `_aj_symbol_keys`, `_aj_hash_with_indifferent_access`, `_aj_globalid`, …), so payloads interoperate byte-for-byte with a real Rails queue.
  • Base models a job class (queue_as, retry_on/discard_on, callbacks) and Job a job instance (perform_later, perform_now, set).
  • Adapter is the queue-adapter interface, with InlineAdapter, TestAdapter and AsyncAdapter implementations plus a Registry.

The Ruby `perform` method body and Ruby class dispatch are injectable seams (PerformFunc and the Registry); the GlobalID conversion/location is a seam on Arguments. This makes the package the ActiveJob backend for a future go-embedded-ruby binding while remaining a standalone, reusable module.

Index

Constants

View Source 
const DefaultJitter = 0.15

DefaultJitter is Rails' config.active_job.retry_jitter default (15%). The bare gem's class attribute defaults to 0.0; call Base.WithRetryJitter to opt in.

Variables

View Source 
var (
	// ErrNoAdapter is returned by perform_later when the job class has no queue
	// adapter configured.
	ErrNoAdapter = errors.New("activejob: no queue adapter configured")
	// ErrNoPerform is returned by perform_now when the job class has no perform
	// seam configured.
	ErrNoPerform = errors.New("activejob: no perform function configured")
)

Functions

func AssertEnqueuedJobs 

func AssertEnqueuedJobs(t TestingT, ta *TestAdapter, n int)

AssertEnqueuedJobs asserts that exactly n jobs are recorded as enqueued on ta, mirroring assert_enqueued_jobs.

func AssertNoEnqueuedJobs 

func AssertNoEnqueuedJobs(t TestingT, ta *TestAdapter)

AssertNoEnqueuedJobs asserts that no jobs are recorded as enqueued on ta, mirroring assert_no_enqueued_jobs.

func AssertNoPerformedJobs 

func AssertNoPerformedJobs(t TestingT, ta *TestAdapter)

AssertNoPerformedJobs asserts that no jobs are recorded as performed on ta, mirroring assert_no_performed_jobs.

func AssertPerformedJobs 

func AssertPerformedJobs(t TestingT, ta *TestAdapter, n int)

AssertPerformedJobs asserts that exactly n jobs are recorded as performed on ta, mirroring assert_performed_jobs.

func MatchArgs 

func MatchArgs(args ...any) []any

MatchArgs returns a non-nil argument slice for JobMatcher.Args, so that MatchArgs() matches a job enqueued with no arguments (distinct from the nil "do not match arguments").

func PerformAllLater 

func PerformAllLater(jobs ...*Job) error

PerformAllLater enqueues several jobs at once (ActiveJob.perform_all_later). When every job shares one adapter that implements BulkAdapter, it enqueues them in a single call; otherwise it enqueues them one by one. It stops at the first error.

func RegisterAdapter 

func RegisterAdapter(name string, factory func() Adapter)

RegisterAdapter registers a named queue-adapter factory (QueueAdapters.register).

Types

type Adapter 

type Adapter interface {
	Enqueue(job *Job) error
	EnqueueAt(job *Job, timestamp time.Time) error
}

Adapter is the queue-adapter interface every backend implements, mirroring ActiveJob::QueueAdapters. Enqueue schedules a job to run as soon as possible; EnqueueAt schedules it to run at (or after) a timestamp.

func LookupAdapter 

func LookupAdapter(name string) (Adapter, bool)

LookupAdapter constructs the adapter registered under name.

type Arguments 

type Arguments struct {
	// ToGlobalID converts a host object to its GlobalID URI. It is consulted
	// for values not matched by any built-in type. Return ok=false to fall
	// through to an "unsupported type" error; return a non-nil err to abort.
	ToGlobalID func(obj any) (uri string, ok bool, err error)

	// LocateGlobalID resolves a GlobalID URI back to a host object during
	// deserialization. When nil, a [GlobalID] value is produced instead.
	LocateGlobalID func(uri string) (any, error)
}

Arguments serializes and deserializes ActiveJob job arguments in MRI's exact wire format. The GlobalID conversion (serialize) and location (deserialize) are injectable seams; when unset, GlobalID values round-trip as GlobalID.

func NewArguments 

func NewArguments() *Arguments

NewArguments returns an Arguments serializer with no seams configured.

func (*Arguments) Deserialize 

func (a *Arguments) Deserialize(args []any) ([]any, error)

Deserialize is the inverse of Serialize. Its input is the parsed-JSON shape (objects as map[string]any or *Object, numbers as json.Number/float64/int).

func (*Arguments) DeserializeJSON 

func (a *Arguments) DeserializeJSON(data []byte) ([]any, error)

DeserializeJSON parses a JSON array of serialized arguments (with UseNumber so integers stay integers) and deserializes it.

func (*Arguments) Serialize 

func (a *Arguments) Serialize(args []any) ([]any, error)

Serialize maps each argument to its JSON-ready wire form. Hash-shaped results are *Object values whose MarshalJSON preserves key order.

func (*Arguments) SerializeJSON 

func (a *Arguments) SerializeJSON(args []any) ([]byte, error)

SerializeJSON serializes args and marshals them to JSON, byte-compatible with MRI's `ActiveJob::Arguments.serialize(args).to_json`.

type AroundFunc 

type AroundFunc func(job *Job, next func() error) error

AroundFunc is an around_* callback. It receives the job and a `next` closure it must invoke to run the wrapped action; it may skip `next` to halt.

type AsyncAdapter 

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

AsyncAdapter performs jobs on background goroutines, mirroring ActiveJob's :async adapter. Call Drain to wait for all in-flight jobs and collect errors. (EnqueueAt runs the job without honouring the delay in this v0.1 foundation.)

func (*AsyncAdapter) Drain 

func (a *AsyncAdapter) Drain() []error

Drain waits for all in-flight jobs to finish and returns any errors they raised, in completion order.

func (*AsyncAdapter) Enqueue 

func (a *AsyncAdapter) Enqueue(job *Job) error

Enqueue performs the job on a new goroutine.

func (*AsyncAdapter) EnqueueAt 

func (a *AsyncAdapter) EnqueueAt(job *Job, _ time.Time) error

EnqueueAt performs the job on a new goroutine, ignoring the timestamp.

type Base 

type Base struct {
	// Name is the job class name recorded in the "job_class" payload field.
	Name string
	// Args is the argument serializer (carrying the GlobalID seams). Never nil.
	Args *Arguments
	// contains filtered or unexported fields
}

Base models an ActiveJob job class: its perform seam, queue adapter, queue name, priority, retry/discard rules and callbacks. Configure it with the chainable builder methods, then create instances with Base.New.

func NewBase 

func NewBase(name string) *Base

NewBase returns a job class named name with the default queue and no adapter.

func (*Base) AfterEnqueue 

func (b *Base) AfterEnqueue(fn CallbackFunc) *Base

AfterEnqueue registers an after_enqueue callback and returns b.

func (*Base) AfterPerform 

func (b *Base) AfterPerform(fn CallbackFunc) *Base

AfterPerform registers an after_perform callback and returns b.

func (*Base) AroundEnqueue 

func (b *Base) AroundEnqueue(fn AroundFunc) *Base

AroundEnqueue registers an around_enqueue callback and returns b.

func (*Base) AroundPerform 

func (b *Base) AroundPerform(fn AroundFunc) *Base

AroundPerform registers an around_perform callback and returns b.

func (*Base) BeforeEnqueue 

func (b *Base) BeforeEnqueue(fn CallbackFunc) *Base

BeforeEnqueue registers a before_enqueue callback and returns b.

func (*Base) BeforePerform 

func (b *Base) BeforePerform(fn CallbackFunc) *Base

BeforePerform registers a before_perform callback and returns b.

func (*Base) DiscardOn 

func (b *Base) DiscardOn(match ErrorMatcher, opts DiscardOptions) *Base

DiscardOn registers a discard_on rule and returns b. A matched error is dropped (optionally via Block), mirroring ActiveJob::Exceptions#discard_on.

func (*Base) New 

func (b *Base) New(args ...any) *Job

New builds a job instance of class b with the given (raw) arguments, a fresh job id, and the class defaults for queue and priority.

func (*Base) QueueAs 

func (b *Base) QueueAs(name string) *Base

QueueAs sets a static queue name (queue_as :name) and returns b.

func (*Base) QueueAsFunc 

func (b *Base) QueueAsFunc(fn func(*Job) string) *Base

QueueAsFunc sets a queue name computed per job at enqueue time (queue_as { … }).

func (*Base) RescueFrom 

func (b *Base) RescueFrom(match ErrorMatcher, handler func(job *Job, err error) error) *Base

RescueFrom registers a rescue_from handler and returns b. When perform raises a matching error, handler runs and its result becomes perform_now's result (return nil to swallow, or re-enqueue via the job to retry), mirroring ActiveSupport::Rescuable#rescue_from — the primitive retry_on / discard_on build on. Handlers are searched bottom-to-top.

func (*Base) RetryOn 

func (b *Base) RetryOn(match ErrorMatcher, opts RetryOptions) *Base

RetryOn registers a retry_on rule and returns b. Errors matched by match are caught and the job is re-enqueued up to Attempts times with the configured backoff, mirroring ActiveJob::Exceptions#retry_on.

func (*Base) WithAdapter 

func (b *Base) WithAdapter(a Adapter) *Base

WithAdapter sets the queue adapter and returns b.

func (*Base) WithPerform 

func (b *Base) WithPerform(fn PerformFunc) *Base

WithPerform sets the perform seam and returns b.

func (*Base) WithPriority 

func (b *Base) WithPriority(p int) *Base

WithPriority sets the default priority and returns b.

func (*Base) WithRetryJitter 

func (b *Base) WithRetryJitter(fraction float64) *Base

WithRetryJitter sets the class-level retry jitter (Rails' retry_jitter, a fraction 0..1 of the computed delay) and returns b. Use DefaultJitter for the Rails app default of 15%.

type BigDecimal 

type BigDecimal string

BigDecimal is a Ruby BigDecimal, carrying its canonical `to_s` text (e.g. "0.15e1"). Serialized/deserialized through the BigDecimalSerializer.

type BulkAdapter 

type BulkAdapter interface {
	EnqueueAll(jobs []*Job) (int, error)
}

BulkAdapter is an optional capability: adapters that can enqueue a batch in one call implement it, and PerformAllLater / Registry will prefer it. EnqueueAll returns the number of jobs successfully enqueued.

type CallbackFunc 

type CallbackFunc func(*Job) error

CallbackFunc is a before_* / after_* callback body. Returning an error halts the chain (mirroring `throw :abort`).

type Date 

type Date struct{ T time.Time }

Date is a Ruby Date (no time component). Serialized as "YYYY-MM-DD".

type DateTime 

type DateTime struct{ T time.Time }

DateTime is a Ruby DateTime. Like Time but always renders a numeric offset ("+00:00" for UTC) rather than "Z", matching Ruby's DateTime#iso8601(9).

type DeserializationError 

type DeserializationError struct{ Cause error }

DeserializationError is raised when a payload cannot be deserialized. It mirrors ActiveJob::DeserializationError and wraps the underlying cause.

func (*DeserializationError) Error 

func (e *DeserializationError) Error() string

func (*DeserializationError) Unwrap 

func (e *DeserializationError) Unwrap() error

type DiscardOptions 

type DiscardOptions struct {
	// Block runs when a matching error is discarded; if nil, the error is swallowed.
	Block func(job *Job, err error) error
}

DiscardOptions configures a discard_on rule.

type Duration 

type Duration struct {
	Value int64
	Parts []DurationPart
}

Duration is a Ruby ActiveSupport::Duration: a total value in seconds plus the ordered parts it was built from.

type DurationPart 

type DurationPart struct {
	Unit   Symbol
	Amount any
}

DurationPart is one component of a Duration (e.g. {minutes: 5}).

type ErrorMatcher 

type ErrorMatcher func(error) bool

ErrorMatcher reports whether a raised error matches a retry_on / discard_on rule. Use MatchError for a errors.Is-based matcher or MatchAny to match all.

func MatchAny 

func MatchAny() ErrorMatcher

MatchAny returns an ErrorMatcher matching every error.

func MatchError 

func MatchError(target error) ErrorMatcher

MatchError returns an ErrorMatcher matching errors that wrap target (errors.Is).

type GlobalID 

type GlobalID struct{ URI string }

GlobalID is a serialized GlobalID reference (`gid://app/Class/id`). It is both an input value (serialize it directly to `{"_aj_globalid": URI}`) and the default output of deserialization when no LocateGlobalID seam is configured, so GlobalID payloads round-trip without a locator.

type Hash 

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

Hash is a Ruby Hash with String and/or Symbol keys, preserving insertion order. Its serialization appends "_aj_symbol_keys" listing which keys were Symbols so deserialization can restore them.

func NewHash 

func NewHash() *Hash

NewHash returns an empty ordered Hash.

func (*Hash) Get 

func (h *Hash) Get(key any) (any, bool)

Get returns the value for key and whether it was present.

func (*Hash) Keys 

func (h *Hash) Keys() []any

Keys returns the keys in insertion order (a copy).

func (*Hash) Len 

func (h *Hash) Len() int

Len returns the number of pairs.

func (*Hash) Set 

func (h *Hash) Set(key, value any) *Hash

Set stores key (a string or Symbol) with value, preserving insertion order. It returns h for chaining.

type IndifferentHash 

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

IndifferentHash is a Ruby ActiveSupport::HashWithIndifferentAccess: all keys are Strings and its serialization carries the "_aj_hash_with_indifferent_access" marker instead of "_aj_symbol_keys".

func NewIndifferentHash 

func NewIndifferentHash() *IndifferentHash

NewIndifferentHash returns an empty ordered IndifferentHash.

func (*IndifferentHash) Get 

func (h *IndifferentHash) Get(key string) (any, bool)

Get returns the value for key and whether it was present.

func (*IndifferentHash) Keys 

func (h *IndifferentHash) Keys() []string

Keys returns the keys in insertion order (a copy).

func (*IndifferentHash) Len 

func (h *IndifferentHash) Len() int

Len returns the number of pairs.

func (*IndifferentHash) Set 

func (h *IndifferentHash) Set(key string, value any) *IndifferentHash

Set stores key with value, preserving insertion order. Returns h for chaining.

type InlineAdapter 

type InlineAdapter struct{}

InlineAdapter performs jobs immediately, on the enqueuing goroutine, mirroring ActiveJob's :inline adapter. EnqueueAt ignores the delay and performs at once.

func (InlineAdapter) Enqueue 

func (InlineAdapter) Enqueue(job *Job) error

Enqueue performs the job immediately.

func (InlineAdapter) EnqueueAt 

func (InlineAdapter) EnqueueAt(job *Job, _ time.Time) error

EnqueueAt performs the job immediately, ignoring the timestamp.

type Job 

type Job struct {
	Base                *Base
	JobID               string
	QueueName           string
	Priority            *int
	Arguments           []any
	Executions          int
	ExceptionExecutions map[string]int
	Locale              string
	Timezone            string
	EnqueuedAt          time.Time
	ScheduledAt         *time.Time
	ProviderJobID       string
}

Job is a job instance: a job class plus its arguments and per-run state.

func AssertEnqueuedWith 

func AssertEnqueuedWith(t TestingT, ta *TestAdapter, m JobMatcher) *Job

AssertEnqueuedWith asserts that a job matching m is recorded as enqueued on ta and returns it (nil on no match), mirroring assert_enqueued_with.

func AssertPerformedWith 

func AssertPerformedWith(t TestingT, ta *TestAdapter, m JobMatcher) *Job

AssertPerformedWith asserts that a job matching m is recorded as performed on ta and returns it (nil on no match), mirroring assert_performed_with.

func (*Job) PerformLater 

func (j *Job) PerformLater() error

PerformLater serializes the arguments and enqueues the job through its adapter (EnqueueAt when scheduled, Enqueue otherwise), running the enqueue callbacks.

func (*Job) PerformNow 

func (j *Job) PerformNow() error

PerformNow runs the job body inline: it increments the execution count, runs the perform callbacks around the perform seam, and applies retry_on/discard_on rules to any error the seam returns.

func (*Job) Serialize 

func (j *Job) Serialize() (*Object, error)

Serialize renders the job's transport payload as an ordered *Object, matching the shape MRI's ActiveJob::Core#serialize produces (job_class, job_id, provider_job_id, queue_name, priority, arguments, executions, exception_executions, locale, timezone, enqueued_at, scheduled_at).

func (*Job) SerializeJSON 

func (j *Job) SerializeJSON() ([]byte, error)

SerializeJSON serializes the job and marshals the payload to JSON.

func (*Job) Set 

func (j *Job) Set(o SetOptions) *Job

Set applies the options to the job (queue, priority, scheduled time) and returns it for chaining before perform_later, mirroring MyJob.set(…).

type JobMatcher 

type JobMatcher struct {
	// Job is the expected job-class name (job:). Empty matches any class.
	Job string
	// Args are the expected raw (pre-serialization) arguments (args:). Nil means
	// "do not match arguments"; a non-nil slice, including an empty one, is
	// compared by their serialized wire form. Use [MatchArgs] to require zero
	// arguments explicitly.
	Args []any
	// Queue is the expected queue name (queue:). Empty matches any queue.
	Queue string
	// Priority is the expected priority (priority:). Nil matches any priority.
	Priority *int
	// At is the expected scheduled time (at:), matched within ±1s. Nil matches
	// any (including unscheduled) job.
	At *time.Time
}

JobMatcher describes the expected attributes of an enqueued or performed job, mirroring the keyword arguments of assert_enqueued_with / assert_performed_with. A zero-valued field is not matched; a non-nil Args slice (even empty) is.

type Module 

type Module string

Module is a Ruby Module or Class reference, carrying its constant name.

type Object 

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

Object is the serialized form of any Ruby Hash-shaped value (a plain Hash, a HashWithIndifferentAccess, or an object routed through a serializer). It is an insertion-ordered, string-keyed map whose MarshalJSON preserves key order, so the JSON bytes match MRI's `ActiveJob::Arguments.serialize(...).to_json` exactly (MRI hashes are insertion-ordered; Go's built-in maps are not).

func NewObject 

func NewObject() *Object

NewObject returns an empty ordered Object.

func (*Object) Get 

func (o *Object) Get(key string) (any, bool)

Get returns the value stored under key and whether it was present.

func (*Object) Has 

func (o *Object) Has(key string) bool

Has reports whether key is present.

func (*Object) Keys 

func (o *Object) Keys() []string

Keys returns the keys in insertion order (a copy).

func (*Object) Len 

func (o *Object) Len() int

Len returns the number of keys.

func (*Object) MarshalJSON 

func (o *Object) MarshalJSON() ([]byte, error)

MarshalJSON renders the object as a JSON object with keys in insertion order.

func (*Object) Set 

func (o *Object) Set(key string, value any) *Object

Set stores key with value, appending the key to the order on first insertion and overwriting the value on subsequent Sets. It returns o for chaining.

type PerformFunc 

type PerformFunc func(jobClass string, args []any) error

PerformFunc is the injectable seam for a Ruby job's `perform` method body. It receives the job class name and its (deserialized) arguments.

type Range 

type Range struct {
	Begin      any
	End        any
	ExcludeEnd bool
}

Range is a Ruby Range. Begin and End are themselves serializable arguments (and may be nil for begin-less / end-less ranges).

type Registry 

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

Registry maps job-class names to their Base, providing the Ruby class dispatch seam used to reconstruct a job from a serialized payload.

func NewRegistry 

func NewRegistry() *Registry

NewRegistry returns an empty registry.

func (*Registry) Deserialize 

func (r *Registry) Deserialize(payload map[string]any) (*Job, error)

Deserialize reconstructs a job instance from a transport payload, dispatching the "job_class" field through the registry to find its Base (the Ruby class dispatch seam). The payload's numbers may be json.Number, float64 or int.

func (*Registry) Lookup 

func (r *Registry) Lookup(name string) (*Base, bool)

Lookup returns the job class registered under name.

func (*Registry) Register 

func (r *Registry) Register(base *Base) *Base

Register adds base under its Name and returns base for chaining.

type RetryOptions 

type RetryOptions struct {
	// Attempts is the maximum number of executions before giving up (Rails
	// default: 5, and the count includes the original run). Ignored when
	// Unlimited is set.
	Attempts int
	// Unlimited retries until the job succeeds (attempts: :unlimited).
	Unlimited bool
	// Wait is a custom delay algorithm (wait: ->(executions) { … }). It receives
	// the current per-exception execution count and its result is used verbatim
	// (jitter is not applied, matching Rails' Proc case). It takes precedence
	// over WaitSeconds / Polynomial.
	Wait func(executions int) time.Duration
	// WaitSeconds is a constant delay in seconds (wait: N.seconds). Jitter is
	// applied. When zero and neither Wait nor Polynomial is set, the Rails
	// default of 3 seconds is used.
	WaitSeconds int
	// Polynomial selects the :polynomially_longer backoff:
	// ((executions**4) + jitter) + 2 seconds (≈3s, 18s, 83s, 258s, …).
	Polynomial bool
	// Jitter overrides the class-level retry jitter for this rule (0..1). Nil
	// uses the job class's [Base.WithRetryJitter] value.
	Jitter *float64
	// Queue re-enqueues the retry on a different queue (queue:).
	Queue string
	// Priority re-enqueues the retry with a different priority (priority:).
	Priority *int
	// Key names the exception_executions bucket incremented on each retry. When
	// empty, [defaultExceptionKey] is used.
	Key string
	// Block runs when attempts are exhausted; if nil, the error is re-raised.
	Block func(job *Job, err error) error
}

RetryOptions configures a retry_on rule, mirroring ActiveJob's retry_on keyword options.

type SerializationError 

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

SerializationError is raised when an argument cannot be serialized (an unsupported type, a reserved or non-string/symbol Hash key, …). It mirrors ActiveJob::SerializationError.

func (*SerializationError) Error 

func (e *SerializationError) Error() string

type SetOptions 

type SetOptions struct {
	Queue     string
	Priority  *int
	Wait      time.Duration
	WaitUntil time.Time
}

SetOptions configures a single enqueue (ActiveJob's `set`).

type Symbol 

type Symbol string

Symbol is a Ruby Symbol (`:name`). It serializes through the SymbolSerializer.

type TestAdapter 

type TestAdapter struct {
	Enqueued  []*Job
	Performed []*Job
	// contains filtered or unexported fields
}

TestAdapter records enqueued jobs instead of running them, mirroring ActiveJob's :test adapter. Call PerformEnqueuedJobs to drain and run them.

func (*TestAdapter) Enqueue 

func (t *TestAdapter) Enqueue(job *Job) error

Enqueue records job as enqueued.

func (*TestAdapter) EnqueueAll 

func (t *TestAdapter) EnqueueAll(jobs []*Job) (int, error)

EnqueueAll records every job as enqueued and reports how many were recorded.

func (*TestAdapter) EnqueueAt 

func (t *TestAdapter) EnqueueAt(job *Job, _ time.Time) error

EnqueueAt records job as enqueued (the scheduled timestamp is on the job).

func (*TestAdapter) EnqueuedJobs 

func (t *TestAdapter) EnqueuedJobs() []*Job

EnqueuedJobs returns a snapshot of the currently enqueued jobs.

func (*TestAdapter) PerformEnqueuedJobs 

func (t *TestAdapter) PerformEnqueuedJobs() error

PerformEnqueuedJobs runs every enqueued job with perform_now, moving each to the performed list. It stops and returns the first error encountered.

func (*TestAdapter) PerformedJobs 

func (t *TestAdapter) PerformedJobs() []*Job

PerformedJobs returns a snapshot of the jobs performed so far.

type TestingT 

type TestingT interface {
	Helper()
	Errorf(format string, args ...any)
}

TestingT is the slice of *testing.T that the assertion helpers need, so they work with any test framework (mirroring ActiveJob::TestHelper's Minitest assertions). A failing assertion calls Errorf, exactly as Minitest's assert records a failure without aborting.

type Time 

type Time struct{ T time.Time }

Time is a Ruby Time. It serializes as ISO-8601 with 9 fractional digits, rendering UTC as a trailing "Z" (matching Ruby's Time#iso8601(9)).

type TimeWithZone 

type TimeWithZone struct {
	T        time.Time
	TimeZone string
}

TimeWithZone is a Ruby ActiveSupport::TimeWithZone: an instant plus the IANA zone name (e.g. "Etc/UTC"). Serialized with a "time_zone" field.

Source Files

View all Source files

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.