fix: linter

Author: bxcodec

.golangci.yaml

@@ -53,7 +53,7 @@ docker-clean:
 lint: $(GOLANGCI) ## Runs golangci-lint with predefined configuration
 	@echo "Applying linter"
 	golangci-lint version
-	golangci-lint run -c .golangci.yaml ./...
+	golangci-lint run -c .golangci.yml ./...
 
 .PHONY: lint lint-prepare clean build unittest 
 

Makefile

@@ -3,6 +3,7 @@ package consumer
 import (
 	"github.com/bxcodec/goqueue/internal/consumer"
 	"github.com/bxcodec/goqueue/internal/consumer/rabbitmq"
+	_ "github.com/bxcodec/goqueue/internal/shared" // Auto-setup logging
 	"github.com/bxcodec/goqueue/options"
 	consumerOpts "github.com/bxcodec/goqueue/options/consumer"
 )

consumer/service.go

@@ -20,11 +20,11 @@ type DecoderFn func(ctx context.Context, data []byte) (m interfaces.Message, err
 var (
 	// JSONEncoder is an implementation of the EncoderFn interface
 	// that encodes a Message into JSON format.
-	JSONEncoder EncoderFn = func(ctx context.Context, m interfaces.Message) (data []byte, err error) {
+	JSONEncoder EncoderFn = func(_ context.Context, m interfaces.Message) (data []byte, err error) {
 		return json.Marshal(m)
 	}
 	// JSONDecoder is a DecoderFn implementation that decodes JSON data into a Message.
-	JSONDecoder DecoderFn = func(ctx context.Context, data []byte) (m interfaces.Message, err error) {
+	JSONDecoder DecoderFn = func(_ context.Context, data []byte) (m interfaces.Message, err error) {
 		err = json.Unmarshal(data, &m)
 		return
 	}
@@ -73,6 +73,7 @@ var (
 	DefaultEncoding = JSONEncoding
 )
 
+//nolint:gochecknoinits // Required for auto-registration of default JSON encoding
 func init() {
 	AddGoQueueEncoding(JSONEncoding.ContentType, JSONEncoding)
 }

encoding.go

@@ -18,7 +18,7 @@ var (
 	}
 
 	// NoDelayFn is a DelayFn implementation that returns 0 delay for retries.
-	NoDelayFn DelayFn = func(currenRetries int64) (delay int64) {
+	NoDelayFn DelayFn = func(_ int64) (delay int64) {
 		return 0
 	}
 	// DefaultDelayFn is the default delay function that will be used if no delay function is provided.

interfaces/delayfn.go

@@ -22,7 +22,7 @@ type InboundMessage struct {
 	RetryCount int64 `json:"retryCount"`
 	// Metadata is the metadata of the message.
 	// This is set by the library to identify the metadata of the message.
-	Metadata map[string]interface{} `json:"metadata"`
+	Metadata map[string]any `json:"metadata"`
 	// Ack is used for confirming the message. It will drop the message from the queue.
 	Ack func(ctx context.Context) (err error) `json:"-"`
 	// Nack is used for rejecting the message. It will requeue the message to be re-delivered again.

interfaces/inboundmessagehandler.go

@@ -40,7 +40,7 @@ type Message struct {
 	// This is set by the publisher to identify the headers of the message.
 	// This library will provide extra headers values by default based on the library type.
 	// Don't use any prefix with :goqueue-, it will conflicted and overrided by the library.
-	Headers map[string]interface{} `json:"-"`
+	Headers map[string]any `json:"-"`
 	// ServiceAgent is the service agent that will be used to publish the message.
 	// This is set by the publisher to identify the service agent that will be used to publish the message.
 	// This will be set by the library and override any value

interfaces/message.go

@@ -13,7 +13,7 @@ type Consumer interface {
 	// Consume consumes messages from the queue and passes them to the provided handler.
 	// It takes a context, an InboundMessageHandler, and a map of metadata as parameters.
 	// It returns an error if there was a problem consuming the messages.
-	Consume(ctx context.Context, handler interfaces.InboundMessageHandler, meta map[string]interface{}) (err error)
+	Consume(ctx context.Context, handler interfaces.InboundMessageHandler, meta map[string]any) (err error)
 
 	// Stop stops the consumer from consuming messages.
 	// It takes a context as a parameter and returns an error if there was a problem stopping the consumer.

internal/consumer/consumer.go

@@ -7,6 +7,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/google/uuid"
+	amqp "github.com/rabbitmq/amqp091-go"
+	"github.com/rs/zerolog/log"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
 	"github.com/bxcodec/goqueue"
 	headerKey "github.com/bxcodec/goqueue/headers/key"
 	headerVal "github.com/bxcodec/goqueue/headers/value"
@@ -15,12 +22,6 @@ import (
 	"github.com/bxcodec/goqueue/middleware"
 	"github.com/bxcodec/goqueue/options"
 	consumerOpts "github.com/bxcodec/goqueue/options/consumer"
-	"github.com/google/uuid"
-	amqp "github.com/rabbitmq/amqp091-go"
-	"github.com/rs/zerolog/log"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
 )
 
 const (
@@ -58,7 +59,7 @@ func TestSuiteRabbitMQConsumer(t *testing.T) {
 	suite.Run(t, rabbitMQTestSuite)
 }
 
-func (s *rabbitMQTestSuite) BeforeTest(_, _ string) {
+func (*rabbitMQTestSuite) BeforeTest(_, _ string) {
 }
 
 func (s *rabbitMQTestSuite) AfterTest(_, _ string) {
@@ -127,11 +128,11 @@ func (s *rabbitMQTestSuite) initQueueForTesting(t *testing.T, exchangePattern ..
 	}
 }
 
-func (s *rabbitMQTestSuite) getMockData(action string) (res interfaces.Message) {
-	res = interfaces.Message{
+func (*rabbitMQTestSuite) getMockData(action string) (res *interfaces.Message) {
+	res = &interfaces.Message{
 		Action: action,
 		Topic:  testExchange,
-		Data: map[string]interface{}{
+		Data: map[string]any{
 			"message": "hello-world-test",
 		},
 		Timestamp: time.Date(2021, 1, 1, 0, 0, 0, 0, time.UTC),
@@ -140,7 +141,7 @@ func (s *rabbitMQTestSuite) getMockData(action string) (res interfaces.Message)
 	return res
 }
 
-func (s *rabbitMQTestSuite) seedPublish(contentType string, action string) {
+func (s *rabbitMQTestSuite) seedPublish(contentType, action string) {
 	mockData := s.getMockData(action)
 	jsonData, err := json.Marshal(mockData)
 	s.Require().NoError(err)
@@ -218,16 +219,16 @@ func (s *rabbitMQTestSuite) TestConsumerWithExchangePatternProvided() {
 	s.Require().NoError(err)
 }
 
-func handler(t *testing.T, expected interfaces.Message) interfaces.InboundMessageHandlerFunc {
+func handler(t *testing.T, expected *interfaces.Message) interfaces.InboundMessageHandlerFunc {
 	return func(ctx context.Context, m interfaces.InboundMessage) (err error) {
 		switch m.ContentType {
 		case headerVal.ContentTypeText:
 			assert.Equal(t, expected.Data, m.Data)
 		case headerVal.ContentTypeJSON:
-			expectedJSON, err := json.Marshal(expected.Data)
-			require.NoError(t, err)
-			actualJSON, err := json.Marshal(m.Data)
-			require.NoError(t, err)
+			expectedJSON, marshalErr := json.Marshal(expected.Data)
+			require.NoError(t, marshalErr)
+			actualJSON, marshalErr := json.Marshal(m.Data)
+			require.NoError(t, marshalErr)
 			assert.JSONEq(t, string(expectedJSON), string(actualJSON))
 		}
 
@@ -271,7 +272,7 @@ func (s *rabbitMQTestSuite) TestRequeueWithouthExchangePatternProvided() {
 func handlerRequeue(t *testing.T) interfaces.InboundMessageHandlerFunc {
 	return func(ctx context.Context, m interfaces.InboundMessage) (err error) {
 		delayFn := func(retries int64) int64 {
-			assert.Equal(t, int64(m.RetryCount)+1, retries) // because the retry++ is done before this delayfn is called
+			assert.Equal(t, m.RetryCount+1, retries) // because the retry++ is done before this delayfn is called
 			return m.RetryCount
 		}
 

internal/consumer/rabbitmq/blackbox_consumer_test.go

@@ -3,14 +3,15 @@ package rabbitmq
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"time"
 
 	"github.com/google/uuid"
 	amqp "github.com/rabbitmq/amqp091-go"
 	"github.com/rs/zerolog/log"
 
-	"github.com/bxcodec/goqueue/errors"
+	goqueueErrors "github.com/bxcodec/goqueue/errors"
 	headerKey "github.com/bxcodec/goqueue/headers/key"
 	headerVal "github.com/bxcodec/goqueue/headers/value"
 	"github.com/bxcodec/goqueue/interfaces"
@@ -19,6 +20,11 @@ import (
 	consumerOpts "github.com/bxcodec/goqueue/options/consumer"
 )
 
+const (
+	// millisecondsMultiplier converts seconds to milliseconds for RabbitMQ expiration
+	millisecondsMultiplier = 10_000
+)
+
 // rabbitMQ is the subscriber handler for rabbitmq
 type rabbitMQ struct {
 	consumerChannel             *amqp.Channel
@@ -213,7 +219,8 @@ func (r *rabbitMQ) initRetryModule() {
 // It takes a context, an inbound message handler, and a map of metadata as input parameters.
 // The function continuously listens for messages from the queue and processes them until the context is canceled.
 // If the context is canceled, the function stops consuming messages and returns.
-// For each received message, the function builds an inbound message, extracts the retry count, and checks if the maximum retry count has been reached.
+// For each received message, the function builds an inbound message, extracts the retry count,
+// and checks if the maximum retry count has been reached.
 // If the maximum retry count has been reached, the message is moved to the dead letter queue.
 // Otherwise, the message is passed to the message handler for processing.
 // The message handler is responsible for handling the message and returning an error if any.
@@ -225,7 +232,7 @@ func (r *rabbitMQ) initRetryModule() {
 // The function returns an error if any occurred during message handling or if the context was canceled.
 func (r *rabbitMQ) Consume(ctx context.Context,
 	h interfaces.InboundMessageHandler,
-	meta map[string]interface{}) (err error) {
+	meta map[string]any) (err error) {
 	log.Info().
 		Str("queue_name", r.option.QueueName).
 		Interface("consumer_meta", meta).
@@ -238,19 +245,19 @@ func (r *rabbitMQ) Consume(ctx context.Context,
 				Str("queue_name", r.option.QueueName).
 				Interface("consumer_meta", meta).
 				Msg("stopping the worker")
-			return
+			return err
 		case receivedMsg, ok := <-r.msgReceiver:
 			if !ok {
 				// deliveries channel closed (e.g., due to Stop/Cancel or connection closure)
 				log.Info().
 					Str("queue_name", r.option.QueueName).
 					Interface("consumer_meta", meta).
 					Msg("message receiver closed, stopping the worker")
-				return
+				return err
 			}
 			msg, err := buildMessage(meta, receivedMsg)
 			if err != nil {
-				if err == errors.ErrInvalidMessageFormat {
+				if errors.Is(err, goqueueErrors.ErrInvalidMessageFormat) {
 					nackErr := receivedMsg.Nack(false, false) // nack with requeue false
 					if nackErr != nil {
 						log.Error().
@@ -283,7 +290,7 @@ func (r *rabbitMQ) Consume(ctx context.Context,
 			m := interfaces.InboundMessage{
 				Message:    msg,
 				RetryCount: retryCount,
-				Metadata: map[string]interface{}{
+				Metadata: map[string]any{
 					"app-id":           receivedMsg.AppId,
 					"consumer-tag":     receivedMsg.ConsumerTag,
 					"content-encoding": receivedMsg.ContentEncoding,
@@ -299,17 +306,17 @@ func (r *rabbitMQ) Consume(ctx context.Context,
 					"type":             receivedMsg.Type,
 					"user-id":          receivedMsg.UserId,
 				},
-				Ack: func(ctx context.Context) (err error) {
+				Ack: func(_ context.Context) (err error) {
 					err = receivedMsg.Ack(false)
 					return
 				},
-				Nack: func(ctx context.Context) (err error) {
+				Nack: func(_ context.Context) (err error) {
 					//  receivedMsg.Nack(false, true) => will redelivered again instantly (same with receivedMsg.reject)
 					//  receivedMsg.Nack(false, false) => will put the message to dead letter queue (same with receivedMsg.reject)
 					err = receivedMsg.Nack(false, true)
 					return
 				},
-				MoveToDeadLetterQueue: func(ctx context.Context) (err error) {
+				MoveToDeadLetterQueue: func(_ context.Context) (err error) {
 					//  receivedMsg.Nack(false, true) => will redelivered again instantly (same with receivedMsg.reject)
 					//  receivedMsg.Nack(false, false) => will put the message to dead letter queue (same with receivedMsg.reject)
 					err = receivedMsg.Nack(false, false)
@@ -342,13 +349,13 @@ func (r *rabbitMQ) Consume(ctx context.Context,
 	}
 }
 
-func buildMessage(consumerMeta map[string]interface{}, receivedMsg amqp.Delivery) (msg interfaces.Message, err error) {
+func buildMessage(consumerMeta map[string]any, receivedMsg amqp.Delivery) (msg interfaces.Message, err error) {
 	if len(receivedMsg.Body) == 0 {
 		log.Error().
 			Interface("consumer_meta", consumerMeta).
 			Str("msg", string(receivedMsg.Body)).
 			Msg("message body is empty, removing the message due to wrong message format")
-		return msg, errors.ErrInvalidMessageFormat
+		return msg, goqueueErrors.ErrInvalidMessageFormat
 	}
 
 	err = json.Unmarshal(receivedMsg.Body, &msg)
@@ -358,7 +365,7 @@ func buildMessage(consumerMeta map[string]interface{}, receivedMsg amqp.Delivery
 			Str("msg", string(receivedMsg.Body)).
 			Err(err).
 			Msg("failed to unmarshal the message, removing the message due to wrong message format")
-		return msg, errors.ErrInvalidMessageFormat
+		return msg, goqueueErrors.ErrInvalidMessageFormat
 	}
 
 	if msg.ID == "" {
@@ -384,14 +391,14 @@ func buildMessage(consumerMeta map[string]interface{}, receivedMsg amqp.Delivery
 			Interface("consumer_meta", consumerMeta).
 			Interface("msg", msg).
 			Msg("message data is empty, removing the message due to wrong message format")
-		return msg, errors.ErrInvalidMessageFormat
+		return msg, goqueueErrors.ErrInvalidMessageFormat
 	}
 
 	msg.SetSchemaVersion(extractHeaderString(receivedMsg.Headers, headerKey.SchemaVer))
 	return msg, nil
 }
 
-func (r *rabbitMQ) requeueMessageWithDLQ(consumerMeta map[string]interface{}, msg interfaces.Message,
+func (r *rabbitMQ) requeueMessageWithDLQ(consumerMeta map[string]any, msg interfaces.Message,
 	receivedMsg amqp.Delivery) func(ctx context.Context, delayFn interfaces.DelayFn) (err error) {
 	return func(ctx context.Context, delayFn interfaces.DelayFn) (err error) {
 		if delayFn == nil {
@@ -417,7 +424,7 @@ func (r *rabbitMQ) requeueMessageWithDLQ(consumerMeta map[string]interface{}, ms
 				Body:        receivedMsg.Body,
 				Timestamp:   time.Now(),
 				AppId:       r.tagName,
-				Expiration:  fmt.Sprintf("%d", delayInSeconds*10000),
+				Expiration:  fmt.Sprintf("%d", delayInSeconds*millisecondsMultiplier),
 			},
 		)