Documentation
¶
Overview ¶
Package queuetest tests code that dispatches or consumes jobs. It offers a recording Fake with typed assertions and a synchronous Fake.Drain, and Run, a conformance suite that every queue.DispatcherDriver and queue.ConsumerDriver pair must pass.
A test that exercises code under test dispatching jobs:
fake := queuetest.New()
dispatcher := queue.NewDispatcher(fake)
consumer := queue.NewConsumer(fake)
_ = consumer.Handle(job, handler)
_ = dispatcher.Dispatch(ctx, job, payload)
fake.AssertDispatched(t, job, func(p Payload) bool {
return p.ID == 1
})
_ = fake.Drain(ctx, consumer)
Drain runs the real queue.Consumer, so handler registration, decoding, retry, rejection, and failure handling behave as in production.
A driver author proves a driver with the suite:
func TestConformance(t *testing.T) {
queuetest.Run(t, func(t *testing.T) queuetest.Driver {
driver := memoryqueue.New()
t.Cleanup(func() {
_ = driver.Stop(context.Background())
})
return queuetest.Driver{Dispatcher: driver, Consumer: driver}
})
}
Example ¶
Example dispatches a job through a Fake and runs it with Drain.
package main
import (
"context"
"fmt"
"arkoren.dev/core/service/queue"
"arkoren.dev/core/service/queue/queuetest"
)
func main() {
ctx := context.Background()
job := queue.NewJob[string]("greet")
fake := queuetest.New()
dispatcher := queue.NewDispatcher(fake)
consumer := queue.NewConsumer(fake)
if err := consumer.Handle(job, func(_ context.Context, name string) error {
fmt.Println("hello", name)
return nil
}); err != nil {
fmt.Println(err)
return
}
if err := dispatcher.Dispatch(ctx, job, "world"); err != nil {
fmt.Println(err)
return
}
fmt.Println("dispatched:", len(fake.Dispatched()))
if err := fake.Drain(ctx, consumer); err != nil {
fmt.Println(err)
return
}
}
Output: dispatched: 1 hello world
Index ¶
- Constants
- Variables
- func Run(t *testing.T, newDriver func(t *testing.T) Driver, opts ...Option)
- type DrainOption
- type Driver
- type Fake
- func (fake *Fake) AssertDispatched[T any](t testing.TB, job queue.Job[T], preds ...func(T) bool)
- func (fake *Fake) AssertDispatchedCount[T any](t testing.TB, job queue.Job[T], n int)
- func (fake *Fake) AssertNotDispatched[T any](t testing.TB, job queue.Job[T], preds ...func(T) bool)
- func (fake *Fake) AssertNothingDispatched(t testing.TB)
- func (fake *Fake) Consume(ctx context.Context, name string, handler queue.DeliveryHandler) error
- func (fake *Fake) Dispatch(ctx context.Context, message queue.Message) error
- func (fake *Fake) Dispatched() []queue.Message
- func (fake *Fake) Drain(ctx context.Context, consumer *queue.Consumer, opts ...DrainOption) error
- func (fake *Fake) FailNext(err error)
- func (fake *Fake) FailWith(err error)
- func (fake *Fake) Failed() []queue.Message
- func (fake *Fake) Payloads[T any](job queue.Job[T]) ([]T, error)
- func (fake *Fake) Pending() []queue.Message
- func (fake *Fake) Rejected() []queue.Message
- func (fake *Fake) Reset()
- func (fake *Fake) Start(context.Context) error
- func (fake *Fake) Stop(context.Context) error
- type Option
Examples ¶
Constants ¶
const DefaultMaxAttempts = 100
DefaultMaxAttempts is the attempt limit of Fake.Drain when WithMaxAttempts is not used.
Variables ¶
var ( // ErrInvalidQueue is returned when a dispatched message names no queue. ErrInvalidQueue = errors.New("queuetest: message has no queue") // ErrInvalidHandler is returned when Consume is given a nil handler. ErrInvalidHandler = errors.New("queuetest: nil delivery handler") // ErrDeliverySettled is returned when a delivery is settled more than once. ErrDeliverySettled = errors.New("queuetest: delivery already settled") // ErrUnsettled marks a delivery whose handler returned without settling it. ErrUnsettled = errors.New("queuetest: delivery was not settled") )
var ErrDrainLimit = errors.New("queuetest: drain attempt limit reached")
ErrDrainLimit is returned by Fake.Drain when a message is retried more times than the limit set by WithMaxAttempts, which stops handlers that retry forever.
var ErrForeignConsumer = errors.New("queuetest: consumer is not backed by this fake")
ErrForeignConsumer is returned by Fake.Drain for a consumer that is not backed by the Fake.
Functions ¶
func Run ¶
Run is the conformance suite for queue drivers. It runs one subtest per behavior documented on queue.DispatcherDriver, queue.ConsumerDriver, and queue.Delivery, plus the end-to-end behavior of queue.Consumer on top of the driver. newDriver is called once per subtest and must return a driver on empty queues, registering any cleanup with t.
Unless WithSynctest is given, the suite uses real time with delays of about a second, so infrastructure backed drivers should run it only outside short mode.
Types ¶
type DrainOption ¶
type DrainOption func(*drainState)
DrainOption changes how Fake.Drain runs pending messages.
func WithMaxAttempts ¶
func WithMaxAttempts(attempts int) DrainOption
WithMaxAttempts sets how often one message may be delivered during a Drain before it fails with ErrDrainLimit. Values below one are ignored.
func WithScheduled ¶
func WithScheduled() DrainOption
WithScheduled makes Drain also run messages whose delivery lies in the future, such as retries requested with a delay. Time does not advance; the delays are skipped.
type Driver ¶
type Driver struct {
// Dispatcher accepts dispatched messages.
Dispatcher queue.DispatcherDriver
// Consumer receives them. It must read the queues Dispatcher writes.
Consumer queue.ConsumerDriver
// Queue names an empty queue the suite may use. Drivers that must
// provision or map queue names, such as SQS, set it to a queue created for
// the test. The default is "conformance".
Queue string
// Other names a second empty queue, distinct from Queue. The default is
// "conformance-other".
Other string
}
Driver is the system under test of Run: the two driver interfaces of the queue contract, usually implemented by one value.
type Fake ¶
type Fake struct {
// contains filtered or unexported fields
}
Fake is an in-memory queue.DispatcherDriver and queue.ConsumerDriver for tests. It records every dispatched message, keeps undelivered messages pending, and executes them only when a test calls Fake.Drain or runs a consumer on it. A Fake is safe for concurrent use.
func (*Fake) AssertDispatched ¶
AssertDispatched fails t unless a message for job was dispatched whose decoded payload satisfies every predicate. Without predicates any payload matches.
Example ¶
ExampleFake_AssertDispatched asserts on a decoded payload.
package main
import (
"context"
"fmt"
"testing"
"arkoren.dev/core/service/queue"
"arkoren.dev/core/service/queue/queuetest"
)
func main() {
t := &exampleT{}
job := queue.NewJob[int]("count")
fake := queuetest.New()
if err := queue.NewDispatcher(fake).Dispatch(context.Background(), job, 3); err != nil {
fmt.Println(err)
return
}
fake.AssertDispatched(t, job, func(count int) bool {
return count == 3
})
fake.AssertNotDispatched(t, job, func(count int) bool {
return count == 4
})
fake.AssertDispatchedCount(t, job, 1)
fmt.Println("failures:", t.failures)
}
// exampleT is a minimal testing.TB for examples, counting failures.
type exampleT struct {
recorder
failures int
}
// Errorf counts the failure.
func (example *exampleT) Errorf(string, ...any) {
example.failures++
}
// recorder is a testing.TB that records Errorf calls instead of failing.
type recorder struct {
testing.TB
failures []string
}
// Helper does nothing.
func (recorder *recorder) Helper() {
}
// Errorf records the formatted failure.
func (recorder *recorder) Errorf(format string, args ...any) {
recorder.failures = append(recorder.failures, fmt.Sprintf(format, args...))
}
Output: failures: 0
func (*Fake) AssertDispatchedCount ¶
AssertDispatchedCount fails t unless exactly n messages were dispatched for job.
func (*Fake) AssertNotDispatched ¶
AssertNotDispatched fails t if a message for job was dispatched whose decoded payload satisfies every predicate. Without predicates any payload matches.
func (*Fake) AssertNothingDispatched ¶
AssertNothingDispatched fails t if any message was dispatched.
func (*Fake) Consume ¶
Consume delivers pending messages of name to handler until ctx is canceled or handler fails. During Fake.Drain it instead returns nil once no message is left to run. Delayed retries are delivered when they fall due. A handler that returns an error or leaves a delivery unsettled fails that delivery and ends Consume.
func (*Fake) Dispatch ¶
Dispatch records message and keeps it pending for delivery. A missing ID is generated. It returns an injected failure, if any, without recording.
func (*Fake) Dispatched ¶
Dispatched returns a copy of every message accepted since the last Reset, in dispatch order, including messages already delivered.
func (*Fake) Drain ¶
Drain synchronously runs the pending messages of every queue that consumer handles until none is left to run. consumer must be backed by the Fake. Messages of one queue run in arrival order through the registered handlers, so retries, rejections, failures, and decoding errors are settled exactly as in production. Queues run concurrently, and Drain repeats the pass while a handler's dispatch to another queue keeps producing work.
By default only messages due now run, including immediate retries; messages delayed by queue.RetryAfter or a backoff stay pending until WithScheduled is used. A message delivered more than DefaultMaxAttempts times fails the Drain with ErrDrainLimit.
Drain returns the first settlement failure, queue.ErrNoQueues when consumer handles no job, or the error of ctx.
Example:
_ = fake.Drain(ctx, consumer)
Example ¶
ExampleFake_Drain runs a delayed retry without waiting for its scheduled time.
package main
import (
"context"
"fmt"
"time"
"arkoren.dev/core/service/queue"
"arkoren.dev/core/service/queue/queuetest"
)
func main() {
ctx := context.Background()
job := queue.NewJob[string]("greet")
fake := queuetest.New()
consumer := queue.NewConsumer(fake)
attempts := 0
if err := consumer.Handle(job, func(context.Context, string) error {
attempts++
if attempts == 1 {
return queue.RetryAfter(fmt.Errorf("later"), time.Hour)
}
return nil
}); err != nil {
fmt.Println(err)
return
}
if err := queue.NewDispatcher(fake).Dispatch(ctx, job, "world"); err != nil {
fmt.Println(err)
return
}
if err := fake.Drain(ctx, consumer); err != nil {
fmt.Println(err)
return
}
fmt.Println("attempts:", attempts, "pending:", len(fake.Pending()))
if err := fake.Drain(ctx, consumer, queuetest.WithScheduled()); err != nil {
fmt.Println(err)
return
}
fmt.Println("attempts:", attempts, "pending:", len(fake.Pending()))
}
Output: attempts: 1 pending: 1 attempts: 2 pending: 0
func (*Fake) Failed ¶
Failed returns a copy of the messages settled with queue.Delivery.Fail or abandoned by a handler that returned without settling them.
func (*Fake) Payloads ¶
Payloads decodes the payloads of the dispatched messages for job, in dispatch order. It returns an error when a payload does not decode into T.
func (*Fake) Pending ¶
Pending returns a copy of the messages not yet delivered or acknowledged, including delayed retries, across all queues.
func (*Fake) Rejected ¶
Rejected returns a copy of the messages settled with queue.Delivery.Reject.
func (*Fake) Reset ¶
func (fake *Fake) Reset()
Reset forgets all recorded and pending messages and clears injected failures.
type Option ¶
type Option func(*suiteConfig)
Option declares a capability the driver under test lacks, so the suite skips the behavior that depends on it. Options never change what a behavior means.
func WithSynctest ¶
func WithSynctest() Option
WithSynctest runs every subtest inside a testing/synctest bubble, so waits take no real time and the retry delay is checked exactly. newDriver is called inside the bubble. Use it for drivers that live in the process and use the time package, such as the memory driver. Never use it with a driver that talks to an external system.
Examples ¶
queuetest.Run(t, func(t *testing.T) queuetest.Driver {
driver := memoryqueue.New()
return queuetest.Driver{Dispatcher: driver, Consumer: driver}
}, queuetest.WithSynctest())
func WithoutRetryDelay ¶
func WithoutRetryDelay() Option
WithoutRetryDelay declares that the driver cannot delay a retry, as the contract allows for queue.RetryAfterError. The suite then skips the delayed-retry check.
Source Files