Documentation
¶
Overview ¶
Package memoryqueue provides a non-durable, in-process queue driver for local development. It implements queue.DispatcherDriver and queue.ConsumerDriver with competing consumers, delayed retries, and exactly-once settlement per delivery. Messages are lost when the process exits. For tests, prefer the recording fake in package queuetest.
Delayed retries are promoted lazily by consumers. Construction starts no goroutines or retry scheduler; timers work inside a testing/synctest bubble. Cancellation is checked before admitting each delivery, including while a queue has a backlog. An admitted handler still owns settlement; a detached settlement context can acknowledge or retry its delivery after cancellation. Undelivered messages and accepted retries remain queued for live consumers. Stop instead rejects new admissions, discards queued work, and waits for admitted handlers until its context expires.
During application composition, register your printText handler (func(context.Context, string) error) before registering the runner:
driver := memoryqueue.New()
consumer := queue.NewConsumer(driver)
job := queue.NewJob[string]("print")
compose.Require(consumer.Handle(job, printText))
compose.Cleanup(driver.Stop)
compose.Run(consumer.Run)
New takes Builder policies for RetryDelay and Logger. They run directly once in order on a fresh draft; nil entries panic and expanded nil/empty slices supply no edits. Publication detaches settings, not the logger. RetryDelay retains its existing meaning for a retry requesting zero delay. Construction starts no consumer or goroutine and retained drafts cannot alter scheduling.
Example ¶
Example adds a default delay for deliveries that request an immediate retry. Construction starts no consumers; this example only configures the driver.
package main
import (
"time"
"arkoren.dev/core/service/queue/memoryqueue"
)
func main() {
driver := memoryqueue.New(func(consumer *memoryqueue.Builder) {
consumer.RetryDelay(time.Second)
})
_ = driver
}
Output:
Index ¶
- Variables
- type Builder
- type BuilderFunc
- type Driver
- func (driver *Driver) Consume(ctx context.Context, name string, handler queue.DeliveryHandler) error
- func (driver *Driver) Dispatch(ctx context.Context, message queue.Message) error
- func (driver *Driver) MessagingSystem() string
- func (driver *Driver) Start(ctx context.Context) error
- func (driver *Driver) Stop(ctx context.Context) error
Examples ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrDriverClosed rejects startup or queue admission after shutdown begins. // Already-canceled contexts can take precedence at admission boundaries. // // if errors.Is(err, memoryqueue.ErrDriverClosed) { // return err // } ErrDriverClosed = broker.ErrDriverClosed // ErrInvalidQueue is returned when a queue name is empty. // // err := driver.Dispatch(ctx, queue.Message{}) // fmt.Println(errors.Is(err, memoryqueue.ErrInvalidQueue)) ErrInvalidQueue = broker.ErrInvalidQueue // ErrInvalidHandler is returned when Consume is given a nil handler. // // err := driver.Consume(ctx, "work", nil) // fmt.Println(errors.Is(err, memoryqueue.ErrInvalidHandler)) ErrInvalidHandler = broker.ErrInvalidHandler // ErrDeliverySettled is returned when a delivery is settled more than once. // It also rejects a concurrent settlement already in progress. // // if err := delivery.Acknowledge(ctx); errors.Is(err, memoryqueue.ErrDeliverySettled) { // return err // } ErrDeliverySettled = broker.ErrDeliverySettled // ErrDeliveryUnsettled is returned when a handler returns without settling a delivery. // The driver attempts terminal cleanup before ending Consume with this error. // // if errors.Is(err, memoryqueue.ErrDeliveryUnsettled) { // return err // } ErrDeliveryUnsettled = broker.ErrDeliveryUnsettled )
Functions ¶
This section is empty.
Types ¶
type Builder ¶
type Builder struct {
// contains filtered or unexported fields
}
Builder configures New before publication. Callbacks run directly once, synchronously in order on one fresh draft. Nil callback entries panic; expanded nil or empty callback slices supply no edits. Retaining the draft cannot change published settings. Methods return the same draft.
Example:
memoryqueue.New(func(consumer *memoryqueue.Builder) {
consumer.RetryDelay(time.Second)
})
func (*Builder) Logger ¶
Logger selects the shared logger for delivery failures and recovered panics. Nil resolves slog.Default after all construction callbacks.
This setting replaces the previous value on the construction draft. Services and loggers remain shared; publication detaches owned settings.
Example:
memoryqueue.New(func(consumer *memoryqueue.Builder) {
consumer.Logger(nil)
})
func (*Builder) RetryDelay ¶
RetryDelay supplies the delay when a delivery requests a zero-delay retry. Zero or negative values enqueue that retry immediately; positive values defer it. An explicit nonzero per-delivery delay retains its runtime precedence.
This setting replaces the previous value on the construction draft. Services and loggers remain shared; publication detaches owned settings.
Example:
memoryqueue.New(func(consumer *memoryqueue.Builder) {
consumer.RetryDelay(time.Second)
})
type BuilderFunc ¶
type BuilderFunc = func(*Builder)
BuilderFunc edits the fresh draft used by New. Callbacks run directly once in order; nil entries panic. Publication detaches the settings, not loggers or closure captures. Retained drafts cannot alter a running queue.
policy := memoryqueue.BuilderFunc(func(consumer *memoryqueue.Builder) {
consumer.RetryDelay(time.Second)
})
driver := memoryqueue.New(policy)
type Driver ¶
type Driver struct {
// contains filtered or unexported fields
}
Driver implements queue.DispatcherDriver and queue.ConsumerDriver with in-memory data structures. It is non-durable: process exit or crash loses queued jobs. A Driver is safe for concurrent use. Create it with New; consumers, rather than construction, advance delayed work.
driver := memoryqueue.New() fmt.Println(driver.MessagingSystem()) // memory
func New ¶
func New(configure ...BuilderFunc) *Driver
New creates a non-durable in-memory job driver. It starts no goroutines.
Construction callbacks run directly once, synchronously in order on one fresh draft. Nil entries panic; expanded nil or empty slices omit edits. Publication detaches owned settings, not live dependencies or captures.
Example:
memoryqueue.New(func(consumer *memoryqueue.Builder) {
consumer.RetryDelay(time.Second)
})
func (*Driver) Consume ¶
func (driver *Driver) Consume(ctx context.Context, name string, handler queue.DeliveryHandler) error
Consume receives and handles deliveries from queue until ctx is canceled or the driver shuts down. Concurrent consumers for one queue compete for each message; each reserved message is delivered to only one handler. It returns the context error on cancellation and ErrDriverClosed on shutdown.
Cancellation and shutdown are checked before each reservation. A delivery already admitted remains the handler's responsibility to settle; cancellation does not revoke it. Messages not yet admitted remain available to other consumers after cancellation. Driver.Stop instead discards queued messages.
Example:
driver := memoryqueue.New()
if err := driver.Dispatch(context.Background(), queue.Message{Queue: "work"}); err != nil {
log.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
err := driver.Consume(ctx, "work", func(ctx context.Context, delivery queue.Delivery) error {
defer cancel()
return delivery.Acknowledge(ctx)
})
if !errors.Is(err, context.Canceled) {
log.Fatal(err)
}
Example ¶
ExampleDriver_Consume cancels consumption after one acknowledged delivery. The context error describes the consumer's exit, not a settlement failure.
package main
import (
"context"
"errors"
"fmt"
"log"
"arkoren.dev/core/service/queue"
"arkoren.dev/core/service/queue/memoryqueue"
)
func main() {
driver := memoryqueue.New()
if err := driver.Dispatch(context.Background(), queue.Message{Queue: "work"}); err != nil {
log.Fatal(err)
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
err := driver.Consume(ctx, "work", func(ctx context.Context, delivery queue.Delivery) error {
defer cancel()
return delivery.Acknowledge(ctx)
})
if !errors.Is(err, context.Canceled) {
log.Fatal(err)
}
fmt.Println("consumer canceled after acknowledgment")
}
Output: consumer canceled after acknowledgment
func (*Driver) Dispatch ¶
Dispatch adds message to its named queue. Payload is copied before acceptance and a missing ID is generated. Metadata is copied too; caller mutations after acceptance do not change the delivery. Input Attempts is ignored. An empty queue, canceled context, or closed driver returns an error without adding a message.
if err := driver.Dispatch(ctx, queue.Message{
Name: "send", Queue: "work", Payload: []byte(`"hello"`),
}); err != nil {
return err
}
func (*Driver) MessagingSystem ¶
MessagingSystem returns "memory", the messaging.system telemetry value of the driver. This is a telemetry classification, not a persistence capability.
fmt.Println(memoryqueue.New().MessagingSystem()) // memory
func (*Driver) Start ¶
Start reports ErrDriverClosed after Driver.Stop and otherwise does nothing: the driver needs no background work. It exists for application lifecycle wiring. An already-canceled context returns its error before checking closure.
driver := memoryqueue.New()
if err := driver.Start(ctx); err != nil {
return err
}
func (*Driver) Stop ¶
Stop stops new admissions, discards queued jobs, and waits for active consumers and handlers until ctx expires, then returns the context's error. The release happens even when ctx is already expired: Stop then returns without waiting for handlers still running. Stop is idempotent. It cannot forcibly stop a handler; handlers must cooperate with their own contexts. Repeated Stop calls can wait again for remaining consumers.
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
if err := driver.Stop(ctx); err != nil {
return err
}
Source Files
Directories