Skip to content

queue

Directory actions

More options

Directory actions

More options

Latest commit

 

History

History
 
 

queue

README.md

Queue Abstractions

Vendor-agnostic interfaces for pub/sub messaging systems.

Interfaces

Queue

Creates publishers and subscribers.

Publisher

Publishes messages to topics.

type Publisher interface {
    Publish(ctx context.Context, topic string, message queue.Message) error
    Close() error
}

Subscriber

Consumes messages from topics with per-subscription configuration.

type Subscriber interface {
    Subscribe(ctx context.Context, topic string, config SubscriptionConfig) (<-chan Delivery, error)
    Close() error
}

Delivery

Message with acknowledgment operations.

type Delivery interface {
    Message() queue.Message
    Ack(ctx context.Context) error
    Nack(ctx context.Context, requeueAfterMillis int64) error
    Reject(ctx context.Context, reason string) error
    ExtendVisibilityTimeout(ctx context.Context, durationMillis int64) error
    DeliveryID() string
    Attempt() int
    ReceivedAt() int64
    Metadata() map[string]string
}
  • Ack — message processed successfully, remove from queue
  • Nack — processing failed, requeue for retry after delay
  • Reject — poison pill, move to DLQ (or ack if DLQ disabled)
  • ExtendVisibilityTimeout — extend processing window for long-running work

SubscriptionConfig

Per-subscription configuration for polling, batching, leasing, retries, and DLQ:

cfg := extqueue.DefaultSubscriptionConfig("worker-1", "consumer-group")
cfg.PollIntervalMs = 50
cfg.BatchSize = 20
cfg.VisibilityTimeoutMs = 60000
cfg.Retry.MaxAttempts = 3
cfg.DLQ.Enabled = true

See subscription_config.go for all fields and defaults.

Usage

q, _ := NewQueue(config)
defer q.Close()

// Publish
pub := q.Publisher()
msg := queue.NewMessage("id", []byte("payload"), "partition-key", nil)
pub.Publish(ctx, "topic", msg)

// Subscribe
sub := q.Subscriber()
cfg := extqueue.DefaultSubscriptionConfig("worker-1", "consumer-group")
deliveries, _ := sub.Subscribe(ctx, "topic", cfg)
for delivery := range deliveries {
    if err := process(delivery.Message().Payload); err != nil {
        delivery.Nack(ctx, 0)  // Retry
        continue
    }
    delivery.Ack(ctx)
}

Implementing a Backend

  1. Create extension/queue/{backend}/ directory
  2. Implement Queue, Publisher, Subscriber, Delivery interfaces
  3. Map queue.Message to backend format

See extension/queue/mysql/ for the reference implementation.