rabbitmq

package
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 26, 2026 License: MIT Imports: 14 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ExchangeDirect  = "direct"
	ExchangeFanout  = "fanout"
	ExchangeTopic   = "topic"
	ExchangeHeaders = "headers"
)

Predefined RabbitMQ exchange types for use in configuration. - ExchangeDirect: Direct exchange type. - ExchangeFanout: Fanout exchange type. - ExchangeTopic: Topic exchange type. - ExchangeHeaders: Headers exchange type.

Variables

This section is empty.

Functions

This section is empty.

Types

type DeadLetterConfig 

type DeadLetterConfig struct {
	Exchange            string
	Queue               string
	RoutingKey          string
	MaxDeliveryAttempts uint32
}

DeadLetterConfig owns RabbitMQ terminal routing and bounded delivery policy.

type Option 

type Option func(*options)

Option is a functional option type for configuring the options struct. It allows for flexible and composable configuration of RabbitMQ workers and queues.

func WithAddr 

func WithAddr(addr string) Option

WithAddr sets the AMQP server URI.

Parameters: - addr: The AMQP URI to connect to.

Returns: - Option: Functional option to set the address.

func WithAutoAck 

func WithAutoAck(val bool) Option

WithAutoAck enables or disables automatic message acknowledgment.

Parameters: - val: true to enable auto-ack, false to disable.

Returns: - Option: Functional option to set autoAck.

func WithDeadLetter 

func WithDeadLetter(config DeadLetterConfig) Option

WithDeadLetter configures the durable exchange, queue, routing key, and maximum broker delivery attempts used for terminal RabbitMQ failures.

func WithExchangeName 

func WithExchangeName(val string) Option

WithExchangeName sets the name of the AMQP exchange.

Parameters: - val: The exchange name.

Returns: - Option: Functional option to set the exchange name.

Exchanges are AMQP 0-9-1 entities where messages are sent to. Exchanges take a message and route it into zero or more queues.

func WithExchangeType 

func WithExchangeType(val string) Option

WithExchangeType sets the type of the AMQP exchange.

Parameters: - val: The exchange type (direct, fanout, topic, headers).

Returns: - Option: Functional option to set the exchange type.

The routing algorithm used depends on the exchange type and rules called bindings. AMQP 0-9-1 brokers provide four exchange types: - Direct exchange (Empty string) and amq.direct - Fanout exchange amq.fanout - Topic exchange amq.topic - Headers exchange amq.match (and amq.headers in RabbitMQ)

func WithLogger 

func WithLogger(l queue.Logger) Option

WithLogger sets a custom logger for the worker or queue.

Parameters: - l: The logger instance.

Returns: - Option: Functional option to set the logger.

func WithPublishTimeout 

func WithPublishTimeout(timeout time.Duration) Option

WithPublishTimeout bounds each RabbitMQ publish operation.

func WithQueue 

func WithQueue(val string) Option

WithQueue sets the name of the queue to use.

Parameters: - val: The queue name.

Returns: - Option: Functional option to set the queue name.

func WithReconnectConfig 

func WithReconnectConfig(config ReconnectConfig) Option

WithReconnectConfig configures connection retry timing.

func WithRequestTimeout 

func WithRequestTimeout(timeout time.Duration) Option

WithRequestTimeout sets how long Request waits for a RabbitMQ delivery.

func WithRoutingKey 

func WithRoutingKey(val string) Option

WithRoutingKey sets the AMQP routing key.

Parameters: - val: The routing key.

Returns: - Option: Functional option to set the routing key.

func WithRunFunc 

func WithRunFunc(fn func(context.Context, core.TaskMessage) error) Option

WithRunFunc sets the function to execute for each task.

Parameters: - fn: The function to run for each task message.

Returns: - Option: Functional option to set the run function.

func WithTag 

func WithTag(val string) Option

WithTag sets the consumer tag for the worker.

Parameters: - val: The consumer tag.

Returns: - Option: Functional option to set the tag.

type ReconnectConfig 

type ReconnectConfig struct {
	MaxRetries   int
	InitialDelay time.Duration
	MaxDelay     time.Duration
}

ReconnectConfig defines the retry policy for RabbitMQ connection.

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

Worker struct implements the core.Worker interface for RabbitMQ. It manages the AMQP connection, channel, and task consumption. Fields: - conn: AMQP connection to RabbitMQ server. - channel: AMQP channel for communication. - stop: Channel to signal worker shutdown. - stopFlag: Atomic flag to indicate if the worker is stopped. - stopOnce: Ensures shutdown logic runs only once. - startOnce: Ensures consumer initialization runs only once. - opts: Configuration options for the worker. - tasks: Channel for receiving AMQP deliveries (tasks).

func NewWorker 

func NewWorker(opts ...Option) *Worker

NewWorker creates and initializes a new Worker instance with the provided options. It establishes a connection to RabbitMQ, sets up the channel, and declares the exchange. If any step fails, it logs a fatal error and terminates the process.

Parameters: - opts: Variadic list of Option functions to configure the worker.

Returns: - Pointer to the initialized Worker.

func NewWorkerE 

func NewWorkerE(opts ...Option) (*Worker, error)

NewWorkerE creates a worker and returns connection and setup errors.

func (*Worker) BackendName 

func (*Worker) BackendName() string

BackendName identifies RabbitMQ in lifecycle events.

func (*Worker) Queue 

func (w *Worker) Queue(job core.TaskMessage) error

Queue publishes a new task message to the RabbitMQ exchange. If the worker is stopped, it returns queue.ErrQueueShutdown.

Parameters: - job: The task message to be published.

Returns: - error: Any error encountered during publishing, or nil on success.

func (*Worker) QueueName 

func (w *Worker) QueueName() string

QueueName returns the configured RabbitMQ queue.

func (*Worker) Request 

func (w *Worker) Request() (core.TaskMessage, error)

Request retrieves a new task message from the queue. It starts the consumer if not already started, waits for a message, and unmarshals it into a job.Message. If no message is received within a timeout, it returns queue.ErrNoTaskInQueue.

Returns: - core.TaskMessage: The received task message, or nil if none. - error: Any error encountered, or queue.ErrNoTaskInQueue if no task is available.

func (*Worker) Run 

func (w *Worker) Run(ctx context.Context, task core.TaskMessage) error

Run executes the worker's task processing function. It delegates the actual task handling to the configured runFunc.

Parameters: - ctx: Context for cancellation and timeout. - task: The task message to process.

Returns: - error: Any error returned by the runFunc.

func (*Worker) Shutdown 

func (w *Worker) Shutdown() (err error)

Shutdown gracefully stops the worker. It ensures shutdown logic runs only once, cancels the consumer, and closes the AMQP connection. If the worker is already stopped, it returns queue.ErrQueueShutdown.

Returns: - error: Any error encountered during shutdown, or nil on success.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.