Merge pull request #1 from afaf-tech/fix/consumer-option

fix: queue option and godoc

Author: afaf-tech

README.md

@@ -10,7 +10,6 @@ Explores the hypothesis of using a single connection for many channels, both for
 
 ## Requirements
 - go 1.21
-- docker
 
 ### Installation
 

consumer.go

@@ -8,11 +8,57 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
+// Message represents a message received by a consumer from a RabbitMQ queue.
+// It includes the message body and methods for acknowledging, negatively acknowledging,
+// or rejecting the message, as well as metadata about the message's origin.
+type Message struct {
+	// Body is the content of the message received from the queue.
+	// It contains the actual data sent by the producer.
+	Body []byte
+
+	// Ack is a function that, when called, acknowledges the message.
+	// It informs RabbitMQ that the message has been successfully processed.
+	Ack func()
+
+	// Nack is a function that, when called, negatively acknowledges the message.
+	// It informs RabbitMQ that the message was not processed successfully, and depending on
+	// the RabbitMQ configuration, it may be re-delivered or discarded.
+	Nack func()
+
+	// Reject is a function that, when called, rejects the message.
+	// It informs RabbitMQ that the message was not processed and does not want it to be re-delivered.
+	Reject func()
+
+	// QueueName is the name of the RabbitMQ queue from which the message was consumed.
+	// It helps to identify the source of the message.
+	QueueName string
+
+	// ConsumerID is the unique identifier of the consumer that received the message.
+	// It can be used for logging or to distinguish between multiple consumers.
+	ConsumerID string
+}
+
+// Consumer represents an AMQP consumer that consumes messages from a RabbitMQ queue.
+// It manages the connection, channel, and queue configurations for the consumer.
 type Consumer struct {
-	conn        *Connection
-	channel     *amqp.Channel
-	queueName   string
-	name        string
+	// conn is the underlying AMQP connection used by the consumer to communicate with RabbitMQ.
+	// It allows for opening channels and managing message consumption.
+	conn *Connection
+
+	// channel is the AMQP channel through which the consumer will consume messages from the queue.
+	// A channel is used to send and receive messages between the consumer and the broker.
+	channel *amqp.Channel
+
+	// queueName is the name of the RabbitMQ queue from which this consumer will consume messages.
+	// It helps the consumer identify which queue to connect to.
+	queueName string
+
+	// name is the unique name of this consumer, used for identification purposes.
+	// It is often used for logging or distinguishing between multiple consumers.
+	name string
+
+	// queueConfig holds the configuration options for the queue, such as whether the queue is durable,
+	// auto-delete, and other properties that affect its behavior.
 	queueConfig QueueOptions
 }
 
@@ -48,41 +94,40 @@ func NewConsumer(conn *Connection, name, queue string, options QueueOptions) (*C
 	}, nil
 }
 
-// Start begins the consumer's main loop
-func (c *Consumer) Start(ctx context.Context, fn func(*Connection, string, string, *<-chan amqp.Delivery)) {
-	var err error
-
+func (c *Consumer) Start(ctx context.Context, handler func(ctx context.Context, msg *Message)) {
 	for {
-		// Ensure connection and channel are valid before starting consumption
-		err = c.checkConnectionAndChannel()
-		if err != nil {
+		if err := c.checkConnectionAndChannel(); err != nil {
 			time.Sleep(c.conn.config.RetryDuration)
 			continue
 		}
 
-		// Register the consumer and handle any errors
-		msgs, err := c.channel.ConsumeWithContext(
+		deliveries, err := c.channel.ConsumeWithContext(
 			ctx,
-			c.queueName, // queue
-			c.name,      // consumer
-			false,       // auto-ack
-			false,       // exclusive
-			false,       // no-local
-			false,       // no-wait
-			nil,         // args
+			c.queueName,
+			c.name,
+			c.queueConfig.AutoAck,
+			c.queueConfig.Exclusive,
+			c.queueConfig.NoLocal,
+			c.queueConfig.NoWait,
+			c.queueConfig.Args,
 		)
 		if err != nil {
-			log.Printf("[%s] Failed to register a consumer: %v. Retrying in 5 seconds.", c.name, err)
+			log.Printf("[%s] Failed to start consuming: %v. Retrying...", c.name, err)
 			time.Sleep(c.conn.config.RetryDuration)
 			continue
 		}
 
-		// Execute the user-defined function
-		fn(c.conn, c.name, c.queueName, &msgs)
+		for d := range deliveries {
+			msg := &Message{
+				Body:       d.Body,
+				Ack:        func() { d.Ack(false) },
+				Nack:       func() { d.Nack(false, true) },
+				Reject:     func() { d.Reject(false) },
+				QueueName:  c.queueName,
+				ConsumerID: c.name,
+			}
 
-		// Close the channel after consuming messages (if necessary)
-		if c.channel != nil {
-			c.channel.Close()
+			handler(ctx, msg)
 		}
 	}
 }

producer.go

@@ -8,11 +8,23 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
+// Producer represents an AMQP producer that sends messages to a specified RabbitMQ queue.
 type Producer struct {
-	conn      *Connection
-	channel   *amqp.Channel
+	// conn is the underlying AMQP connection that this producer uses to communicate with the RabbitMQ broker.
+	// It provides the connection functionality for creating channels and sending messages.
+	conn *Connection
+
+	// channel is the AMQP channel used by this producer to publish messages to RabbitMQ.
+	// A channel is a lightweight connection to the broker and is used to send and receive messages.
+	channel *amqp.Channel
+
+	// queueName is the name of the RabbitMQ queue to which this producer sends messages.
+	// It identifies the target queue for message delivery.
 	queueName string
-	name      string
+
+	// name is the unique name of this producer, often used for logging or identifying the producer.
+	// It helps distinguish multiple producers in the system.
+	name string
 }
 
 // NewProducer initializes a new Producer instance and declares the queue with the specified options
@@ -40,14 +52,35 @@ func NewProducer(conn *Connection, name, queue string) (*Producer, error) {
 	}, nil
 }
 
+// PublishOptions represents the configuration options for publishing a message to RabbitMQ.
 type PublishOptions struct {
-	Exchange    string     // The exchange to publish the message to
-	RoutingKey  string     // The routing key to use for the message
-	ContentType string     // The content type of the message (e.g., "text/plain")
-	Body        []byte     // The actual message body
-	Mandatory   bool       // Whether the message is mandatory
-	Immediate   bool       // Whether the message is immediate
-	Headers     amqp.Table // Optional headers for the message
+	// Exchange is the name of the exchange to which the message will be published.
+	// The exchange determines how the message will be routed to queues.
+	Exchange string
+
+	// RoutingKey is the routing key used by the exchange to decide where to route the message.
+	// The value depends on the type of exchange (e.g., direct, topic).
+	RoutingKey string
+
+	// ContentType specifies the content type of the message. It helps the consumer interpret the message body.
+	// For example, "text/plain" or "application/json".
+	ContentType string
+
+	// Body is the actual message content, represented as a byte slice.
+	// This is the payload of the message being sent to the RabbitMQ exchange.
+	Body []byte
+
+	// Mandatory indicates whether the message is mandatory.
+	// If true, RabbitMQ will return the message to the producer if it cannot be routed to a queue.
+	Mandatory bool
+
+	// Immediate indicates whether the message is immediate.
+	// If true, RabbitMQ will try to deliver the message to a consumer immediately, if possible.
+	Immediate bool
+
+	// Headers is an optional map of headers that can be included with the message.
+	// These headers can carry metadata or additional information to help the consumer process the message.
+	Headers amqp.Table
 }
 
 // Publish sends a message to the queue with retry logic and context support

rabbitmq.go

@@ -9,13 +9,35 @@ import (
 	amqp "github.com/rabbitmq/amqp091-go"
 )
 
-// QueueOptions defines RabbitMQ queue configurations
+// QueueOptions represents the configuration options for declaring a queue in RabbitMQ.
 type QueueOptions struct {
-	Durable    bool
+	// Durable specifies whether the queue should survive broker restarts. If true, the queue will
+	// be durable, meaning it will persist even if RabbitMQ crashes or restarts.
+	Durable bool
+
+	// AutoAck enables or disables automatic message acknowledgment. If true, messages are automatically
+	// acknowledged by the broker once they are received by the consumer.
+	AutoAck bool
+
+	// AutoDelete determines whether the queue should be automatically deleted when no consumers are
+	// connected. If true, the queue will be deleted when the last consumer disconnects.
 	AutoDelete bool
-	Exclusive  bool
-	NoWait     bool
-	Args       amqp.Table
+
+	// Exclusive makes the queue private to the connection that created it. If true, the queue can only
+	// be used by the connection that declared it and will be deleted once that connection closes.
+	Exclusive bool
+
+	// NoWait prevents the server from sending a response to the queue declaration. If true, the declaration
+	// will not wait for an acknowledgment from the server and no error will be returned if the queue already exists.
+	NoWait bool
+
+	// NoLocal prevents the delivery of messages to the connection that published them. If true, messages
+	// will not be delivered to the connection that created the queue.
+	NoLocal bool
+
+	// Args allows additional arguments to be passed when declaring the queue. This can be used for advanced
+	// RabbitMQ configurations, such as setting arguments for policies or defining queue TTLs (Time-To-Live).
+	Args amqp.Table
 }
 
 const (
@@ -24,21 +46,48 @@ const (
 	defaultMaxChannels   = 10                                   // Default maximum number of channels
 )
 
-// Config holds the configuration for connecting to RabbitMQ
+// Config holds the configuration options for establishing a RabbitMQ connection and managing
+// consumer channels.
 type Config struct {
-	URI           string        // URI for RabbitMQ connection (e.g., "amqp://guest:guest@localhost:5672/")
-	RetryDuration time.Duration // Time to wait before retrying a failed connection attempt
-	AMQPConfig    *amqp.Config  // AMQP-specific configuration, can be nil to use defaults
-	MaxChannels   int           // Maximum number of channels allowed (default is 10)
+	// URI is the connection string for RabbitMQ. It should follow the AMQP URI format, e.g.,
+	// "amqp://guest:guest@localhost:5672/". It is used to establish the connection to the RabbitMQ broker.
+	URI string
+
+	// RetryDuration specifies the time duration to wait before retrying a failed connection attempt.
+	// This is useful to implement a backoff strategy in case of temporary network issues.
+	RetryDuration time.Duration
+
+	// AMQPConfig holds AMQP-specific configuration options. If nil, default AMQP configurations
+	// are used. This can include settings like heartbeat intervals and other advanced AMQP features.
+	AMQPConfig *amqp.Config
+
+	// MaxChannels defines the maximum number of channels that can be opened to the RabbitMQ server.
+	// If the value is 0 or negative, the default is used (which may be 10 channels).
+	MaxChannels int
 }
 
-// Connection wraps the actual AMQP connection and provides reconnection logic
+// Connection wraps the actual AMQP connection and provides reconnection logic, ensuring
+// that the connection and channels remain active or are re-established when necessary.
 type Connection struct {
+	// Connection is the underlying AMQP connection provided by the AMQP client library.
+	// It provides the basic connection functionalities such as opening channels, closing the connection, etc.
 	*amqp.Connection
-	config       *Config         // Custom configuration for the connection
-	mu           sync.RWMutex    // Mutex to synchronize access to the connection
-	reconnecting bool            // Flag to indicate ongoing reconnection
-	channels     []*amqp.Channel // Slice to hold active channels
+
+	// config holds the custom configuration settings for the RabbitMQ connection, such as URI, retry duration, etc.
+	// These settings are used for connection management and reconnection attempts.
+	config *Config
+
+	// mu is a read-write mutex used to synchronize access to the connection and channels,
+	// ensuring thread-safe operations for shared resources.
+	mu sync.RWMutex
+
+	// reconnecting is a flag indicating whether a reconnection attempt is currently in progress.
+	// This is used to prevent multiple concurrent reconnection attempts.
+	reconnecting bool
+
+	// channels holds a slice of active channels associated with the current connection.
+	// Channels are used to send and receive messages within a specific connection context.
+	channels []*amqp.Channel
 }
 
 // NewConnection creates a new Connection object and initializes it with the provided configuration