Documentation
¶
Overview ¶
Package queue defines task publication, Kafka producer and consumer adapters, transactional outbox support, and process-local task tracking. Durable work must use a production queue or outbox-backed adapter.
Index ¶
- Variables
- func AsyncTask(name string, maxRetries int, fn func(ctx context.Context) error) string
- func OutboxMigrations(startVersion int, dialect string) ([]dbcore.Migration, error)
- type KafkaConfig
- type KafkaConsumer
- type KafkaConsumerConfig
- type KafkaMessage
- type KafkaProducer
- type KafkaProducerConfig
- type KafkaSASLConfig
- type KafkaWorker
- type Outbox
- type OutboxConfig
- type OutboxDatabase
- type OutboxMessage
- type OutboxPublisher
- type Queue
- type TaskInfo
- type TaskStatus
- type TaskTracker
Constants ¶
This section is empty.
Variables ¶
var ( // ErrKafkaNotConfigured indicates an uninitialized Kafka producer or consumer. ErrKafkaNotConfigured = errors.New("queue: Kafka provider is not configured") // ErrKafkaConsumerRun indicates that Run was called concurrently. ErrKafkaConsumerRun = errors.New("queue: Kafka consumer is already running") )
var ErrQueueNotConfigured = errors.New("queue: no durable queue provider is configured")
ErrQueueNotConfigured indicates that a Queue has no durable publisher.
Functions ¶
Types ¶
type KafkaConfig ¶
type KafkaConfig struct {
Brokers []string
ClientID string
TLS *tls.Config
SASL KafkaSASLConfig
DialTimeout time.Duration
DeliveryTimeout time.Duration
MaxMessageBytes int
MaxBufferedBytes int
RecordRetries int
}
KafkaConfig defines shared broker, security, timeout, retry, and capacity policy.
type KafkaConsumer ¶
type KafkaConsumer struct {
// contains filtered or unexported fields
}
KafkaConsumer processes and explicitly commits records for one consumer group.
func NewKafkaConsumer ¶
func NewKafkaConsumer(ctx context.Context, config KafkaConsumerConfig) (*KafkaConsumer, error)
NewKafkaConsumer validates, connects, and verifies a Kafka group consumer.
func (*KafkaConsumer) Close ¶
func (consumer *KafkaConsumer) Close(ctx context.Context) error
Close permits rebalance and waits for the consumer client to close.
func (*KafkaConsumer) Run ¶
func (consumer *KafkaConsumer) Run(ctx context.Context, handler func(context.Context, KafkaMessage) error) error
Run blocks until cancellation, closure, or a consumption failure.
type KafkaConsumerConfig ¶
type KafkaConsumerConfig struct {
KafkaConfig
Topics []string
GroupID string
DLQTopic string
MaxPollRecords int
MaxHandlerRetries int
RetryBackoff time.Duration
}
KafkaConsumerConfig defines topics, group, retry, batching, and dead-letter policy.
type KafkaMessage ¶
type KafkaMessage struct {
Topic string
Key []byte
Value []byte
Headers map[string][]byte
Partition int32
Offset int64
Time time.Time
}
KafkaMessage is the bounded application view of one consumed record.
type KafkaProducer ¶
type KafkaProducer struct {
// contains filtered or unexported fields
}
KafkaProducer synchronously publishes fully acknowledged records.
func NewKafkaProducer ¶
func NewKafkaProducer(ctx context.Context, config KafkaProducerConfig) (*KafkaProducer, error)
NewKafkaProducer validates, connects, and verifies a Kafka producer.
func (*KafkaProducer) Close ¶
func (producer *KafkaProducer) Close(ctx context.Context) error
Close flushes pending records within ctx and releases the producer.
func (*KafkaProducer) Publish ¶
func (producer *KafkaProducer) Publish( ctx context.Context, topic string, key, payload []byte, headers map[string][]byte, ) error
Publish sends a bounded record and waits for broker acknowledgement.
func (*KafkaProducer) PublishMessage ¶
func (producer *KafkaProducer) PublishMessage(ctx context.Context, key string, payload []byte) error
PublishMessage sends a record to the producer's default topic.
type KafkaProducerConfig ¶
type KafkaProducerConfig struct {
KafkaConfig
Topic string
}
KafkaProducerConfig adds a validated default topic to KafkaConfig.
type KafkaSASLConfig ¶
KafkaSASLConfig defines an optional PLAIN, SCRAM-SHA-256, or SCRAM-SHA-512 identity.
type KafkaWorker ¶
type KafkaWorker = KafkaProducer
KafkaWorker is retained as the default-topic producer type.
func NewKafkaWorker ¶
func NewKafkaWorker(brokers []string, topic string) *KafkaWorker
NewKafkaWorker is retained for source compatibility but intentionally fails closed. Applications should construct a verified producer with NewKafkaProducer.
type Outbox ¶
type Outbox struct {
// contains filtered or unexported fields
}
Outbox claims and publishes persisted messages with at-least-once delivery.
func NewOutbox ¶
func NewOutbox(database OutboxDatabase, publisher OutboxPublisher, config OutboxConfig) (*Outbox, error)
NewOutbox validates dependencies and applies bounded defaults.
func (*Outbox) Enqueue ¶
func (outbox *Outbox) Enqueue(ctx context.Context, tx *sql.Tx, message OutboxMessage) (string, error)
Enqueue inserts an event using the caller's business transaction. Delivery is at least once: consumers should deduplicate using the gpp-outbox-id header.
func (*Outbox) PublishPending ¶
PublishPending claims and attempts one bounded batch of available messages.
type OutboxConfig ¶
type OutboxConfig struct {
BatchSize int
PollInterval time.Duration
LeaseDuration time.Duration
RetryBackoff time.Duration
MaxAttempts int
}
OutboxConfig defines batch, polling, lease, and retry bounds.
type OutboxDatabase ¶
type OutboxDatabase interface {
BeginTx(context.Context, *sql.TxOptions) (*sql.Tx, error)
Dialect() string
}
OutboxDatabase is the transaction and dialect contract required by Outbox.
type OutboxMessage ¶
type OutboxMessage struct {
ID string
Topic string
Key []byte
Payload []byte
Headers map[string][]byte
AvailableAt time.Time
}
OutboxMessage is an event inserted atomically with application data.
type OutboxPublisher ¶
type OutboxPublisher interface {
Publish(context.Context, string, []byte, []byte, map[string][]byte) error
}
OutboxPublisher publishes one keyed message with binary headers.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue handles async background worker job dispatching.
func NewWithPublisher ¶
func NewWithPublisher(publisher OutboxPublisher, topic string) (*Queue, error)
NewWithPublisher constructs a queue using an explicit durable publisher and topic.
type TaskInfo ¶
type TaskInfo struct {
ID string `json:"id"`
Name string `json:"name"`
Status TaskStatus `json:"status"`
Retries int `json:"retries"`
MaxRetries int `json:"max_retries"`
LastError string `json:"last_error,omitempty"`
CreatedAt time.Time `json:"created_at"`
CompletedAt *time.Time `json:"completed_at,omitempty"`
}
TaskInfo stores complete execution status, retries, and error details for a background task.
func GetTaskStatus ¶
GetTaskStatus retrieves the status and execution details of a background task by ID.
type TaskStatus ¶
type TaskStatus string
TaskStatus identifies process-local task execution state.
const ( // StatusPending indicates a registered task not yet running. StatusPending TaskStatus = "PENDING" // StatusRunning indicates an active task attempt. StatusRunning TaskStatus = "RUNNING" // StatusCompleted indicates successful process-local completion. StatusCompleted TaskStatus = "COMPLETED" // StatusFailed indicates exhausted retries. StatusFailed TaskStatus = "FAILED" // StatusRetrying indicates a subsequent attempt is running. StatusRetrying TaskStatus = "RETRYING" )
type TaskTracker ¶
type TaskTracker struct {
// contains filtered or unexported fields
}
TaskTracker manages process-local background task statuses and retries. State is lost on restart; use a durable queue for production work.
func NewTaskTracker ¶
func NewTaskTracker() *TaskTracker
NewTaskTracker creates a new task tracker instance.
Source Files