queueservice

package module
v1.0.0 Latest Latest
Warning

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

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 14 Imported by: 0

README

Queue service lifecycle adapter

queueservice is the independently versioned lifecycle integration between github.com/faustbrian/go-queue and github.com/faustbrian/go-service. It connects caller-owned producers and workers to service startup, readiness, supervision, drain, and shutdown without choosing a backend or moving retry, scheduling, acknowledgement, redelivery, or dead-letter policy out of queue.

The module follows stable v1 compatibility. Consumers should pin an exact released version.

Quick start

producer, err := queueservice.NewProducer(
	queueservice.ProducerOptions[*queue.Queue]{
		Name:        "orders-producer",
		Resource:    concreteQueue,
		Correlation: correlationFactory,
		Publish: func(
			_ context.Context,
			resource *queue.Queue,
			message core.QueuedMessage,
			options ...job.AllowOption,
		) error {
			return resource.Queue(message, options...)
		},
	},
)
if err != nil {
	return err
}

runtime, err := service.New(service.Config{
	Components: []service.Component{producer.Component()},
})
if err != nil {
	return err
}

Use NewHandler around the application handler before constructing a concrete queue.Worker. Use NewWorker when a *queue.Queue owns scheduling and drain. Use NewLifecycleWorker when a backend exposes explicit startup, readiness, blocking run, and shutdown callbacks that should be supervised as one service plan.

Lifecycle sequence

The adapter lifecycle is monotonic. The following model applies to producers and typed workers; a shared producer is the one documented exception that may return from starting to constructed after startup validation fails because no close ownership was transferred.

State Allowed work Transition and terminal result
constructed one start attempt; stop-before-start start enters starting; stop permanently rejects work
starting the owned startup callback success enters ready; shared failure returns to constructed; owned failure rolls back and becomes terminal
ready publish or one worker run owner, handler delivery, readiness drain or run exit enters draining
draining already admitted calls only the one termination budget joins admitted calls; expiry leaves drain resumable
stopping the owned shutdown callback callback result is cached without another close attempt
stopped repeated stop result inspection terminal; start, readiness, publish, run, and handler admission are rejected

Producer publish results are separately terminal as not accepted, accepted, or unknown. Worker delivery settlement remains a backend-owned terminal result and must never acknowledge an incomplete handler.

Producer
  1. Construction validates the stable name, typed resource, correlation and trace dependencies, and exactly one publish callback without starting work.
  2. Component start runs the optional bounded startup callback and opens admission only after it succeeds.
  3. The optional readiness check runs only while admission is open.
  4. Service drain synchronously closes adapter admission, making readiness and new publishes unavailable before the service cancels supervised work.
  5. Component stop waits for admitted publish and readiness calls under the caller's shutdown context, then runs an owned shutdown callback. Concurrent callers share one attempt, and repeated stops return the same result without invoking resource closure again.

A startup failure closes an owned producer immediately. StartupError keeps both the validation and cleanup causes available to errors.Is and errors.As without including their text in its diagnostic.

Typed worker

NewLifecycleWorker returns a service.Plan fragment with one component, one supervised task, and an optional readiness check under the same name. Its optional CloseAdmission callback synchronously stops backend intake at service drain; it must return promptly and must not wait for handlers.

start resource -> expose readiness -> run intake
remove readiness -> stop intake -> cancel run -> join handlers
-> settle completed work -> release incomplete work -> close resource

The run callback owns intake and backend settlement. It returns only after it has stopped admitting deliveries and joined every handler it admitted. A completed handler may be settled according to backend policy; an incomplete handler is left or released for backend redelivery. The component shutdown callback runs only after the supervised run and concurrent readiness calls return. Its caller-owned shutdown context is the single budget for handler drain and resource closure; the preceding synchronous admission hook must return promptly and must not wait for handlers.

Exactly one run callback may acquire intake ownership. A concurrent or repeated task invocation returns ErrUnavailable without starting another backend read loop.

An unexpected successful run return produces ErrWorkerExited and makes the adapter unavailable. The service runtime converts that result, or any other non-cancellation task failure, into process drain and shutdown. A return after the run context is canceled is a normal shutdown result.

Concrete queue convenience

NewWorker adapts an existing *queue.Queue. Its component calls CloseAdmission during service drain, then ReleaseContext during stop, inheriting the queue module's tested admission withdrawal, handler join, settlement, and redelivery behavior. The same queue must not be released independently after its ownership is transferred to the component. Concrete worker shutdown panics become secret-safe shutdown callback failures; the queue caches the terminal classification so repeated stop calls observe the same result without closing the worker again.

Publish cancellation and duplicate windows

PublishWithAcceptance calls the backend once and returns one of:

Result Meaning Safe caller action
PublishNotAccepted The backend definitely rejected the task Retry according to application policy
PublishAccepted The backend definitely accepted the task Do not retry the same logical task
PublishUnknown Acceptance may have occurred Reconcile or use an idempotency key; do not retry blindly

A context already canceled at adapter admission produces PublishNotAccepted without calling the backend. Once the callback begins, the callback owns cancellation and acceptance classification. The adapter never retries. The compatibility Publish callback cannot classify failures, so any callback error is reported as PublishUnknown and matches ErrPublishOutcomeUnknown.

No queue adapter can make a publish and an application-side effect atomic. Process death after either side commits creates a duplicate or missing-work window. Durable backends, idempotent handlers, transactional outbox patterns, and reconciliation remain application and backend choices.

Readiness, SIGTERM, and scaling

Service drain makes readiness false and invokes each adapter's idempotent admission hook before component stop begins. In Kubernetes, send SIGTERM through service.Run or the platform runtime, configure a termination grace period longer than the service shutdown timeout, and let the readiness endpoint withdraw the pod before the worker drain completes. A preStop sleep is not a substitute for readiness withdrawal and bounded drain.

During scale-down or rolling deployment, durable backends may redeliver work whose lease or acknowledgement was not completed before the deadline. At-most- once backends may lose that work instead. Size handler concurrency and the termination budget so normal work can finish, while keeping every handler's external operations bounded by its context.

The process-termination contract is exercised against real Redis Streams and Valkey Streams at three kill points. Termination before the handler effect leaves unacknowledged work for one replacement; termination after the effect but before settlement redelivers and demonstrates the documented duplicate window; termination after settlement does not replay. Two replacement processes contend for the expired lease, and a separate podintegration suite repeats the same boundaries by force-deleting worker pods in an isolated Kind cluster.

The recovery lease must exceed the longest handler and settlement window. A replacement may correctly reclaim the same delivery again when that lease expires while another replacement is still handling it. The integration suite uses an explicit bounded lease, waits for observable expiry, and proves one effect owner only while work completes inside that lease.

Backend differences

The adapter does not flatten backend guarantees:

  • the in-memory ring exists only in one process and cannot redeliver after process loss;
  • Redis Pub/Sub and NATS Core are transient and may lose disconnected work;
  • Redis Streams and Valkey Streams use pending-entry ownership and reclaim;
  • RabbitMQ uses publisher confirmation plus acknowledgement or rejection;
  • NSQ uses its own finish, requeue, and timeout behavior.

Read the queue module's backend and delivery-semantics documentation before selecting retry, visibility, dead-letter, and shutdown settings. The adapter integration gate composes NewWorker with real Redis Streams and Valkey Streams services. It proves delivery, disconnect and reconnect, handler timeout, lease expiry, redelivery, dead-letter destination failure, shutdown, helper-process termination, competing recovery, scale-up/down, and rolling replacement without expanding verification to unrelated queue packages. make pod-integration adds the literal pod-kill proof and requires QUEUE_SERVICE_POD_CONTEXT to name a kind-queueservice-* context plus the preloaded test, Redis, and Valkey image environment variables.

API and ownership

  • NewProducer retains a typed producer and optional startup, readiness, and shutdown callbacks. Omitting shutdown keeps the resource shared.
  • Producer.Component provides ordered startup, admission closure, and drain.
  • Producer.Readiness returns an opt-in service.ReadinessCheck.
  • Producer.PublishWithAcceptance exposes safe retry information; Producer.Publish is the compatibility surface.
  • NewHandler establishes a fresh delivery-attempt request ID and optional caller-owned trace extraction.
  • NewLifecycleWorker retains a typed worker and returns a supervised plan. Its optional CloseAdmission callback owns backend intake withdrawal.
  • NewWorker is the focused *queue.Queue convenience adapter.

Names are valid UTF-8, non-blank, and at most MaxNameBytes. Callback errors become CallbackError, preserving their causes without formatting their text. Callback panics become CallbackPanicError; panic values are not retained or formatted. Backend and application errors remain available for programmatic inspection, but the module does not log them, task payloads, credentials, or endpoints.

Adoption and migration

For a new service:

  1. construct and validate the concrete backend client;
  2. wrap the application handler with NewHandler or configure it through NewLifecycleWorker;
  3. add the worker plan before dependent producer components so reverse shutdown drains producers first;
  4. add only backend checks that are required to accept new work;
  5. run the backend integration suite for the selected transport.

Existing NewWorker(WorkerOptions{Name, Queue}) users remain compatible. To supervise a backend run loop or surface worker exit, migrate to NewLifecycleWorker, move fallible validation into Startup, move the blocking intake loop into Run, and move final transport closure into Shutdown. Existing producer callbacks remain valid; migrate to PublishWithAcceptance when the backend can classify definite rejection and ambiguous acceptance.

Security

Inbound correlation metadata is untrusted by default. Enable TrustedMetadata only after authenticating the immediate queue boundary. Trace propagation is disabled unless a caller supplies an explicit OpenTelemetry propagator. Metadata is bounded and cloned before transport use, so caller-owned maps and timestamps are not aliased.

FAQ

Does the adapter retry publish or handler work?

No. Publish retry belongs to the application and concrete producer. Handler retry, scheduling, dead-letter, and settlement belong to queue and its backend.

What happens when shutdown times out?

The stop call returns the context cause without closing a resource still used by an admitted callback. A later stop can resume drain. Once resource shutdown begins, its result is cached so repeated signals cannot double-close the resource. Work not completed and settled before process termination follows the concrete backend's redelivery or loss semantics.

Should readiness prove that every broker operation will succeed?

No. Readiness should answer only whether this process may accept new work. Use a bounded backend check when that dependency is required; omit it for roles that can remain ready while a transient dependency recovers.

How is callback failure text handled?

The adapter preserves errors for errors.Is and errors.As and emits no logs. Its callback, aggregate, and panic errors use bounded diagnostics that do not include callback error text or panic values. Applications remain responsible for a safe error-rendering policy outside the adapter boundary.

Development

make check

The module contract includes formatting, vet, unit tests, race detection, exact statement coverage, fuzzing, allocation-reporting benchmarks, documentation examples, and package-scoped backend integration. Repository gates add API compatibility, mutation, static analysis, vulnerability, secret, license, and SBOM checks without changing this module's local test scope.

BenchmarkProducerPublish and BenchmarkHandlerDelivery isolate adapter hot path overhead. BenchmarkProducerDrain and BenchmarkLifecycleWorkerDrain measure lifecycle drain separately from broker latency and handler work.

Ecosystem

Use the Golib documentation portal to choose companion packages, supported stacks, recipes, and operations guidance.

Documentation

Overview

Package queueservice composes explicit queue producers and workers with the service lifecycle and the existing correlation queue semantics.

Producer resources remain concrete and caller-visible. A non-nil Shutdown transfers close ownership; shared resources omit it. Service shutdown first rejects new publishes, waits for active publishes within the service context, and only then closes an owned transport. PublishWithAcceptance exposes definite rejection, confirmed acceptance, and ambiguous acceptance; the adapter never retries an application publish.

LifecycleWorker adapts typed startup, readiness, blocking run, handler, and shutdown callbacks into one service plan. Worker retains the smaller *queue.Queue convenience boundary. Both paths stop intake, join admitted work within the service context, and release the concrete transport only after the drain. NewHandler creates a new request ID for every delivery. TrustedMetadata must be enabled only after authenticating the immediate queue boundary. Callback failures preserve their causes without formatting their text, and callback panics are returned without retaining their values.

Index

Examples

Constants

View Source 
const MaxNameBytes = 128

MaxNameBytes bounds component, task, and readiness identifiers.

Variables

View Source 
var (
	// ErrInvalidOptions identifies invalid adapter construction.
	ErrInvalidOptions = errors.New("invalid queue service options")
	// ErrUnavailable reports an inactive or draining adapter.
	ErrUnavailable = errors.New("queue service adapter unavailable")
	// ErrMissingCorrelation reports a publish without an explicit parent
	// workflow. Callers beginning new work must start it with their factory.
	ErrMissingCorrelation = errors.New("queue service producer correlation missing")
	// ErrCallbackPanic reports a recovered application callback panic.
	ErrCallbackPanic = errors.New("queue service callback panicked")
	// ErrPublishOutcomeUnknown reports a publish that may have reached the
	// backend and therefore must not be retried blindly.
	ErrPublishOutcomeUnknown = errors.New("queue service publish outcome unknown")
	// ErrInvalidPublishAcceptance reports a callback result outside the public
	// acceptance contract.
	ErrInvalidPublishAcceptance = errors.New("queue service publish acceptance invalid")
	// ErrWorkerExited reports a worker run callback that returned successfully
	// before its context was canceled.
	ErrWorkerExited = errors.New("queue service worker exited unexpectedly")
)

Functions

This section is empty.

Types

type CallbackError 

type CallbackError struct {
	// Operation identifies the callback boundary.
	Operation CallbackOperation
	// Err is the original callback failure.
	Err error
}

CallbackError preserves a callback failure for errors.Is and errors.As without formatting potentially sensitive backend or application text.

func (*CallbackError) Error 

func (err *CallbackError) Error() string

Error returns a secret-safe callback failure.

func (*CallbackError) Unwrap 

func (err *CallbackError) Unwrap() error

Unwrap preserves the callback cause for errors.Is and errors.As.

type CallbackOperation 

type CallbackOperation uint8

CallbackOperation identifies one application callback boundary. 

const (
	// CallbackStartup identifies resource validation during service start.
	CallbackStartup CallbackOperation = 1
	// CallbackReadiness identifies an opt-in dependency readiness check.
	CallbackReadiness CallbackOperation = 2
	// CallbackPublish identifies concrete producer publication.
	CallbackPublish CallbackOperation = 3
	// CallbackHandler identifies application task handling.
	CallbackHandler CallbackOperation = 4
	// CallbackRun identifies supervised worker intake.
	CallbackRun CallbackOperation = 5
	// CallbackShutdown identifies transferred resource cleanup.
	CallbackShutdown CallbackOperation = 6
	// CallbackAdmission identifies synchronous worker intake closure.
	CallbackAdmission CallbackOperation = 7
)

type CallbackPanicError 

type CallbackPanicError struct {
	// Operation identifies the callback boundary.
	Operation CallbackOperation
}

CallbackPanicError identifies a recovered callback without retaining or formatting the panic value.

func (*CallbackPanicError) Error 

func (err *CallbackPanicError) Error() string

Error returns a secret-safe callback failure.

func (*CallbackPanicError) Unwrap 

func (err *CallbackPanicError) Unwrap() error

Unwrap exposes the stable panic classification.

type Check 

type Check[R any] func(context.Context, R) error

Check evaluates whether a resource can accept new work.

type CloseAdmission 

type CloseAdmission[R any] func(R) error

CloseAdmission synchronously and idempotently stops new worker intake. It must return promptly and must not wait for admitted handlers to finish.

type Handler 

type Handler func(context.Context, core.TaskMessage) error

Handler is the queue worker handler signature.

func NewHandler 

func NewHandler(options HandlerOptions) (Handler, error)

NewHandler wraps application work with the existing queue receive boundary.

type HandlerOptions 

type HandlerOptions struct {
	// Correlation creates a new request ID for every delivery attempt.
	Correlation *correlation.Factory
	// CorrelationOptions configure the existing queue propagation adapter.
	CorrelationOptions queuecorrelation.Options
	// TrustedMetadata preserves inbound correlation only when explicitly true.
	TrustedMetadata bool
	// TracePropagator explicitly extracts caller-owned telemetry context when
	// configured. Nil disables trace propagation.
	TracePropagator propagation.TextMapPropagator
	// Handler performs application-owned work.
	Handler Handler
}

HandlerOptions configure a correlation-aware delivery boundary.

type LifecycleWorker 

type LifecycleWorker[R any] struct {
	// contains filtered or unexported fields
}

LifecycleWorker retains a concrete worker and explicit lifecycle callbacks.

func NewLifecycleWorker 

func NewLifecycleWorker[R any](
	options LifecycleWorkerOptions[R],
) (*LifecycleWorker[R], error)

NewLifecycleWorker validates and constructs an inert typed worker adapter.

Example 
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/faustbrian/go-correlation"
	"github.com/faustbrian/go-queue/core"
	"github.com/faustbrian/go-queue/queueservice"
)

type typedWorkerResource struct{}

type typedDelivery string

func (delivery typedDelivery) Bytes() []byte   { return []byte(delivery) }
func (delivery typedDelivery) Payload() []byte { return []byte(delivery) }

func main() {
	factory, _ := correlation.NewFactory(correlation.FactoryOptions{})
	worker, err := queueservice.NewLifecycleWorker(
		queueservice.LifecycleWorkerOptions[*typedWorkerResource]{
			Name:        "orders-worker",
			Resource:    &typedWorkerResource{},
			Correlation: factory,
			Handler: func(_ context.Context, task core.TaskMessage) error {
				fmt.Println(string(task.Payload()))

				return nil
			},
			Run: func(
				ctx context.Context,
				_ *typedWorkerResource,
				handler queueservice.Handler,
			) error {
				return handler(ctx, typedDelivery("delivery"))
			},
			Shutdown: func(context.Context, *typedWorkerResource) error {
				return nil
			},
		},
	)
	if err != nil {
		return
	}
	plan := worker.Plan()
	if err = plan.Components[0].Start(context.Background()); err != nil {
		return
	}
	if err = plan.Tasks[0].Run(context.Background()); !errors.Is(err, queueservice.ErrWorkerExited) {
		return
	}
	stopContext, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
	if err = plan.Components[0].Stop(stopContext); err != nil {
		return
	}

}

Output:
delivery

func (*LifecycleWorker[R]) Plan 

func (worker *LifecycleWorker[R]) Plan() service.Plan

Plan returns the worker component, supervised run task, and optional readiness check under one stable identity.

func (*LifecycleWorker[R]) Resource 

func (worker *LifecycleWorker[R]) Resource() R

Resource returns the exact caller-provided worker resource.

type LifecycleWorkerOptions 

type LifecycleWorkerOptions[R any] struct {
	// Name is the secret-safe component, task, and readiness name.
	Name string
	// Resource is the caller-constructed concrete worker.
	Resource R
	// Correlation creates a new request ID for every delivery attempt.
	Correlation *correlation.Factory
	// CorrelationOptions configure the existing queue propagation adapter.
	CorrelationOptions queuecorrelation.Options
	// TrustedMetadata preserves inbound correlation only when explicitly true.
	TrustedMetadata bool
	// TracePropagator explicitly extracts caller-owned telemetry context when
	// configured. Nil disables trace propagation.
	TracePropagator propagation.TextMapPropagator
	// Handler performs application-owned work.
	Handler Handler
	// Startup optionally validates Resource before intake can run.
	Startup Startup[R]
	// Readiness optionally checks Resource after successful startup.
	Readiness Check[R]
	// CloseAdmission stops backend intake when service drain begins. The adapter
	// independently rejects handler calls after the callback starts.
	CloseAdmission CloseAdmission[R]
	// Run starts intake and joins admitted handlers before returning.
	Run Run[R]
	// Shutdown stops remaining transport work and closes Resource exactly once.
	Shutdown Shutdown[R]
}

LifecycleWorkerOptions configure one typed, supervised worker lifecycle.

type OptionsError 

type OptionsError struct {
	// Field identifies the rejected option.
	Field string
	// Reason describes the safe failure category.
	Reason string
}

OptionsError identifies one rejected option.

func (*OptionsError) Error 

func (err *OptionsError) Error() string

Error returns a secret-safe construction diagnostic.

func (*OptionsError) Unwrap 

func (err *OptionsError) Unwrap() error

Unwrap exposes the stable option classification.

type Producer 

type Producer[R any] struct {
	// contains filtered or unexported fields
}

Producer retains a concrete producer and coordinates its in-flight calls.

func NewProducer 

func NewProducer[R any](options ProducerOptions[R]) (*Producer[R], error)

NewProducer validates and constructs an inert producer adapter.

func (*Producer[R]) Component 

func (producer *Producer[R]) Component() service.Component

Component returns the producer's ordered service lifecycle component.

func (*Producer[R]) Publish 

func (producer *Producer[R]) Publish(
	ctx context.Context,
	message core.QueuedMessage,
	options ...job.AllowOption,
) (correlation.Values, error)

Publish creates a message hop, attaches its carrier to cloned job metadata, and invokes the concrete producer with that child correlation in its context. Correlation values are also returned for caller-owned logging and telemetry.

func (*Producer[R]) PublishWithAcceptance 

func (producer *Producer[R]) PublishWithAcceptance(
	ctx context.Context,
	message core.QueuedMessage,
	options ...job.AllowOption,
) (correlation.Values, PublishAcceptance, error)

PublishWithAcceptance creates a message hop and reports whether the concrete backend accepted the task. The adapter performs exactly one callback call and never retries an unknown result.

func (*Producer[R]) Readiness 

func (producer *Producer[R]) Readiness() (service.ReadinessCheck, bool)

Readiness returns the configured opt-in dependency check.

func (*Producer[R]) Resource 

func (producer *Producer[R]) Resource() R

Resource returns the exact caller-provided producer.

type ProducerOptions 

type ProducerOptions[R any] struct {
	// Name is the secret-safe component name.
	Name string
	// Resource is the caller-constructed concrete producer.
	Resource R
	// Correlation creates message-hop identifiers.
	Correlation *correlation.Factory
	// CorrelationOptions configure the existing queue propagation adapter.
	CorrelationOptions queuecorrelation.Options
	// TracePropagator explicitly injects caller-owned telemetry context when
	// configured. Nil disables trace propagation.
	TracePropagator propagation.TextMapPropagator
	// Startup optionally validates Resource before admission begins.
	Startup Startup[R]
	// Readiness optionally checks Resource after successful startup.
	Readiness Check[R]
	// Publish performs one concrete, caller-bounded append. A returned error has
	// unknown backend acceptance. Prefer PublishWithAcceptance when the backend
	// can distinguish a definite rejection from an ambiguous result.
	Publish Publish[R]
	// PublishWithAcceptance performs one concrete append with an explicit
	// backend-acceptance result. Exactly one publish callback is required.
	PublishWithAcceptance PublishWithAcceptance[R]
	// Shutdown transfers transport close ownership when non-nil.
	Shutdown Shutdown[R]
}

ProducerOptions configure one producer lifecycle adapter.

type Publish 

type Publish[R any] func(
	context.Context,
	R,
	core.QueuedMessage,
	...job.AllowOption,
) error

Publish appends one correlation-aware message through a concrete resource.

type PublishAcceptance 

type PublishAcceptance uint8

PublishAcceptance reports whether a failed publish reached the backend. 

const (
	// PublishNotAccepted means the backend definitively did not accept the task.
	PublishNotAccepted PublishAcceptance = 1
	// PublishAccepted means the backend definitively accepted the task.
	PublishAccepted PublishAcceptance = 2
	// PublishUnknown means the backend may have accepted the task. Applications
	// must reconcile or rely on idempotency instead of retrying blindly.
	PublishUnknown PublishAcceptance = 3
)

type PublishError 

type PublishError struct {
	// Acceptance describes whether the task reached the backend.
	Acceptance PublishAcceptance
	// Err is the original classifiable failure.
	Err error
}

PublishError preserves the backend cause and acceptance classification without formatting potentially sensitive backend details.

func (*PublishError) Error 

func (err *PublishError) Error() string

Error returns a secret-safe publish diagnostic.

func (*PublishError) Unwrap 

func (err *PublishError) Unwrap() error

Unwrap preserves the backend and stable acceptance causes.

type PublishWithAcceptance 

type PublishWithAcceptance[R any] func(
	context.Context,
	R,
	core.QueuedMessage,
	...job.AllowOption,
) (PublishAcceptance, error)

PublishWithAcceptance appends one task and reports whether a failure reached the backend. It must not retry application work internally.

type Run 

type Run[R any] func(context.Context, R, Handler) error

Run owns worker intake until cancellation or backend failure and must join every handler it admits before returning.

type Shutdown 

type Shutdown[R any] func(context.Context, R) error

Shutdown releases an explicitly transferred producer resource.

type Startup 

type Startup[R any] func(context.Context, R) error

Startup validates an explicitly constructed resource before admission.

type StartupError 

type StartupError struct {
	// Validation is the startup-check failure.
	Validation error
	// Cleanup is an optional transferred-resource shutdown failure.
	Cleanup error
}

StartupError preserves validation and partial-cleanup failures without formatting either potentially sensitive cause.

func (*StartupError) Error 

func (err *StartupError) Error() string

Error returns a secret-safe startup diagnostic.

func (*StartupError) Unwrap 

func (err *StartupError) Unwrap() []error

Unwrap preserves both causes for errors.Is and errors.As.

type Worker 

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

Worker retains one concrete queue.

func NewWorker 

func NewWorker(options WorkerOptions) (*Worker, error)

NewWorker validates and constructs an inert worker lifecycle adapter.

Example 
package main

import (
	"context"
	"fmt"
	"time"

	"github.com/faustbrian/go-correlation"
	queue "github.com/faustbrian/go-queue"
	"github.com/faustbrian/go-queue/core"
	"github.com/faustbrian/go-queue/job"
	"github.com/faustbrian/go-queue/queueservice"
	"go.opentelemetry.io/otel/propagation"
)

type payload string

func (value payload) Bytes() []byte { return []byte(value) }

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

	factory, _ := correlation.NewFactory(correlation.FactoryOptions{})
	handled := make(chan string, 1)
	handler, err := queueservice.NewHandler(queueservice.HandlerOptions{
		Correlation:     factory,
		TrustedMetadata: true,
		TracePropagator: propagation.TraceContext{},
		Handler: func(_ context.Context, task core.TaskMessage) error {
			handled <- string(task.Payload())

			return nil
		},
	})
	if err != nil {
		return
	}
	ring := queue.NewRing(queue.WithFn(handler))
	concrete, err := queue.NewQueue(
		queue.WithWorker(ring),
		queue.WithWorkerCount(1),
	)
	if err != nil {
		return
	}
	worker, err := queueservice.NewWorker(queueservice.WorkerOptions{
		Name: "jobs-worker", Queue: concrete,
	})
	if err != nil {
		return
	}
	producer, err := queueservice.NewProducer(
		queueservice.ProducerOptions[*queue.Queue]{
			Name: "jobs-producer", Resource: concrete, Correlation: factory,
			TracePropagator: propagation.TraceContext{},
			Publish: func(
				_ context.Context,
				resource *queue.Queue,
				message core.QueuedMessage,
				options ...job.AllowOption,
			) error {
				return resource.Queue(message, options...)
			},
		},
	)
	if err != nil {
		return
	}

	workerComponent := worker.Component()
	producerComponent := producer.Component()
	if err = workerComponent.Start(ctx); err != nil {
		return
	}
	if err = producerComponent.Start(ctx); err != nil {
		return
	}
	parent, _ := factory.Start()
	if _, err = producer.Publish(
		correlation.WithValues(ctx, parent),
		payload("delivery"),
	); err != nil {
		return
	}
	select {
	case value := <-handled:
		fmt.Println(value)
	case <-ctx.Done():
		return
	}
	if err = producerComponent.Stop(ctx); err != nil {
		return
	}
	if err = workerComponent.Stop(ctx); err != nil {
		return
	}

}

Output:
delivery

func (*Worker) Component 

func (worker *Worker) Component() service.Component

Component starts the queue and gracefully releases it during service stop.

func (*Worker) Queue 

func (worker *Worker) Queue() *queue.Queue

Queue returns the exact caller-provided queue.

type WorkerOptions 

type WorkerOptions struct {
	// Name is the secret-safe component name.
	Name string
	// Queue is the caller-constructed concrete queue.
	Queue *queue.Queue
}

WorkerOptions configure one concrete queue worker lifecycle.

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.