Documentation
¶
Overview ¶
Package huma provides RabbitMQ consumers and publishers with connection recovery, publisher channel pooling, delayed delivery, metrics, and tracing.
Package huma is a generated GoMock package.
Index ¶
- type AMQPHeaderCarrier
- type DelayStrategy
- type ExchangeName
- type Logger
- type Message
- func (m *Message) WithBody(body []byte) *Message
- func (m *Message) WithContentType(contentType string) *Message
- func (m *Message) WithExpiration(d time.Duration) *Message
- func (m *Message) WithHeader(key string, value any) *Message
- func (m *Message) WithJSONBody(v any) *Message
- func (m *Message) WithMessageID(id string) *Message
- func (m *Message) WithPriority(priority uint8) *Message
- type Metrics
- type MockRabbitMQKit
- func (m *MockRabbitMQKit) BatchPublish(ctx context.Context, exchange, routingKey string, messages []*Message) error
- func (m *MockRabbitMQKit) DeclareQueues(ctx context.Context, queues ...QueueConfig) error
- func (m *MockRabbitMQKit) EXPECT() *MockRabbitMQKitMockRecorder
- func (m *MockRabbitMQKit) Publish(ctx context.Context, exchange, routingKey string, msg *Message) error
- func (m *MockRabbitMQKit) PublishWithDelayDLXTTL(ctx context.Context, queueName string, msg *Message) error
- func (m *MockRabbitMQKit) SetLogger(logger Logger)
- func (m *MockRabbitMQKit) SetQos(ctx context.Context, prefetchCount, prefetchSize int, global bool) error
- func (m *MockRabbitMQKit) Shutdown(ctx context.Context) error
- func (m *MockRabbitMQKit) Start(ctx context.Context, queues []QueueConfig) error
- type MockRabbitMQKitMockRecorder
- func (mr *MockRabbitMQKitMockRecorder) BatchPublish(ctx, exchange, routingKey, messages any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) DeclareQueues(ctx any, queues ...any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) Publish(ctx, exchange, routingKey, msg any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) PublishWithDelayDLXTTL(ctx, queueName, msg any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) SetLogger(logger any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) SetQos(ctx, prefetchCount, prefetchSize, global any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) Shutdown(ctx any) *gomock.Call
- func (mr *MockRabbitMQKitMockRecorder) Start(ctx, queues any) *gomock.Call
- type MsgHandler
- type NoOpLogger
- type QueueConfig
- type QueueName
- type QueueType
- type RabbitMQKit
- type RabbitMQMsg
- type RoutingKeyName
- type SDK
- func (s *SDK) BatchPublish(ctx context.Context, exchange, routingKey string, messages []*Message) error
- func (s *SDK) DeclareQueues(ctx context.Context, queues ...QueueConfig) error
- func (s *SDK) Publish(ctx context.Context, exchange, routingKey string, msg *Message) error
- func (s *SDK) PublishWithDelayDLXTTL(ctx context.Context, queueName string, msg *Message) error
- func (s *SDK) SetLogger(logger Logger)
- func (s *SDK) SetQos(ctx context.Context, prefetchCount, prefetchSize int, global bool) error
- func (s *SDK) Shutdown(ctx context.Context) error
- func (s *SDK) Start(ctx context.Context, queues []QueueConfig) error
- type SDKConfig
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type AMQPHeaderCarrier ¶
AMQPHeaderCarrier adapts amqp091.Table to satisfy propagation.TextMapCarrier, enabling OTel propagators to inject/extract trace context via AMQP message headers.
func (AMQPHeaderCarrier) Get ¶
func (c AMQPHeaderCarrier) Get(key string) string
Get returns a string header value or an empty string for other value types.
func (AMQPHeaderCarrier) Keys ¶
func (c AMQPHeaderCarrier) Keys() []string
Keys returns all header keys.
func (AMQPHeaderCarrier) Set ¶
func (c AMQPHeaderCarrier) Set(key string, value string)
Set stores a string header value.
type DelayStrategy ¶
type DelayStrategy int
DelayStrategy defines which mechanism a delay queue uses.
const ( // DelayDLXTTL uses a dead-letter exchange and queue-level TTL to implement delay. DelayDLXTTL DelayStrategy = iota )
type Logger ¶
type Logger interface {
Infof(template string, args ...any)
Infow(msg string, keysAndValues ...any)
Errorw(msg string, keysAndValues ...any)
}
Logger defines the logging interface for the SDK.
type Message ¶
type Message struct {
Body []byte
Headers amqp091.Table
ContentType string
Expiration time.Duration
Priority uint8
MessageID string
Timestamp time.Time
// contains filtered or unexported fields
}
Message holds the data for a message to be published.
func NewMessage ¶
func NewMessage() *Message
NewMessage initializes an empty message with a current timestamp.
func (*Message) WithContentType ¶
WithContentType sets the Content-Type header.
func (*Message) WithExpiration ¶
WithExpiration sets the message TTL.
func (*Message) WithHeader ¶
WithHeader adds a custom AMQP header.
func (*Message) WithJSONBody ¶
WithJSONBody JSON-encodes v and sets Content-Type to application/json.
func (*Message) WithMessageID ¶
WithMessageID sets a custom message ID.
func (*Message) WithPriority ¶
WithPriority sets the message priority.
type Metrics ¶
type Metrics interface {
IncMessagesReceived(ctx context.Context, queue string)
ObserveProcessingDuration(ctx context.Context, queue string, durationSeconds float64)
IncMessagesAcked(ctx context.Context, queue string)
IncMessagesNacked(ctx context.Context, queue string)
IncPublishSuccess(ctx context.Context, queue string)
IncPublishFailed(ctx context.Context, queue string)
IncReconnects(ctx context.Context)
}
Metrics wraps the SDK metrics interface.
type MockRabbitMQKit ¶
type MockRabbitMQKit struct {
// contains filtered or unexported fields
}
MockRabbitMQKit is a mock of RabbitMQKit interface.
func NewMockRabbitMQKit ¶
func NewMockRabbitMQKit(ctrl *gomock.Controller) *MockRabbitMQKit
NewMockRabbitMQKit creates a new mock instance.
func (*MockRabbitMQKit) BatchPublish ¶
func (m *MockRabbitMQKit) BatchPublish(ctx context.Context, exchange, routingKey string, messages []*Message) error
BatchPublish mocks base method.
func (*MockRabbitMQKit) DeclareQueues ¶
func (m *MockRabbitMQKit) DeclareQueues(ctx context.Context, queues ...QueueConfig) error
DeclareQueues mocks base method.
func (*MockRabbitMQKit) EXPECT ¶
func (m *MockRabbitMQKit) EXPECT() *MockRabbitMQKitMockRecorder
EXPECT returns an object that allows the caller to indicate expected use.
func (*MockRabbitMQKit) Publish ¶
func (m *MockRabbitMQKit) Publish(ctx context.Context, exchange, routingKey string, msg *Message) error
Publish mocks base method.
func (*MockRabbitMQKit) PublishWithDelayDLXTTL ¶
func (m *MockRabbitMQKit) PublishWithDelayDLXTTL(ctx context.Context, queueName string, msg *Message) error
PublishWithDelayDLXTTL mocks base method.
func (*MockRabbitMQKit) SetLogger ¶
func (m *MockRabbitMQKit) SetLogger(logger Logger)
SetLogger mocks base method.
func (*MockRabbitMQKit) SetQos ¶
func (m *MockRabbitMQKit) SetQos(ctx context.Context, prefetchCount, prefetchSize int, global bool) error
SetQos mocks base method.
func (*MockRabbitMQKit) Shutdown ¶
func (m *MockRabbitMQKit) Shutdown(ctx context.Context) error
Shutdown mocks base method.
func (*MockRabbitMQKit) Start ¶
func (m *MockRabbitMQKit) Start(ctx context.Context, queues []QueueConfig) error
Start mocks base method.
type MockRabbitMQKitMockRecorder ¶
type MockRabbitMQKitMockRecorder struct {
// contains filtered or unexported fields
}
MockRabbitMQKitMockRecorder is the mock recorder for MockRabbitMQKit.
func (*MockRabbitMQKitMockRecorder) BatchPublish ¶
func (mr *MockRabbitMQKitMockRecorder) BatchPublish(ctx, exchange, routingKey, messages any) *gomock.Call
BatchPublish indicates an expected call of BatchPublish.
func (*MockRabbitMQKitMockRecorder) DeclareQueues ¶
func (mr *MockRabbitMQKitMockRecorder) DeclareQueues(ctx any, queues ...any) *gomock.Call
DeclareQueues indicates an expected call of DeclareQueues.
func (*MockRabbitMQKitMockRecorder) Publish ¶
func (mr *MockRabbitMQKitMockRecorder) Publish(ctx, exchange, routingKey, msg any) *gomock.Call
Publish indicates an expected call of Publish.
func (*MockRabbitMQKitMockRecorder) PublishWithDelayDLXTTL ¶
func (mr *MockRabbitMQKitMockRecorder) PublishWithDelayDLXTTL(ctx, queueName, msg any) *gomock.Call
PublishWithDelayDLXTTL indicates an expected call of PublishWithDelayDLXTTL.
func (*MockRabbitMQKitMockRecorder) SetLogger ¶
func (mr *MockRabbitMQKitMockRecorder) SetLogger(logger any) *gomock.Call
SetLogger indicates an expected call of SetLogger.
func (*MockRabbitMQKitMockRecorder) SetQos ¶
func (mr *MockRabbitMQKitMockRecorder) SetQos(ctx, prefetchCount, prefetchSize, global any) *gomock.Call
SetQos indicates an expected call of SetQos.
type MsgHandler ¶
type MsgHandler func(ctx context.Context, queueName QueueName, msg RabbitMQMsg) error
MsgHandler is the prototype of a function that processes messages.
The target context is a context.WithCancel that is canceled on Shutdown, enabling graceful shutdown of heavy processing loops. The queueName parameter lets a single handler serve multiple queue types.
type NoOpLogger ¶
type NoOpLogger struct{}
NoOpLogger is a logger that discards all output.
func (*NoOpLogger) Errorw ¶
func (l *NoOpLogger) Errorw(_ string, _ ...any)
Errorw discards a structured error log entry.
func (*NoOpLogger) Infof ¶
func (l *NoOpLogger) Infof(_ string, _ ...any)
Infof discards a formatted informational log entry.
func (*NoOpLogger) Infow ¶
func (l *NoOpLogger) Infow(_ string, _ ...any)
Infow discards a structured informational log entry.
type QueueConfig ¶
type QueueConfig struct {
Name QueueName `yaml:"NAME" json:"NAME" mapstructure:"NAME"`
QueueType QueueType `yaml:"QUEUE_TYPE" json:"QUEUE_TYPE" mapstructure:"QUEUE_TYPE"`
Handler MsgHandler `yaml:"-" json:"-" mapstructure:"-"`
NumWorkers int `yaml:"NUM_WORKERS" json:"NUM_WORKERS" mapstructure:"NUM_WORKERS"`
IsDelayQueue bool `yaml:"IS_DELAY_QUEUE" json:"IS_DELAY_QUEUE" mapstructure:"IS_DELAY_QUEUE"`
DelayStrategy DelayStrategy `yaml:"DELAY_STRATEGY" json:"DELAY_STRATEGY" mapstructure:"DELAY_STRATEGY"`
DLXTTL time.Duration `yaml:"TTL" json:"TTL" mapstructure:"TTL"` // Only used for DLX+TTL delay queue
// MaxRedelivery controls message redelivery behavior when processing fails:
//
// MaxRedelivery = 0: INFINITE REDELIVERY (standard RabbitMQ behavior)
// Messages are requeued indefinitely until successfully processed.
// Use this for critical messages that must eventually be processed.
//
// MaxRedelivery > 0: LIMITED REDELIVERY
// Messages are redelivered up to N times, then discarded or sent to DLQ.
// Example: MaxRedelivery = 3 means the message is processed up to 4 times total
// (original delivery + 3 retries).
//
// MaxRedelivery = -1: NO SDK-REQUESTED REDELIVERY
// Handler failures are rejected without requeuing. RabbitMQ can still
// redeliver after a connection loss. Configured DLQ routing still applies.
MaxRedelivery int `yaml:"MAX_REDELIVERY" json:"MAX_REDELIVERY" mapstructure:"MAX_REDELIVERY"`
Exchange ExchangeName `yaml:"EXCHANGE" json:"EXCHANGE" mapstructure:"EXCHANGE"`
RoutingKey RoutingKeyName `yaml:"ROUTING_KEY" json:"ROUTING_KEY" mapstructure:"ROUTING_KEY"`
Durable bool `yaml:"DURABLE" json:"DURABLE" mapstructure:"DURABLE"`
AutoDelete bool `yaml:"AUTO_DELETE" json:"AUTO_DELETE" mapstructure:"AUTO_DELETE"`
ConsumerTag string `yaml:"CONSUMER_TAG" json:"CONSUMER_TAG" mapstructure:"CONSUMER_TAG"`
Exclusive bool `yaml:"EXCLUSIVE" json:"EXCLUSIVE" mapstructure:"EXCLUSIVE"`
NoWait bool `yaml:"NO_WAIT" json:"NO_WAIT" mapstructure:"NO_WAIT"`
ConsumerTimeout time.Duration `yaml:"CONSUMER_TIMEOUT" json:"CONSUMER_TIMEOUT" mapstructure:"CONSUMER_TIMEOUT"`
// EnableDLQ declares a dead-letter exchange and dead-letter queue for this queue.
// When true, two additional resources are created: an exchange named <Queue>.DLX
// and a queue named <Queue>.DLQ.
EnableDLQ bool `yaml:"ENABLE_DLQ" json:"ENABLE_DLQ" mapstructure:"ENABLE_DLQ"`
ProcessTimeout time.Duration `yaml:"PROCESS_TIMEOUT" json:"PROCESS_TIMEOUT" mapstructure:"PROCESS_TIMEOUT"`
DummyMessageEnabled bool `yaml:"DUMMY_MESSAGE_ENABLED" json:"DUMMY_MESSAGE_ENABLED" mapstructure:"DUMMY_MESSAGE_ENABLED"`
DummyMessageFrequency time.Duration `yaml:"DUMMY_MESSAGE_FREQUENCY" json:"DUMMY_MESSAGE_FREQUENCY" mapstructure:"DUMMY_MESSAGE_FREQUENCY"`
}
QueueConfig is the queue configuration struct.
type RabbitMQKit ¶
type RabbitMQKit interface {
DeclareQueues(ctx context.Context, queues ...QueueConfig) error
Publish(ctx context.Context, exchange, routingKey string, msg *Message) error
BatchPublish(ctx context.Context, exchange, routingKey string, messages []*Message) error
PublishWithDelayDLXTTL(ctx context.Context, queueName string, msg *Message) error
SetQos(ctx context.Context, prefetchCount, prefetchSize int, global bool) error
Start(ctx context.Context, queues []QueueConfig) error
Shutdown(ctx context.Context) error
SetLogger(logger Logger)
}
RabbitMQKit is the primary interface for the huma SDK.
type RabbitMQMsg ¶
RabbitMQMsg wraps the amqp091 Delivery struct.
type SDK ¶
type SDK struct {
// contains filtered or unexported fields
}
SDK is the huma provider struct.
func (*SDK) BatchPublish ¶
func (s *SDK) BatchPublish(ctx context.Context, exchange, routingKey string, messages []*Message) error
BatchPublish sends multiple messages to the same exchange/routing key.
For delay queues, publish to the exchange. For regular queues, use the routing key (queue name).
func (*SDK) DeclareQueues ¶
func (s *SDK) DeclareQueues(ctx context.Context, queues ...QueueConfig) error
DeclareQueues declares queues on RabbitMQ and stores them for reconnect replay.
func (*SDK) Publish ¶
Publish publishes a non-delay message to the given exchange or queue.
For delay queues, publish to the exchange. For regular queues, publish using the routing key (queue name).
func (*SDK) PublishWithDelayDLXTTL ¶
PublishWithDelayDLXTTL publishes a delayed message to the associated delay queue.
Only use this when DelayStrategy is DelayDLXTTL.
type SDKConfig ¶
type SDKConfig struct {
Addr string `yaml:"ADDRESS" json:"ADDRESS" mapstructure:"ADDRESS"`
VHost string `yaml:"VHOST" json:"VHOST" mapstructure:"VHOST"`
Username string `yaml:"USERNAME" json:"USERNAME" mapstructure:"USERNAME"`
Password string `yaml:"PASSWORD" json:"PASSWORD" mapstructure:"PASSWORD"`
Heartbeat time.Duration `yaml:"HEARTBEAT" json:"HEARTBEAT" mapstructure:"HEARTBEAT"` // Heartbeat interval
DialTimeout time.Duration `yaml:"DIAL_TIMEOUT" json:"DIAL_TIMEOUT" mapstructure:"DIAL_TIMEOUT"` // TCP connection timeout
ConnectionName string `yaml:"CONNECTION_NAME" json:"CONNECTION_NAME" mapstructure:"CONNECTION_NAME"` // Connection identifier
ReconnectDelay time.Duration `yaml:"RECONNECT_DELAY" json:"RECONNECT_DELAY" mapstructure:"RECONNECT_DELAY"` // How often to try reconnecting
TLSConfig *tls.Config `yaml:"-" json:"-" mapstructure:"-"` // TLS configuration for AMQPS
// Prometheus metrics options.
EnableMetrics bool `yaml:"ENABLE_METRICS" json:"ENABLE_METRICS" mapstructure:"ENABLE_METRICS"` // Enable Prometheus metrics
MetricsNamespace string `yaml:"METRICS_NAMESPACE" json:"METRICS_NAMESPACE" mapstructure:"METRICS_NAMESPACE"` // Prometheus metrics namespace
MetricsRegisterer prometheus.Registerer `yaml:"-" json:"-" mapstructure:"-"` // Prometheus collector registerer
PublisherPoolSize int `yaml:"PUBLISHER_POOL_SIZE" json:"PUBLISHER_POOL_SIZE" mapstructure:"PUBLISHER_POOL_SIZE"` // Initial channel pool size (default 5)
PublisherMaxPoolSize int `yaml:"PUBLISHER_MAX_POOL_SIZE" json:"PUBLISHER_MAX_POOL_SIZE" mapstructure:"PUBLISHER_MAX_POOL_SIZE"` // Maximum channel pool size
PublisherPoolWait time.Duration `yaml:"PUBLISHER_POOL_WAIT" json:"PUBLISHER_POOL_WAIT" mapstructure:"PUBLISHER_POOL_WAIT"` // How long a goroutine waits for a free channel before a new one is created
EnableTracing bool `yaml:"ENABLE_TRACING" json:"ENABLE_TRACING" mapstructure:"ENABLE_TRACING"` // Enable OpenTelemetry trace context propagation through AMQP headers
// MetricLabelName, when non-empty, adds one extra Prometheus label to queue metrics.
// MetricLabelValue derives its value from the message context.
MetricLabelName string `yaml:"-" json:"-" mapstructure:"-"`
MetricLabelValue func(ctx context.Context) string `yaml:"-" json:"-" mapstructure:"-"`
// InjectHeaders, if set, is called on publish to enrich AMQP headers from the context.
// The returned table is used as the final headers.
InjectHeaders func(ctx context.Context, headers amqp091.Table) amqp091.Table `yaml:"-" json:"-" mapstructure:"-"`
// ExtractContext, if set, is called on consume to derive a child context from the delivery
// (e.g. restore an app id into the context before the handler runs).
ExtractContext func(ctx context.Context, d amqp091.Delivery) context.Context `yaml:"-" json:"-" mapstructure:"-"`
}
SDKConfig is the config struct for the huma SDK.
Source Files
¶
Directories
¶
| Path | Synopsis |
|---|---|
|
examples
|
|
|
basic
command
|
|
|
dlq-and-delay
command
|
|
|
metrics
command
|
|
|
tracing
command
|