Documentation
¶
Overview ¶
Package broker owns memory queue admission, delivery and delayed retries.
Index ¶
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 = errors.New("memory queue driver is closed") // ErrInvalidQueue is returned when a queue name is empty. // // err := driver.Dispatch(ctx, queue.Message{}) // fmt.Println(errors.Is(err, memoryqueue.ErrInvalidQueue)) ErrInvalidQueue = errors.New("invalid job queue") // ErrInvalidHandler is returned when Consume is given a nil handler. // // err := driver.Consume(ctx, "work", nil) // fmt.Println(errors.Is(err, memoryqueue.ErrInvalidHandler)) ErrInvalidHandler = errors.New("invalid job delivery handler") // 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 = errors.New("job delivery already settled") // 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 = errors.New("job delivery was not settled") )
Functions ¶
This section is empty.
Types ¶
type Broker ¶
type Broker struct {
// RetryDelay is the delay applied to retries that request none.
RetryDelay time.Duration
// Logger records delivery failures.
Logger *slog.Logger
// contains filtered or unexported fields
}
Broker implements queue.DispatcherDriver and queue.ConsumerDriver with in-memory data structures. It is non-durable: process exit or crash loses queued jobs. A Broker 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 (*Broker) Consume ¶
func (driver *Broker) 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. Broker.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)
}
func (*Broker) 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 (*Broker) Start ¶
Start reports ErrDriverClosed after Broker.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 (*Broker) 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