rabbitmq

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 identifies the infrastructure-owned durable terminal exchange, queue, routing key, and maximum delivery attempts. The adapter publishes to the exchange and routing key but does not declare or repair the queue.

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. The compatibility adapter accepts direct and topic exchanges. Fanout and headers constants remain for source compatibility but NewWorkerE rejects them.

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 WithNativeConfig ¶

func WithNativeConfig(config NativeConfig) Option

WithNativeConfig enables the compatibility adapter with explicit native connection, resource, queue-type, and stable message-identity policy.

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 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.

func NewWorker ¶

func NewWorker(opts ...Option) *Worker

NewWorker creates a compatibility worker and panics when its explicit native policy is invalid or its producer cannot be opened.

func NewWorkerE ¶

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

NewWorkerE creates a compatibility worker with an eagerly opened producer. The consumer remains unopened until Request is called.

package main

import (
	"context"
	"time"

	"github.com/faustbrian/go-queue/adapters/rabbitmq"
	"github.com/faustbrian/go-queue/core"

	rabbitmqqueue "github.com/faustbrian/go-rabbitmq-queues"
)

func main() {
	credentials := rabbitmqqueue.CredentialProviderFunc(
		func(context.Context) (rabbitmqqueue.Credentials, error) {
			return rabbitmqqueue.Credentials{Username: "worker", Password: []byte("owned-secret")}, nil
		},
	)
	_, _ = rabbitmq.NewWorkerE(
		rabbitmq.WithNativeConfig(rabbitmq.NativeConfig{
			Connection: rabbitmqqueue.ConnectionConfig{
				Endpoints:   []rabbitmqqueue.Endpoint{{Host: "rabbitmq.internal", Port: 5671}},
				VirtualHost: "/", Credentials: credentials,
				TLS:         rabbitmqqueue.TLSConfig{ServerName: "rabbitmq.internal"},
				DialTimeout: 5 * time.Second, Heartbeat: 30 * time.Second,
				Recovery: rabbitmqqueue.RecoveryPolicy{
					MaxAttempts: 8, InitialDelay: 100 * time.Millisecond, MaxDelay: 5 * time.Second,
				},
			},
			Producer: rabbitmqqueue.ProducerConfig{
				Limits: rabbitmqqueue.DefaultLimits(), MaxOutstanding: 256,
				PublishTimeout: 5 * time.Second,
			},
			Consumer: rabbitmqqueue.ConsumerConfig{
				Limits: rabbitmqqueue.DefaultLimits(),
				Queue:  rabbitmqqueue.QueueReference{Name: "jobs", Type: rabbitmqqueue.QueueQuorum},
				Name:   "jobs-worker", Prefetch: 32, Concurrency: 8,
				HandlerTimeout: time.Minute, MaxRequeues: 1,
				Failure: rabbitmqqueue.Reject(false),
			},
			MessageID: func(core.TaskMessage) (string, error) { return "stable-job-id", nil },
		}),
		rabbitmq.WithQueue("jobs"),
		rabbitmq.WithTag("jobs-worker"),
		rabbitmq.WithExchangeName("jobs.events"),
		rabbitmq.WithExchangeType(rabbitmq.ExchangeTopic),
		rabbitmq.WithRoutingKey("jobs.created"),
	)
}

Output: