spooler

package module
v0.2.1 Latest Latest
Warning

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

 Go to latest Published: Oct 22, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README

spooler-go

CI Go Reference

The Go client for Spooler message queue.

Installation

go get github.com/spoolersh/spooler-go

Usage

Run Spooler locally:

docker run --rm -p 127.0.0.1:8080:8080 spoolersh/memspoold

The image is an evaluation build: nothing persists, only the default spool exists, and the conditions it never produces are listed on docs.spooler.sh/local.

The example below is in example_test.go, so it compiles with the package.

Create the client
c := spooler.Client{
	InsecureDisableTLS: true,     // The local image runs http only, so no TLS for it.
	Host:       "127.0.0.1:8080", // Address you bound the image to.
	APIKey:     "local",          // Any API key would work with the local image.
}
Create your first queue
_, err := c.CreateQueue(ctx, spooler.CreateQueueRequest{
	Spool: "default",
	Queue: "greetings",
})
if err != nil && !errors.Is(err, spooler.ErrQueueExists) {
	// Handle error.
}
Send your first message
_, err = c.Send(ctx, spooler.SendRequest{
	Spool: "default",
	Queue: "greetings",
	Data:  []byte("Hello, Spooler!"),
})
if err != nil {
	// Handle error.
}
Receive the message
msg, ok, err := c.Recv(ctx, spooler.RecvRequest{
	Spool: "default",
	Queue: "greetings",
})
if err != nil {
	log.Fatal(err)
}
if !ok {
    // The wait elapsed with nothing to receive (someone else received it!).
}
fmt.Printf("%s\n", msg.Data)
Acknowledge the message
err = c.Ack(ctx, spooler.AckRequest{
	Spool: "default",
	Lease: msg.Lease,
})
if err != nil {
	// Handle error.
}

Against the production API, leave Host and InsecureDisableTLS at their zero values (Host defaults to "api.spooler.sh") and set a real APIKey.

Errors

Every error the API answers is one type, *spooler.Error. Match its kind with errors.Is against a sentinel, and read the details a kind carries with errors.As:

_, err := c.Queue(ctx, spooler.QueueRequest{
    Spool: "default", 
    Queue: "missing",
})
switch {
case errors.Is(err, spooler.ErrQueueNotFound):
	var detail *spooler.QueueNotFoundError
	if errors.As(err, &detail) {
		fmt.Println("no queue named", detail.Queue)
	}

case errors.Is(err, spooler.ErrRateLimited):
	var detail *spooler.RateLimitedError
	if errors.As(err, &detail) {
		fmt.Println("retry after", detail.RetryAfter)
	}

case err != nil:
    // Handle error.
}

The error text is the server's message and is human readable; a program acts on the kind and the details. The kinds, what each means and which are worth retrying are on docs.spooler.sh/errors.

A request the client refuses before sending, such as one missing a required field, matches spooler.ErrInvalidRequest.

Retries

The client retries transient errors on its own, waiting out a Retry-After when the server sends one, but only failures that changed nothing. Client.Attempts sets how many tries an operation gets; the default is five. Put a deadline on the context to cap the wait. An application with a retry policy of its own can set Attempts: 1.

When an operation may have changed state anyway (a 5xx other than 503 and 507, or no answer once the request may have reached the server) the error matches spooler.ErrResultUnknown. The client retries such a result only when the request sets RetryUnknown: for a send, accepting that the message may be appended twice, as a new message with its own id; an ack-and-send retried that way never appends twice. What a retry does is on docs.spooler.sh/errors.

Testing

The integration tests run against the local image when SPOOLER_HOST and SPOOLER_API_KEY are set and skip otherwise. The variables are the same as spooler-cli's:

SPOOLER_HOST=localhost:8080 SPOOLER_API_KEY=local go test ./...

Documentation

License

Apache 2.0. Copyright 2026 Sergey Kamardin; see NOTICE.

Documentation

Overview

Package spooler is the Go client for Spooler, a managed message queue.

Client.Send sends a message, Client.Recv receives and leases it. Client.Renew renews the lease. Client.Ack, Client.Nack, Client.Release, Client.Fail and Client.Discard settle the message.

More info is on https://docs.spooler.sh/delivery.

Example

The send, receive, ack loop against the local evaluation image, which any key authenticates. Run it with: 

docker run --rm -p 127.0.0.1:8080:8080 spoolersh/memspoold

package main

import (
	"context"
	"errors"
	"fmt"
	"log"

	"github.com/spoolersh/spooler-go"
)

func main() {
	ctx := context.Background()
	c := spooler.Client{
		Host:               "localhost:8080",
		InsecureDisableTLS: true,
		APIKey:             "local",
	}

	_, err := c.CreateQueue(ctx, spooler.CreateQueueRequest{
		Spool: "default",
		Queue: "greetings_and_salutations",
	})
	if err != nil && !errors.Is(err, spooler.ErrQueueExists) {
		log.Fatal(err)
	}

	_, err = c.Send(ctx, spooler.SendRequest{
		Spool: "default",
		Queue: "greetings_and_salutations",
		Data:  []byte("hello"),
	})
	if err != nil {
		log.Fatal(err)
	}

	msg, ok, err := c.Recv(ctx, spooler.RecvRequest{
		Spool: "default",
		Queue: "greetings_and_salutations",
	})
	if err != nil {
		log.Fatal(err)
	}
	if !ok {
		fmt.Println("nothing to receive")
		return
	}
	fmt.Printf("%s\n", msg.Data)

	err = c.Ack(ctx, spooler.AckRequest{
		Spool: "default",
		Lease: msg.Lease,
	})
	if err != nil {
		log.Fatal(err)
	}
}

Index

Examples

Constants

View Source 
const DefaultAttempts = 5

DefaultAttempts is how many times an operation is tried when Client.Attempts is zero: the first request and four retries.

View Source 
const DefaultHost = "api.spooler.sh"

DefaultHost is the public API's host, used when Client.Host is empty.

View Source 
const Version = "v0.2.1"

Version is the SDK release, a semantic version with the v prefix; the user agent carries it without the v. A release bumps it and tags the commit with the same value; the release workflow checks that the tag and the constant agree.

Variables

View Source 
var (
	ErrInvalidToken         = &KindError{Kind: ErrorKindInvalidToken}
	ErrDedupDisabled        = &KindError{Kind: ErrorKindDedupDisabled}
	ErrUnknownHeader        = &KindError{Kind: ErrorKindUnknownHeader}
	ErrRetentionLimit       = &KindError{Kind: ErrorKindRetentionLimit}
	ErrQueueLimit           = &KindError{Kind: ErrorKindQueueLimit}
	ErrQueueNotFound        = &KindError{Kind: ErrorKindQueueNotFound}
	ErrQueueReplaced        = &KindError{Kind: ErrorKindQueueReplaced}
	ErrSpoolNotFound        = &KindError{Kind: ErrorKindSpoolNotFound}
	ErrQueueExists          = &KindError{Kind: ErrorKindQueueExists}
	ErrQueueCreating        = &KindError{Kind: ErrorKindQueueCreating}
	ErrQueueDeleting        = &KindError{Kind: ErrorKindQueueDeleting}
	ErrQueueBusy            = &KindError{Kind: ErrorKindQueueBusy}
	ErrNotFailed            = &KindError{Kind: ErrorKindNotFailed}
	ErrDedupInFlight        = &KindError{Kind: ErrorKindDedupInFlight}
	ErrDedupClaimed         = &KindError{Kind: ErrorKindDedupClaimed}
	ErrStaleToken           = &KindError{Kind: ErrorKindStaleToken}
	ErrPayloadTooLarge      = &KindError{Kind: ErrorKindPayloadTooLarge}
	ErrRateLimited          = &KindError{Kind: ErrorKindRateLimited}
	ErrOperationUnconfirmed = &KindError{Kind: ErrorKindOperationUnconfirmed}
	ErrSpoolFull            = &KindError{Kind: ErrorKindSpoolFull}
	ErrUnauthenticated      = &KindError{Kind: ErrorKindUnauthenticated}
	ErrAccountSuspended     = &KindError{Kind: ErrorKindAccountSuspended}
	ErrAccountBlocked       = &KindError{Kind: ErrorKindAccountBlocked}
)

The sentinels, one per ErrorKind, for errors.Is. See the kind for what each means.

View Source 
var ErrInvalidRequest = errors.New("invalid request")

ErrInvalidRequest is what every request the SDK refuses before sending. Matches with errors.Is: a required field missing, or two that exclude each other both set. The text says which.

View Source 
var ErrResultUnknown = errors.New("result unknown")

ErrResultUnknown matches a failure that leaves the result unknown: an answer the docs class as "may have changed", or no answer once the request may have reached the server. For an operation that changes state, the change may have happened anyway; what a retry then does depends on the operation, see "Retrying a change" at https://docs.spooler.sh/errors.

It describes this call, the retries Client.Attempts allows included: it matches if any attempt left the result unknown, whatever a later attempt answered. The client retries such an attempt only when the request sets RetryUnknown (SendRequest, AckAndSendRequest); otherwise whether to retry is the caller's decision.

View Source 
var ErrUnavailable = errors.New("unavailable")

ErrUnavailable matches a spool momentarily unavailable: a 503 that names no kind this SDK knows, per https://docs.spooler.sh/errors. It is no kind, since a gateway in front of the API can answer it too. The attempt that answered it changed nothing; an earlier attempt of the same call may have, which ErrResultUnknown reports, so check that first.

Functions

func Messages 

func Messages(ctx context.Context, c *Client, req MessagesRequest) iter.Seq2[Message, error]

Messages returns an iterator over a queue's messages across every page, where Client.Messages returns one.

func Queues 

func Queues(ctx context.Context, c *Client, req QueuesRequest) iter.Seq2[Queue, error]

Queues returns an iterator over a spool's queues across every page, where Client.Queues returns one.

Types

type AckAndSendRequest 

type AckAndSendRequest struct {
	Spool string

	// Lease is the delivery to ack. It names the queue it came from.
	Lease Lease

	// Queue is the queue to append the new message to.
	Queue string

	// DelaySeconds is an optional delay until the new message becomes
	// visible.
	//
	// Zero means visible at once.
	DelaySeconds int

	// Dedup is an optional deduplication key for the new message.
	Dedup Dedup

	// Data is the new message's optional payload.
	//
	// Zero sends an empty message.
	Data []byte

	// DataSource is an optional stream to send as the payload instead of Data.
	// Setting both is an error.
	DataSource DataSource

	// ReturnHandle is an optional request for a handle to the new message.
	//
	// A handle is a credential, so it is off by default.
	ReturnHandle bool

	// RetryUnknown is an optional request to retry an attempt whose result is
	// unknown, within the attempts [Client.Attempts] allows. Unlike
	// [SendRequest.RetryUnknown], such a retry never appends twice, with or
	// without a dedup key: one after a call that committed fails with
	// [ErrStaleToken] and appends nothing. The error then also matches
	// [ErrResultUnknown], since that answer does not prove the first attempt
	// succeeded; check it first. A payload from [DataSourceReader] is still
	// sent once.
	//
	// Zero retries no attempt whose result is unknown.
	RetryUnknown bool
}

AckAndSendRequest describes a delivery to ack and the message to append, both in a single atomic operation.

type AckAndSendResult 

type AckAndSendResult struct {
	// ID is the new message's id.
	ID string

	// Handle is the handle to the new message, when the request asked for one.
	Handle Handle
}

AckAndSendResult is the outcome of an ack-and-send.

type AckRequest 

type AckRequest struct {
	Spool string
	Lease Lease
}

AckRequest describes a delivery to ack.

type Client 

type Client struct {
	// Trace is an optional set of runtime hooks the client calls.
	Trace ClientTrace

	// Host specifies the API host, as "host" or "host:port".
	//
	// Zero means [DefaultHost].
	Host string

	// APIKey authenticates every request made by this Client.
	// Required.
	APIKey string

	// Attempts is how many times at most an operation is tried before its
	// error is returned, the initial request included. Only a failure the
	// docs class as transient and as having changed nothing is retried; one
	// that leaves the result unknown is returned at once, matching
	// [ErrResultUnknown], unless the request sets RetryUnknown, as
	// [SendRequest] and [AckAndSendRequest] can. An application with a retry
	// policy of its own can set 1.
	//
	// Zero means [DefaultAttempts], 1 makes no retries.
	Attempts int

	// InsecureDisableTLS makes Client using insecure connection without TLS,
	// so the API key travels in plain text. For testing only.
	InsecureDisableTLS bool

	// DebugWriter optionally specifies a destination for a dump of every
	// request and response, payloads included. The API key, leases and handles
	// are redacted. It must be safe for concurrent use, as [os.Stderr] is.
	//
	// For debugging only.
	//
	// Zero dumps nothing.
	DebugWriter io.Writer
	// contains filtered or unexported fields
}

Client implements Spooler API spec.

It is safe for concurrent use and should be shared, since each one holds its own connection pool.

Every operation retries the conditions the API documents as transient, up to Client.Attempts or DefaultAttempts.

An operation on a Client without Client.APIKey set is refused with ErrInvalidRequest before any request is made.

func (*Client) Ack 

func (c *Client) Ack(ctx context.Context, req AckRequest) error

Ack acks the delivery: the message is removed.

func (*Client) AckAndSend 

func (c *Client) AckAndSend(ctx context.Context, req AckAndSendRequest) (AckAndSendResult, error)

AckAndSend acks a delivery and appends a message to another queue as one atomic step: acknowledge and append together, or neither.

A dedup hit on the new message refuses the whole call with ErrDedupClaimed and acks nothing.

Unlike Client.Send, a retry needs no dedup key: a retry of a call that committed fails with ErrStaleToken and appends nothing.

func (*Client) Clone 

func (c *Client) Clone() *Client

Clone returns a copy of c with its settings and none of its connections; the copy opens its own on first use. Use it to run several clients from one set of settings, since a Client must not be copied by value.

func (*Client) CreateQueue 

func (c *Client) CreateQueue(ctx context.Context, req CreateQueueRequest) (Queue, error)

CreateQueue creates a queue and returns it as created; one that already exists is ErrQueueExists. The returned UID names this queue even if another is created under the name since. 

// Ensure a queue "foo" exists.
_, err := c.CreateQueue(ctx, spooler.CreateQueueRequest{
	Spool: "default",
	Queue: "foo",
})
if errors.Is(err, spooler.ErrQueueExists) {
	err = nil
}
if err != nil {
	// Handle error.
}

func (*Client) DeleteQueue 

func (c *Client) DeleteQueue(ctx context.Context, req DeleteQueueRequest) error

DeleteQueue deletes a queue, per the request's Mode. When the call returns the deletion is accepted, not completed: the queue reports QueueStateDeleting or QueueStateDraining until it is gone.

func (*Client) Discard 

func (c *Client) Discard(ctx context.Context, req DiscardRequest) error

Discard drops the message, by its lease or its handle, exactly one.

func (*Client) Fail 

func (c *Client) Fail(ctx context.Context, req FailRequest) error

Fail marks the message failed.

func (*Client) Inspect 

func (c *Client) Inspect(ctx context.Context, req InspectRequest) (InspectResult, error)

Inspect reads the message the handle names without moving it; the handle stays valid.

When the message was read but its payload could not be, Inspect returns the result and the error. Data then holds what was read before the failure.

func (*Client) Messages 

func (c *Client) Messages(ctx context.Context, req MessagesRequest) (Page[Message], error)

Messages lists one page of a queue's messages, without their payloads.

func (*Client) Nack 

func (c *Client) Nack(ctx context.Context, req NackRequest) error

Nack returns the message for redelivery, counting a retry. Past the queue's max retries the message is failed instead.

func (*Client) Queue 

func (c *Client) Queue(ctx context.Context, req QueueRequest) (Queue, error)

Queue reads a queue: its state, its settings and its message counts.

func (*Client) Queues 

func (c *Client) Queues(ctx context.Context, req QueuesRequest) (Page[Queue], error)

Queues lists one page of a spool's queues, each with its settings and its message counts.

func (*Client) Recover 

func (c *Client) Recover(ctx context.Context, req RecoverRequest) error

Recover returns a failed message to its queue, after the request's delay if any.

func (*Client) Recv 

func (c *Client) Recv(ctx context.Context, req RecvRequest) (ret RecvResult, ok bool, err error)

Recv leases the next message of the queue. When none becomes visible within the wait, Recv reports ok as false, with a zero result and no error.

When a message was leased but its payload could not be read, Recv returns the result with the lease and ok as true, and the error: the lease is the caller's to settle. Data then holds what was read before the failure.

func (*Client) Release 

func (c *Client) Release(ctx context.Context, req ReleaseRequest) error

Release returns the message for redelivery without counting a retry, after the request's delay if any.

func (*Client) Renew 

func (c *Client) Renew(ctx context.Context, req RenewRequest) (RenewResult, error)

Renew resets the lease's expiry and reports the new one. The token stays the same.

Example

A consumer that keeps a long lease alive while it works: renew on a ticker, and stop renewing once the message is settled. 

package main

import (
	"context"

	"github.com/spoolersh/spooler-go"
)

func main() {
	ctx := context.Background()
	c := spooler.Client{
		APIKey: "...",
	}

	msg, ok, err := c.Recv(ctx, spooler.RecvRequest{
		Spool:       "default",
		Queue:       "greetings_and_salutations",
		WaitSeconds: new(20),
	})
	if err != nil || !ok {
		return
	}
	if msg.LeaseExpiresAt.IsZero() {
		// The queue has no lease timeout: nothing to renew.
		return
	}

	_, err = c.Renew(ctx, spooler.RenewRequest{
		Spool: "default",
		Lease: msg.Lease,
	})
	if err != nil {
		return
	}
}

func (*Client) Send 

func (c *Client) Send(ctx context.Context, req SendRequest) (SendResult, error)

Send appends a message to a queue.

Message becomes visible to receivers at once, or if set after the request's delay.

Example (Dedup)

Sending with a dedup key makes a retry safe: a repeat inside the queue's dedup window returns the original message instead of appending another. 

package main

import (
	"context"
	"fmt"

	"github.com/spoolersh/spooler-go"
)

func main() {
	ctx := context.Background()
	c := spooler.Client{
		APIKey: "...",
	}

	res, err := c.Send(ctx, spooler.SendRequest{
		Spool: "default",
		Queue: "orders",
		Dedup: spooler.DedupString("order-12345"),
		Data:  []byte(`{"order":12345}`),
	})
	if err != nil {
		return
	}
	if res.Duplicate {
		fmt.Println("already sent as", res.ID)
	}
}

func (*Client) Spool 

func (c *Client) Spool(ctx context.Context, req SpoolRequest) (Spool, error)

Spool reads a spool: its queue count and its message counts summed across every queue.

func (*Client) UpdateQueue 

func (c *Client) UpdateQueue(ctx context.Context, req UpdateQueueRequest) error

UpdateQueue updates queue settings.

type ClientTrace 

type ClientTrace struct {
	// OnConnect is called when a connection to the host is being opened, and
	// the func it returns when the attempt ends.
	OnConnect func(network, addr string) func(error)

	// OnRequest is called once per operation, before any request is made, and
	// the func it returns when the operation ends.
	//
	// A retry inside the operation is not a second call. The context is the
	// caller's, for logging with it or starting a span under it.
	OnRequest func(context.Context, RequestTrace) func(error)

	// OnMalformedResponse is called when a successful answer carries a value
	// the client cannot read, such as a lease expiry that is not a time. The
	// operation still succeeds, with that value zero, since the answer is
	// otherwise good and a lease it granted must reach the caller; the error
	// says which value and why.
	//
	// Getting it called means a server bug: report it.
	OnMalformedResponse func(context.Context, RequestTrace, error)
}

ClientTrace is a set of runtime hooks a Client calls as it works. Every field is optional; a nil hook is skipped. Hooks receive plain values and never a token, so anything they log is safe to keep.

Use ClientTrace.Compose to compose different hook sets, such as logging, metrics, etc.

func LoggerClientTrace 

func LoggerClientTrace(logger *slog.Logger) ClientTrace

LoggerClientTrace builds a ClientTrace that logs every hook to the given logger.

func (ClientTrace) Compose 

func (t ClientTrace) Compose(x ClientTrace) (ret ClientTrace)

Compose returns a new ClientTrace which has functional fields composed both from t and x.

type CreateQueueRequest 

type CreateQueueRequest struct {
	Spool string
	Queue string

	// Settings are optional settings to apply.
	//
	// Zero means the server's defaults.
	Settings QueueSettingsCreate
}

CreateQueueRequest describes a queue to create and the settings to create it with.

type DataSource 

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

DataSource is a streamed payload for a send, from DataSourceReader or DataSourceReadSeeker. The zero value is no source.

func DataSourceReadSeeker 

func DataSourceReadSeeker(rs io.ReadSeeker) DataSource

DataSourceReadSeeker sends rs from its current offset to its end, and rewinds it to send again when a retry is due. To send part of a seekable source, pass an io.SectionReader.

Only Read and Seek are called on r, even if rs implements other interfaces.

func DataSourceReader 

func DataSourceReader(r io.Reader) DataSource

DataSourceReader sends r to its end, streamed without a length. The reader is read once, so a send that fails with an answer that may be retried is not retried. Use DataSourceReadSeeker for a source that can be sent again.

Only Read is called on r, even if r implements other interfaces.

func (DataSource) IsZero 

func (d DataSource) IsZero() bool

IsZero reports whether no source is set.

type Dedup 

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

Dedup is a send's deduplication key, from DedupString or DedupHash; the rules are on https://docs.spooler.sh/dedup.

Zero means no dedup key.

func DedupHash 

func DedupHash(h string) Dedup

DedupHash keys a send by a hash the caller computed, hex-encoded, per the recipe on https://docs.spooler.sh/dedup.

func DedupString 

func DedupString(str string) Dedup

DedupString keys a send by a string; the server hashes it.

type DedupClaimedError 

type DedupClaimedError struct {
	// ID is the message holding the dedup key.
	ID string
}

DedupClaimedError is the details of an ErrorKindDedupClaimed answer.

func (*DedupClaimedError) Error 

func (e *DedupClaimedError) Error() string

Error returns the message's id.

type DeleteQueueRequest 

type DeleteQueueRequest struct {
	Spool string
	Queue string

	// QueueUID is an optional uid of the queue to delete, as [Queue.UID]
	// reports it. Another queue under the name is refused with
	// [ErrQueueReplaced], and nothing is changed, so a retry cannot delete a
	// queue created under the name since.
	//
	// Zero deletes whatever queue has the name.
	QueueUID *string

	Mode QueueDeleteMode
}

DeleteQueueRequest describes a queue to delete and how.

type DiscardRequest 

type DiscardRequest struct {
	Spool string

	// Lease is the delivery to discard, from a receive.
	// Set exactly one of Lease and Handle.
	Lease Lease

	// Handle is the message to discard, from a send or a listing.
	Handle Handle
}

DiscardRequest describes a message to discard, by exactly one of its tokens.

type Error 

type Error struct {
	// Kind is what went wrong, as the API names it. It is ErrorKindUnknown
	// for an answer without a kind and for a kind this SDK does not know.
	Kind ErrorKind

	// Message is the server's text, complete on its own.
	Message string

	// Details is the facts the message states, as a type fixed per kind,
	// such as [QueueNotFoundError]; reach it with [errors.As].
	//
	// Zero means the kind has no details.
	Details error
}

Error is an error the API answered: an answer that names a kind, carries a message, or both. Every operation returns one, wrapped, for such an answer. An answer that does neither, such as a gateway's, is not the API's and is returned as a plain error whose text is its status and body.

You can reach it with errors.As, or match its kind directly with errors.Is against the Err sentinels, as errors.Is(err, ErrQueueNotFound). Its [error.Error] message is for a human only -- a program acts on the kind and the details.

Example

Every error the API answers is an spooler.Error; match its kind with errors.Is, and read a kind's details with errors.As. 

package main

import (
	"context"
	"errors"
	"fmt"

	"github.com/spoolersh/spooler-go"
)

func main() {
	ctx := context.Background()
	c := spooler.Client{
		APIKey: "...",
	}

	_, err := c.Queue(ctx, spooler.QueueRequest{
		Spool: "default",
		Queue: "missing",
	})
	switch {
	case errors.Is(err, spooler.ErrQueueNotFound):
		var detail *spooler.QueueNotFoundError
		if errors.As(err, &detail) {
			fmt.Println("no queue named", detail.Queue)
		}
	case errors.Is(err, spooler.ErrRateLimited):
		var detail *spooler.RateLimitedError
		if errors.As(err, &detail) {
			fmt.Println("retry after", detail.RetryAfter)
		}
	case err != nil:
		fmt.Println(err)
	}
}

func (*Error) Error 

func (e *Error) Error() string

Error returns the server's message, which is complete on its own and already states what Details holds as fields. Without a message it returns the details' text, and failing that the kind's words.

func (*Error) Is 

func (e *Error) Is(x error) bool

Is reports whether x is the KindError of e's kind, so that errors.Is(err, ErrQueueNotFound) matches every queue-not-found answer.

func (*Error) Unwrap 

func (e *Error) Unwrap() error

Unwrap returns the details, so that errors.As reaches them.

type ErrorKind 

type ErrorKind uint

ErrorKind is what an Error reports went wrong: the API's kinds, as https://docs.spooler.sh/errors lists them, and nothing else. A 503 with no known kind is no kind but a sentinel of its own, ErrUnavailable. 

const (
	// ErrorKindUnknown is a kind this SDK does not know, or none.
	ErrorKindUnknown ErrorKind = iota
	// ErrorKindInvalidToken is the API's invalid_token kind.
	ErrorKindInvalidToken
	// ErrorKindDedupDisabled is the API's dedup_disabled kind.
	ErrorKindDedupDisabled
	// ErrorKindUnknownHeader is the API's unknown_header kind.
	ErrorKindUnknownHeader
	// ErrorKindRetentionLimit is the API's retention_limit kind.
	ErrorKindRetentionLimit
	// ErrorKindQueueLimit is the API's queue_limit kind.
	ErrorKindQueueLimit
	// ErrorKindQueueNotFound is the API's queue_not_found kind.
	ErrorKindQueueNotFound
	// ErrorKindSpoolNotFound is the API's spool_not_found kind.
	ErrorKindSpoolNotFound
	// ErrorKindQueueExists is the API's queue_exists kind.
	ErrorKindQueueExists
	// ErrorKindQueueCreating is the API's queue_creating kind.
	ErrorKindQueueCreating
	// ErrorKindQueueDeleting is the API's queue_deleting kind.
	ErrorKindQueueDeleting
	// ErrorKindQueueBusy is the API's queue_busy kind.
	ErrorKindQueueBusy
	// ErrorKindNotFailed is the API's not_failed kind.
	ErrorKindNotFailed
	// ErrorKindDedupInFlight is the API's dedup_in_flight kind.
	ErrorKindDedupInFlight
	// ErrorKindDedupClaimed is the API's dedup_claimed kind.
	ErrorKindDedupClaimed
	// ErrorKindStaleToken is the API's stale_token kind.
	ErrorKindStaleToken
	// ErrorKindPayloadTooLarge is the API's payload_too_large kind.
	ErrorKindPayloadTooLarge
	// ErrorKindRateLimited is the API's rate_limited kind.
	ErrorKindRateLimited
	// ErrorKindOperationUnconfirmed is the API's operation_unconfirmed kind.
	ErrorKindOperationUnconfirmed
	// ErrorKindSpoolFull is the API's spool_full kind.
	ErrorKindSpoolFull
	// ErrorKindQueueReplaced is the API's queue_replaced kind.
	ErrorKindQueueReplaced
	// ErrorKindUnauthenticated is the API's unauthenticated kind.
	ErrorKindUnauthenticated
	// ErrorKindAccountSuspended is the API's account_suspended kind.
	ErrorKindAccountSuspended
	// ErrorKindAccountBlocked is the API's account_blocked kind.
	ErrorKindAccountBlocked
)

func (ErrorKind) String 

func (k ErrorKind) String() string

String names the kind as Error prints it: its wire name in words.

type FailRequest 

type FailRequest struct {
	Spool string
	Lease Lease
}

FailRequest describes a delivery to fail.

type Handle 

type Handle string

Handle is a reference to a message as it is now, from a send that asked for one or from a listing. It grants no ownership and has no timeout, and it goes stale once anything moves the message.

The token is opaque. It is a credential, and prints redacted.

func (Handle) String 

func (Handle) String() string

String returns a redacted form, never the token.

type InspectRequest 

type InspectRequest struct {
	Spool string

	// Handle is the message to read, from a listing or a send.
	Handle Handle

	// Stream is an optional request to return the payload unread as
	// [InspectResult.DataStream], for the caller to read.
	//
	// For [ClientTrace.OnRequest], the operation ends when the stream is read
	// to its end or closed.
	//
	// Zero reads the payload into [InspectResult.Data].
	Stream bool
}

InspectRequest describes a message to read, by its handle.

type InspectResult 

type InspectResult struct {
	// ID is the message id. It identifies the message in listings and logs.
	// No operation takes it.
	ID string

	State MessageState

	// Retries is the redeliveries so far, or -1 when the server reported a
	// value that did not parse.
	Retries int

	// CreatedAt is when the message was sent.
	CreatedAt time.Time

	// VisibleAt is when a delayed message becomes visible.
	//
	// Zero means the message is not delayed.
	VisibleAt time.Time

	// LeaseExpiresAt is when the current lease expires.
	//
	// Zero means no lease, or one that never expires.
	LeaseExpiresAt time.Time

	// ExpiresAt is when retention discards the message.
	//
	// Zero means retention is off.
	ExpiresAt time.Time

	// Data is the payload. It is nil when the request set Stream, and when
	// the message was sent without one.
	Data []byte

	// DataStream is the unread payload stream, set when the request had
	// Stream set (empty messages included).
	//
	// Reading is bounded by the inspect's context. A payload cut short reads
	// as [io.ErrUnexpectedEOF].
	//
	// Users must close it as soon as the data is read.
	DataStream io.ReadCloser

	// DataSize is the payload's size in bytes as the server declared it, or
	// -1 when it did not.
	DataSize int64
}

InspectResult is the outcome of an inspect: the message as it is now.

type KindError 

type KindError struct {
	Kind ErrorKind
}

KindError is a sentinel that every Error of its kind matches with errors.Is. The package declares one per kind, as ErrQueueNotFound; a KindError is never returned itself.

func (*KindError) Error 

func (e *KindError) Error() string

Error returns the kind's words.

type Lease 

type Lease string

Lease is exclusive ownership of one delivery, from a receive. The token is opaque. It is a credential, and prints redacted.

func (Lease) String 

func (Lease) String() string

String returns a redacted form, never the token.

type Message 

type Message struct {
	ID    string
	State MessageState

	// Retries is the redeliveries so far.
	Retries int

	// CreatedAt is when the message was sent.
	CreatedAt time.Time

	// VisibleAt is when a delayed message becomes visible.
	//
	// Zero means the message is not delayed.
	VisibleAt time.Time

	// LeaseExpiresAt is when the current lease expires.
	//
	// Zero means no lease, or one that never expires.
	LeaseExpiresAt time.Time

	// ExpiresAt is when retention discards the message.
	//
	// Zero means retention is off.
	ExpiresAt time.Time

	// Handle is the message's handle.
	//
	// Zero means the listing did not ask for one with ReturnHandle.
	Handle Handle
}

Message is a message as Client.Messages lists it.

type MessageState 

type MessageState uint

MessageState is the state a message is in. 

const (
	// MessageStateUnknown is the zero value and no state.
	MessageStateUnknown MessageState = iota
	// MessageStateDelayed is a message not yet visible.
	MessageStateDelayed
	// MessageStateVisible is a message waiting for a receive.
	MessageStateVisible
	// MessageStatePending is a message leased to a receiver.
	MessageStatePending
	// MessageStateFailed is a message failed, waiting for a recover or a
	// discard.
	MessageStateFailed
	// MessageStateUnavailable is a message whose payload the service could
	// not read.
	MessageStateUnavailable
)

func ParseMessageState 

func ParseMessageState(s string) (MessageState, error)

ParseMessageState returns the state named s, as the API spells it.

func (MessageState) String 

func (s MessageState) String() string

String returns the state's name as the API spells it, or MessageState(n) for a value the SDK does not define.

type MessageStats 

type MessageStats struct {
	// SampledAt is when the counts were taken; equal values mean one
	// snapshot.
	SampledAt time.Time

	// StartedAt is when the totals started from zero. A restart or a
	// failover moves it and starts them over, so compute rates only
	// between samples with equal StartedAt.
	StartedAt time.Time

	// Total is the stored messages, every state counted.
	Total int64

	// DataBytes is the payload bytes of the counted messages.
	DataBytes int64

	Delayed int64
	Visible int64

	// Pending is the messages leased.
	Pending int64

	Failed int64

	// Unavailable is the messages whose payload the service could not
	// read.
	Unavailable int64

	// AddedTotal is the messages sent, duplicates excluded.
	AddedTotal int64

	AckedTotal     int64
	NackedTotal    int64
	ReleasedTotal  int64
	FailedTotal    int64
	DiscardedTotal int64

	// ExpiredTotal is the messages discarded by retention.
	ExpiredTotal int64
}

MessageStats counts messages by state as of SampledAt, and cumulative totals since StartedAt.

type MessagesRequest 

type MessagesRequest struct {
	Spool string
	Queue string

	// State is an optional filter keeping only messages in this state on
	// the page.
	//
	// Zero keeps messages of every state on the page.
	State MessageState

	// Limit is an optional cap on the messages on the page.
	//
	// Zero means the server's default.
	Limit int

	// After is an optional cursor resuming the page just past this message
	// id, from the previous page's Next.
	//
	// Zero starts from the first page.
	After string

	// ReturnHandle is an optional request for a handle on each message, for
	// [Client.Inspect], [Client.Recover] and [Client.Discard].
	//
	// A handle is a credential, so it is off by default.
	ReturnHandle bool
}

MessagesRequest describes a queue's messages to list, and their pagination.

type NackRequest 

type NackRequest struct {
	Spool string
	Lease Lease
}

NackRequest describes a delivery to nack.

type Op 

type Op uint

Op is an operation of the Client, for tracing. 

const (
	OpUnknown Op = iota
	OpSpool
	OpQueues
	OpMessages
	OpCreateQueue
	OpQueue
	OpUpdateQueue
	OpDeleteQueue
	OpSend
	OpAckAndSend
	OpRecv
	OpRenew
	OpAck
	OpNack
	OpRelease
	OpFail
	OpDiscard
	OpInspect
	OpRecover
)

The operations, one per Client method.

func (Op) String 

func (o Op) String() string

String names the operation as the method is named.

type Page 

type Page[T any] struct {
	Items []T

	// Next is the cursor of the following page; empty when this page
	// ends the listing.
	Next string
}

Page is one page of a listing. Every listing is paged the same way: pass Next as the next request's After until it comes back empty.

type Queue 

type Queue struct {
	Name string

	// UID is unique to this queue, opaque: a queue created again under
	// the same name gets a new one. Present it as
	// [DeleteQueueRequest.QueueUID] to delete this queue only.
	UID string

	CreatedAt time.Time
	State     QueueState

	// Settings is the queue's settings, as applied.
	Settings QueueSettings

	// Stats is the queue's message counts, from the same snapshot as the
	// rest; all zero on the queue a create returns.
	Stats MessageStats
}

Queue is a queue as Client.Queue and Client.Queues describe it.

type QueueDeleteMode 

type QueueDeleteMode uint

QueueDeleteMode is how Client.DeleteQueue deletes a queue. 

const (
	// QueueDeleteModeUnknown is the zero value and no mode; a request
	// carrying it is refused before sending.
	QueueDeleteModeUnknown QueueDeleteMode = iota
	// QueueDeleteModeForce deletes the queue and its messages at once.
	QueueDeleteModeForce
	// QueueDeleteModeDrain deletes the queue once consumers have emptied it.
	QueueDeleteModeDrain
)

func ParseQueueDeleteMode 

func ParseQueueDeleteMode(s string) (QueueDeleteMode, error)

ParseQueueDeleteMode returns the mode named s, as the API spells it.

func (QueueDeleteMode) String 

func (q QueueDeleteMode) String() string

String returns the mode's name as the API spells it, or QueueDeleteMode(n) for a value the SDK does not define.

type QueueLimitError 

type QueueLimitError struct {
	Max int64
}

QueueLimitError is the details of an ErrorKindQueueLimit answer.

func (*QueueLimitError) Error 

func (e *QueueLimitError) Error() string

Error returns the limit.

type QueueNotFoundError 

type QueueNotFoundError struct {
	Queue string
}

QueueNotFoundError is the details of an ErrorKindQueueNotFound answer.

func (*QueueNotFoundError) Error 

func (e *QueueNotFoundError) Error() string

Error returns the queue's name.

type QueueReplacedError 

type QueueReplacedError struct {
	// UID is the uid of the queue now under the name.
	UID string
}

QueueReplacedError is the details of an ErrorKindQueueReplaced answer.

func (*QueueReplacedError) Error 

func (e *QueueReplacedError) Error() string

Error returns the current queue's uid.

type QueueRequest 

type QueueRequest struct {
	Spool string
	Queue string
}

QueueRequest describes a queue to read.

type QueueSettings 

type QueueSettings struct {
	// LeaseTimeoutSeconds is the lease timeout in seconds.
	//
	// Zero means leases never expire.
	LeaseTimeoutSeconds int

	// MaxRetries is the redeliveries a message gets before it is failed.
	//
	// Zero means none.
	MaxRetries int

	// DedupWindowSeconds is the dedup window in seconds.
	//
	// Zero means dedup is off.
	DedupWindowSeconds int

	// RetentionSeconds is the retention in seconds, as set.
	RetentionSeconds int

	// EffectiveRetentionSeconds is the retention in seconds, as applied.
	EffectiveRetentionSeconds int

	// RecvRateLimit is the rate limit on receives.
	//
	// Zero means no limit.
	RecvRateLimit RateLimit
}

QueueSettings is a queue's settings as applied, every field present.

type QueueSettingsCreate 

type QueueSettingsCreate struct {
	// LeaseTimeoutSeconds is the lease timeout in seconds; a value of 0
	// disables it.
	//
	// Zero means the server's default.
	LeaseTimeoutSeconds *int

	// MaxRetries is the redeliveries a message gets before it is failed;
	// a value of 0 means none.
	//
	// Zero means the server's default.
	MaxRetries *int

	// DedupWindowSeconds is the dedup window in seconds; a value of 0
	// disables dedup.
	//
	// Zero means the server's default.
	DedupWindowSeconds *int

	// RetentionSeconds is the retention in seconds.
	//
	// Zero means the server's default.
	RetentionSeconds *int

	// RecvRateLimit is the rate limit on receives.
	//
	// Zero means no limit.
	RecvRateLimit *RateLimit
}

QueueSettingsCreate is the settings a queue is created with. A nil field takes the server's default.

type QueueSettingsUpdate 

type QueueSettingsUpdate struct {
	// LeaseTimeoutSeconds is the lease timeout in seconds; a value of 0
	// disables it.
	//
	// Zero keeps the current value.
	LeaseTimeoutSeconds *int

	// MaxRetries is the redeliveries a message gets before it is failed;
	// a value of 0 means none.
	//
	// Zero keeps the current value.
	MaxRetries *int

	// DedupWindowSeconds is the dedup window in seconds; a value of 0
	// disables dedup.
	//
	// Zero keeps the current value.
	DedupWindowSeconds *int

	// RetentionSeconds is the retention in seconds.
	//
	// Zero keeps the current value.
	RetentionSeconds *int

	// RecvRateLimit is the rate limit on receives; a pointer to the zero
	// RateLimit removes it.
	//
	// Zero keeps the current value.
	RecvRateLimit *RateLimit
}

QueueSettingsUpdate is the settings to replace on a queue. A nil field keeps its current value; at least one must be set.

type QueueState 

type QueueState uint

QueueState is the state a queue is in. 

const (
	// QueueStateUnknown is the zero value and no state.
	QueueStateUnknown QueueState = iota
	// QueueStateActive is a queue in service.
	QueueStateActive
	// QueueStateDeleting is a queue being deleted.
	QueueStateDeleting
	// QueueStateDraining is a queue being emptied before its deletion.
	QueueStateDraining
)

func (QueueState) String 

func (s QueueState) String() string

String returns the state's name as the API spells it, or QueueState(n) for a value the SDK does not define.

type QueuesRequest 

type QueuesRequest struct {
	Spool string

	// Match is an optional filter keeping on the page only the queues whose
	// names the expression matches, unanchored.
	//
	// Zero filters nothing.
	Match *regexp.Regexp

	// Limit is an optional cap on the queues on the page.
	//
	// Zero means the server's default.
	Limit int

	// After is an optional cursor resuming the page from the previous page's
	// Next.
	//
	// Zero starts from the first page.
	After string
}

QueuesRequest describes a spool's queues to list, and their pagination.

type RateLimit 

type RateLimit struct {
	// IntervalMicros is the refill interval in microseconds.
	IntervalMicros int64

	// Burst is the bucket size.
	Burst int
}

RateLimit is a token bucket rate limit.

Zero means no limit.

func (RateLimit) IsZero 

func (r RateLimit) IsZero() bool

IsZero reports whether r sets no limit. A limit needs both fields.

type RateLimitedError 

type RateLimitedError struct {
	// RetryAfter is how long the server asked to wait before retrying.
	RetryAfter time.Duration
}

RateLimitedError is the details of an ErrorKindRateLimited answer.

func (*RateLimitedError) Error 

func (e *RateLimitedError) Error() string

Error returns the wait.

type RecoverRequest 

type RecoverRequest struct {
	Spool string

	// Handle is the failed message, from a listing.
	Handle Handle

	// DelaySeconds is an optional delay until the recovered message becomes
	// visible.
	//
	// Zero means visible at once.
	DelaySeconds int
}

RecoverRequest describes a failed message to return to its queue.

type RecvRequest 

type RecvRequest struct {
	Spool string
	Queue string

	// WaitSeconds is an optional bound on the long poll; a value of 0
	// returns without waiting.
	//
	// Zero means the server's default.
	WaitSeconds *int

	// Stream is an optional request to return the payload unread as
	// [RecvResult.DataStream], for the caller to read.
	//
	// If set, Recv then returns as soon as the message is leased.
	//
	// For [ClientTrace.OnRequest], the operation ends when the stream is read
	// to its end or closed.
	//
	// Zero reads the payload into [RecvResult.Data].
	Stream bool
}

RecvRequest describes a queue to receive from and how long to wait.

type RecvResult 

type RecvResult struct {
	// ID is the message id. It identifies the message in listings and logs.
	// No operation takes it.
	ID string

	// Retries is the redeliveries so far, or -1 when the server reported a
	// value that did not parse.
	Retries int

	// CreatedAt is when the message was sent.
	CreatedAt time.Time

	Lease Lease

	// LeaseExpiresAt is when the lease expires.
	//
	// Zero means the lease never expires.
	LeaseExpiresAt time.Time

	// Data is the message payload. It is nil when the request set Stream,
	// and when the message was sent without one.
	Data []byte

	// DataStream is the unread payload stream, set when the request had
	// Stream set (empty messages included).
	//
	// Reading is bounded by the receive's context. A payload cut short reads
	// as [io.ErrUnexpectedEOF], and the lease stays the caller's to settle.
	//
	// Users must close it as soon as the data is read.
	DataStream io.ReadCloser

	// DataSize is the payload's size in bytes as the server declared it, or
	// -1 when it did not.
	DataSize int64
}

RecvResult is the outcome of a receive: the message and the lease on it.

type ReleaseRequest 

type ReleaseRequest struct {
	Spool string
	Lease Lease

	// DelaySeconds is an optional delay until the message becomes visible
	// again.
	//
	// Zero means visible at once.
	DelaySeconds int
}

ReleaseRequest describes a delivery to release.

type RenewRequest 

type RenewRequest struct {
	Spool string
	Lease Lease
}

RenewRequest describes a lease to renew.

type RenewResult 

type RenewResult struct {
	// LeaseExpiresAt is the new expiry.
	//
	// Zero means the lease never expires.
	LeaseExpiresAt time.Time
}

RenewResult is the outcome of a renew.

type RequestTrace 

type RequestTrace struct {
	Op    Op
	Spool string

	// Queue is the queue the operation names, or empty for a spool-scoped
	// one such as a settle.
	Queue string
}

RequestTrace describes the operation an OnRequest hook is called for.

func (RequestTrace) LogValue 

func (t RequestTrace) LogValue() slog.Value

LogValue renders the trace as a group of its fields, so that a logger can attach it as one attribute.

type RetentionLimitError 

type RetentionLimitError struct {
	Max time.Duration
}

RetentionLimitError is the details of an ErrorKindRetentionLimit answer.

func (*RetentionLimitError) Error 

func (e *RetentionLimitError) Error() string

Error returns the limit.

type SendRequest 

type SendRequest struct {
	Spool string
	Queue string

	// DelaySeconds is an optional delay until the message becomes visible.
	//
	// Zero means visible at once.
	DelaySeconds int

	// Dedup is an optional deduplication key.
	Dedup Dedup

	// Data is an optional payload.
	//
	// Zero sends an empty message.
	Data []byte

	// DataSource is an optional stream to send as the payload instead of Data.
	// Setting both is an error.
	DataSource DataSource

	// ReturnHandle is an optional request for a handle to the new message,
	// which can be used in [Client.Discard] before its first delivery.
	//
	// A handle is a credential, so it is off by default.
	ReturnHandle bool

	// RetryUnknown is an optional request to retry an attempt whose result is
	// unknown, within the attempts [Client.Attempts] allows. Such a retry can
	// append the message a second time, as a new message with its own id:
	// consumers whose effect is idempotent absorb it, those that only track
	// handled message ids do not. With Dedup, a retry the queue's dedup window
	// still remembers appends nothing new. A payload from [DataSourceReader] is
	// still sent once.
	//
	// Zero retries no attempt whose result is unknown.
	RetryUnknown bool
}

SendRequest describes a message to append to a queue.

type SendResult 

type SendResult struct {
	// ID is the message id. It identifies the message in listings and logs.
	// No operation takes it.
	ID string

	// Duplicate reports a dedup hit: nothing was appended, and ID is the
	// original message's.
	Duplicate bool

	// Handle is the handle to the new message, when the request asked for one.
	// It is absent on a dedup hit.
	Handle Handle
}

SendResult is the outcome of a send.

type Spool 

type Spool struct {
	Name       string
	QueueCount int64

	// Stats is the message counts summed across every queue. The
	// cumulative totals are the spool's own and keep counting deleted
	// queues, so they are not a sum over the queues listed.
	Stats MessageStats
}

Spool is a spool as Client.Spool describes it.

type SpoolFullError 

type SpoolFullError struct {
	// Limit is the cap that was reached, named after the [MessageStats]
	// field it is enforced against.
	Limit SpoolLimit

	Max int64
}

SpoolFullError is the details of an ErrorKindSpoolFull answer.

func (*SpoolFullError) Error 

func (e *SpoolFullError) Error() string

Error returns the cap and its name.

type SpoolLimit 

type SpoolLimit uint

SpoolLimit is which of a spool's caps SpoolFullError reports reached. 

const (
	// SpoolLimitUnknown is a cap this SDK does not know, or none.
	SpoolLimitUnknown SpoolLimit = iota
	// SpoolLimitMessages is the cap on stored messages.
	SpoolLimitMessages
	// SpoolLimitDataBytes is the cap on stored payload bytes.
	SpoolLimitDataBytes
)

func (SpoolLimit) String 

func (l SpoolLimit) String() string

String returns the cap's name as the API spells it, or SpoolLimit(n) for a value the SDK does not define.

type SpoolRequest 

type SpoolRequest struct {
	Spool string
}

SpoolRequest describes a spool to read.

type UnknownHeaderError 

type UnknownHeaderError struct {
	// Header is the refused header's name.
	Header string
}

UnknownHeaderError is the details of an ErrorKindUnknownHeader answer.

func (*UnknownHeaderError) Error 

func (e *UnknownHeaderError) Error() string

Error returns the header's name.

type UpdateQueueRequest 

type UpdateQueueRequest struct {
	Spool string
	Queue string

	// QueueUID is an optional uid of the queue to update, as [Queue.UID]
	// reports it. Another queue under the name is refused with
	// [ErrQueueReplaced], and nothing is changed.
	//
	// Zero updates whatever queue has the name.
	QueueUID *string

	// Settings are the settings to replace.
	// At least one Settings's field must be set.
	Settings QueueSettingsUpdate
}

UpdateQueueRequest describes a queue and the settings to replace on it.

Source Files

View all Source files

Directories

Path Synopsis
internal
backoff
httputil

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.