Queue is a Go library that gives Hub pipeline and workflow services one API for publishing, consuming, routing, retrying, and dead-lettering messages across different message brokers.
| Broker | Implementation | Guide |
|---|---|---|
| RabbitMQ | amqp091-go |
RabbitMQ guide |
| Kafka | Confluent Kafka Go client | Kafka guide |
| Azure Event Hubs | Kafka-compatible endpoint | Event Hubs guide |
| Amazon SQS | AWS SDK for Go v2 | SQS guide |
The broker guides contain provider-specific setup, configuration, security, delivery semantics, examples, and troubleshooting notes.
go get github.com/uug-ai/queueEach broker exposes a typed options builder that implements QueueOptions.
Options are validated when the client is created.
New(options) selects the broker from the concrete options type and returns a
Queue containing the validated options and a QueueInterface client.
func connect(options queue.QueueOptions) (*queue.Queue, error) {
client, err := queue.New(options)
if err != nil {
return nil, err
}
if err := client.Client.Connect(); err != nil {
return nil, err
}
return client, nil
}Call Close when the client is no longer needed.
QueueInterface provides the shared broker lifecycle and messaging operations:
Connect,Reconnect, andClosePublishandPublishWithDelayReadMessagesandRouteMessagesAddToDeadletterandDisasterRecoverySetDisasterRecoveryHandlerLoadMessages
Dead-letter inspection and replay are deliberately exposed through the separate
DeadLetterAdmin interface. Runtime consumers therefore do not need
administrative methods, while operational tooling can use one contract across
all supported brokers.
RabbitMQ services that need to deliberately park raw deliveries for operational
testing can enable confirmed delivery and call ReadMessagesToDeadletter. The
method preserves the original body and acknowledges it only after the broker
confirms the dead-letter envelope.
RabbitMQ can log one structured warning before each new dead-letter publication
using RabbitOptions.Logger / SetLogger. The warning records the existing
terminal reason and available correlation identifiers, without parsing or logging
payloads. It records a request, not successful delivery.
See dead-letter request logging.
RabbitMQ also offers an optional RabbitOptions.DeadLetterObserver callback for
payload-free final DLQ publish outcomes, including failures and confirmation
status. See dead-letter publish outcomes.
Message handlers return a models.PipelineAction. The queue client maps that
action to broker operations:
| Action | Result |
|---|---|
PipelineForward |
Advance and publish the event to the configured router |
PipelineCancel |
Complete the message without forwarding |
PipelineRetry |
Republish with a bounded retry count |
PipelineError |
Publish to the configured dead-letter destination |
All providers implement at-least-once processing. Handlers and downstream writes should therefore be idempotent.
AddToDeadletter stores a versioned uug.ai/dead-letter/v1 envelope containing:
- The exact original payload, encoded safely even when it is not valid JSON
- The source queue or topic
- The dead-letter destination
- The replay destination: the configured pipeline router, or the source when no router is configured
- The failure reason, attempt count, and UTC timestamp
- Optional service and provider-neutral attributes
For compatibility with existing services, calling Publish with the configured
dead-letter destination is also enveloped automatically. Existing callers do not
need to migrate in lockstep to start recording their source queue. Runtime
payloads are always wrapped, even if their JSON resembles an envelope, so payload
data cannot choose its own replay destination. Administrative tools that need to
set source metadata use DeadLetterAdmin.PublishDeadLetter.
Existing raw dead-letter messages remain readable. They are reported with the
source unknown and require an explicit replay destination because the library
cannot safely infer where they came from.
Replay prefers an operator-supplied destination, then the destination recorded
in the envelope, and finally the source queue for envelopes written before
replayDestination was introduced. Pipeline workers therefore replay through
their configured router, which resolves the next queue from the event's
remaining stages, while non-pipeline workers preserve source-queue behavior.
DeadLetterReplayRequest.Transform can replace or skip payloads before
publication. Set BatchSize, BatchTimeout, and BatchDelay to process a
larger bounded Limit in chunks within one provider operation. The callback
receives decoded messages in replay order and must return one transformation
for each message. A skipped message remains dead-lettered, while valid messages
in the same batch continue where the provider can settle them safely. Providers
retain the current batch when transformation fails and always publish a
transformed payload before settling its original dead-letter message.
RabbitMQ keeps retained deliveries unacknowledged and SQS keeps them invisible
until the bounded operation finishes, preventing a retained message from being
scanned repeatedly across batches. Kafka offsets are contiguous per partition:
after a skipped message, later messages in that partition remain uncommitted,
while replay can continue on other partitions. Because retained RabbitMQ
deliveries and SQS receipt handles are held for the operation, callers should
treat Limit as both a work and memory safety bound. Batched SQS replay requires
either BatchTimeout or an overall context deadline so its visibility window
can safely cover the run.
The operational commands are provided by the separate
uug-ai/cli project. The underlying
DeadLetterAdmin implementations always publish first and settle the source
message only after broker acknowledgement:
- RabbitMQ uses a dedicated confirm-mode channel, then acknowledges the delivery.
- SQS waits for
SendMessage, then deletes the message using its receipt handle. - Kafka and Azure Event Hubs wait for delivery, then commit the source offset.
Inspection remains non-destructive but not side-effect-free: RabbitMQ deliveries are durably republished before the originals are acknowledged, and SQS visibility is reset after scanning. These operations may affect ordering. Kafka/Event Hubs inspection does not commit offsets.
Use the broker-specific builder and follow its guide:
Application code can depend on QueueInterface after construction, keeping
most processing logic independent of the selected broker.
Run the complete test suite:
go test ./...Verify the non-CGO build used by RabbitMQ-only applications:
CGO_ENABLED=0 go test ./...New providers should implement QueueInterface, expose a typed options builder,
validate required configuration, and include broker-free unit tests for publish,
consume, retry, and dead-letter behavior.
This project is licensed under the MIT License. See LICENSE.