refactor: rename messagepool package to queue (#28)

CONTRIBUTING.md

@@ -24,7 +24,7 @@ deckard                     # Root folder with project files and Golang service/
 │   ├── cmd                 # Executable main file for the Deckard service
 │   ├── config              # Configuration variables managed by viper
 │   ├── logger              # Logging configuration
-│   ├── messagepool         # Contains all main implementation files and housekeeping program logic
+│   ├── queue               # Contains all queue implementation files and housekeeping program logic
 │   │   ├── cache           # Caching implementation
 │   │   ├── entities        # Message and QueueConfiguration internal definitions
 │   │   ├── queue           # Queue definition

internal/audit/audit.go

@@ -15,10 +15,10 @@ import (
 
 	"github.com/takenet/deckard/internal/config"
 	"github.com/takenet/deckard/internal/logger"
-	"github.com/takenet/deckard/internal/messagepool/entities"
-	"github.com/takenet/deckard/internal/messagepool/utils"
 	"github.com/takenet/deckard/internal/metrics"
 	"github.com/takenet/deckard/internal/project"
+	"github.com/takenet/deckard/internal/queue/entities"
+	"github.com/takenet/deckard/internal/queue/utils"
 	"go.uber.org/zap"
 
 	"github.com/elastic/go-elasticsearch/v7"

internal/cmd/deckard/main.go

@@ -11,12 +11,11 @@ import (
 	"github.com/takenet/deckard/internal/audit"
 	"github.com/takenet/deckard/internal/config"
 	"github.com/takenet/deckard/internal/logger"
-	"github.com/takenet/deckard/internal/messagepool"
-	"github.com/takenet/deckard/internal/messagepool/cache"
-	"github.com/takenet/deckard/internal/messagepool/queue"
-	"github.com/takenet/deckard/internal/messagepool/storage"
-	"github.com/takenet/deckard/internal/messagepool/utils"
 	"github.com/takenet/deckard/internal/metrics"
+	"github.com/takenet/deckard/internal/queue"
+	"github.com/takenet/deckard/internal/queue/cache"
+	"github.com/takenet/deckard/internal/queue/storage"
+	"github.com/takenet/deckard/internal/queue/utils"
 	"github.com/takenet/deckard/internal/service"
 	"github.com/takenet/deckard/internal/shutdown"
 	"github.com/takenet/deckard/internal/trace"
@@ -88,16 +87,16 @@ func main() {
 	}
 	go auditor.StartSender(ctx)
 
-	queueService := queue.NewConfigurationService(ctx, dataStorage)
+	configurationService := queue.NewQueueConfigurationService(ctx, dataStorage)
 
-	messagePool := messagepool.NewMessagePool(auditor, dataStorage, queueService, dataCache)
+	queue := queue.NewQueue(auditor, dataStorage, configurationService, dataCache)
 
 	if config.GrpcEnabled.GetBool() {
-		server = startGrpcServer(messagePool, queueService)
+		server = startGrpcServer(queue, configurationService)
 	}
 
 	if config.HousekeeperEnabled.GetBool() {
-		startHouseKeeperJobs(messagePool)
+		startHouseKeeperJobs(queue)
 	}
 
 	// Handle sigterm and await termChan signal
@@ -113,8 +112,8 @@ func isMemoryInstance() bool {
 	return config.CacheType.Get() == string(cache.MEMORY) && config.StorageType.Get() == string(storage.MEMORY)
 }
 
-func startGrpcServer(messagepool *messagepool.MessagePool, queueService queue.ConfigurationService) *grpc.Server {
-	deckard := service.NewDeckardInstance(messagepool, queueService, isMemoryInstance())
+func startGrpcServer(queue *queue.Queue, queueService queue.QueueConfigurationService) *grpc.Server {
+	deckard := service.NewDeckardInstance(queue, queueService, isMemoryInstance())
 
 	server, err := deckard.ServeGRPCServer(ctx)
 	if err != nil {
@@ -125,14 +124,14 @@ func startGrpcServer(messagepool *messagepool.MessagePool, queueService queue.Co
 	return server
 }
 
-func startHouseKeeperJobs(pool *messagepool.MessagePool) {
+func startHouseKeeperJobs(pool *queue.Queue) {
 	go scheduleTask(
 		UNLOCK,
 		nil,
 		shutdown.WaitGroup,
 		config.HousekeeperTaskUnlockDelay.GetDuration(),
 		func() bool {
-			messagepool.ProcessLockPool(ctx, pool)
+			queue.ProcessLockPool(ctx, pool)
 
 			return true
 		},
@@ -144,7 +143,7 @@ func startHouseKeeperJobs(pool *messagepool.MessagePool) {
 		shutdown.WaitGroup,
 		config.HousekeeperTaskTimeoutDelay.GetDuration(),
 		func() bool {
-			_ = messagepool.ProcessTimeoutMessages(ctx, pool)
+			_ = queue.ProcessTimeoutMessages(ctx, pool)
 
 			return true
 		},
@@ -156,7 +155,7 @@ func startHouseKeeperJobs(pool *messagepool.MessagePool) {
 		shutdown.WaitGroup,
 		config.HousekeeperTaskMetricsDelay.GetDuration(),
 		func() bool {
-			messagepool.ComputeMetrics(ctx, pool)
+			queue.ComputeMetrics(ctx, pool)
 
 			return true
 		},
@@ -168,7 +167,7 @@ func startHouseKeeperJobs(pool *messagepool.MessagePool) {
 		shutdown.CriticalWaitGroup,
 		config.HousekeeperTaskUpdateDelay.GetDuration(),
 		func() bool {
-			return messagepool.RecoveryMessagesPool(ctx, pool)
+			return queue.RecoveryMessagesPool(ctx, pool)
 		},
 	)
 
@@ -178,7 +177,7 @@ func startHouseKeeperJobs(pool *messagepool.MessagePool) {
 		shutdown.WaitGroup,
 		config.HousekeeperTaskMaxElementsDelay.GetDuration(),
 		func() bool {
-			metrify, _ := messagepool.RemoveExceedingMessages(ctx, pool)
+			metrify, _ := queue.RemoveExceedingMessages(ctx, pool)
 
 			return metrify
 		},
@@ -192,7 +191,7 @@ func startHouseKeeperJobs(pool *messagepool.MessagePool) {
 		func() bool {
 			now := time.Now()
 
-			metrify, _ := messagepool.RemoveTTLMessages(ctx, pool, &now)
+			metrify, _ := queue.RemoveTTLMessages(ctx, pool, &now)
 
 			return metrify
 		},

internal/metrics/metrics.go

@@ -13,8 +13,8 @@ import (
 	dto "github.com/prometheus/client_model/go"
 	"github.com/takenet/deckard/internal/config"
 	"github.com/takenet/deckard/internal/logger"
-	"github.com/takenet/deckard/internal/messagepool/utils"
 	"github.com/takenet/deckard/internal/project"
+	"github.com/takenet/deckard/internal/queue/utils"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/exporters/prometheus"
 	otelmetric "go.opentelemetry.io/otel/metric"
@@ -32,7 +32,7 @@ var (
 	mutex             = sync.Mutex{}
 
 	meter      otelmetric.Meter
-	MetricsMap *MessagePoolMetricsMap
+	MetricsMap *QueueMetricsMap
 
 	// Keep tracking of features used by clients to be able to deprecate them
 	CodeUsage instrument.Int64Counter
@@ -48,12 +48,12 @@ var (
 	HousekeeperTotalElements           instrument.Int64ObservableGauge
 
 	// Message Pool
-	MessagePoolTimeout           instrument.Int64Counter
-	MessagePoolAck               instrument.Int64Counter
-	MessagePoolNack              instrument.Int64Counter
-	MessagePoolEmptyQueue        instrument.Int64Counter
-	MessagePoolEmptyQueueStorage instrument.Int64Counter
-	MessagePoolNotFoundInStorage instrument.Int64Counter
+	QueueTimeout           instrument.Int64Counter
+	QueueAck               instrument.Int64Counter
+	QueueNack              instrument.Int64Counter
+	QueueEmptyQueue        instrument.Int64Counter
+	QueueEmptyQueueStorage instrument.Int64Counter
+	QueueNotFoundInStorage instrument.Int64Counter
 
 	// Storage
 	StorageLatency instrument.Int64Histogram
@@ -178,7 +178,7 @@ func createDefaultMetrics() []*dto.LabelPair {
 func createMetrics() {
 	meter = global.MeterProvider().Meter(project.Name)
 
-	MetricsMap = NewMessagePoolMetricsMap()
+	MetricsMap = NewQueueMetricsMap()
 
 	// Housekeeper
 
@@ -243,39 +243,39 @@ func createMetrics() {
 	)
 	panicInstrumentationError(err)
 
-	// MessagePool
+	// Queue
 
-	MessagePoolTimeout, err = meter.Int64Counter(
+	QueueTimeout, err = meter.Int64Counter(
 		"deckard_message_timeout",
 		instrument.WithDescription("Number of message timeouts"),
 	)
 	panicInstrumentationError(err)
 
-	MessagePoolAck, err = meter.Int64Counter(
+	QueueAck, err = meter.Int64Counter(
 		"deckard_ack",
 		instrument.WithDescription("Number of acks received"),
 	)
 	panicInstrumentationError(err)
 
-	MessagePoolNack, err = meter.Int64Counter(
+	QueueNack, err = meter.Int64Counter(
 		"deckard_nack",
 		instrument.WithDescription("Number of nacks received"),
 	)
 	panicInstrumentationError(err)
 
-	MessagePoolEmptyQueue, err = meter.Int64Counter(
+	QueueEmptyQueue, err = meter.Int64Counter(
 		"deckard_messages_empty_queue",
 		instrument.WithDescription("Number of times a pull is made against an empty queue."),
 	)
 	panicInstrumentationError(err)
 
-	MessagePoolEmptyQueueStorage, err = meter.Int64Counter(
+	QueueEmptyQueueStorage, err = meter.Int64Counter(
 		"deckard_messages_empty_queue_not_found_storage",
 		instrument.WithDescription("Number of times a pull is made against an empty queue because the messages were not found in the storage."),
 	)
 	panicInstrumentationError(err)
 
-	MessagePoolNotFoundInStorage, err = meter.Int64Counter(
+	QueueNotFoundInStorage, err = meter.Int64Counter(
 		"deckard_messages_not_found_in_storage",
 		instrument.WithDescription("Number of messages in cache but not found in the storage."),
 	)

internal/metrics/metrics_map.go

@@ -1,23 +1,23 @@
 package metrics
 
 // Type to hold map of oldest queue elements
-type MessagePoolMetricsMap struct {
+type QueueMetricsMap struct {
 	OldestElement map[string]int64
 	TotalElements map[string]int64
 }
 
-func NewMessagePoolMetricsMap() *MessagePoolMetricsMap {
-	return &MessagePoolMetricsMap{
+func NewQueueMetricsMap() *QueueMetricsMap {
+	return &QueueMetricsMap{
 		OldestElement: make(map[string]int64, 0),
 		TotalElements: make(map[string]int64, 0),
 	}
 }
 
-func (oldestMap *MessagePoolMetricsMap) UpdateOldestElementMap(data map[string]int64) {
+func (oldestMap *QueueMetricsMap) UpdateOldestElementMap(data map[string]int64) {
 	oldestMap.OldestElement = mergeData(oldestMap.OldestElement, data)
 }
 
-func (oldestMap *MessagePoolMetricsMap) UpdateTotalElementsMap(data map[string]int64) {
+func (oldestMap *QueueMetricsMap) UpdateTotalElementsMap(data map[string]int64) {
 	oldestMap.TotalElements = mergeData(oldestMap.TotalElements, data)
 }
 

internal/metrics/metrics_map_test.go

@@ -7,7 +7,7 @@ import (
 )
 
 func TestChangeMapShouldChangeSuccessfully(t *testing.T) {
-	data := NewMessagePoolMetricsMap()
+	data := NewQueueMetricsMap()
 
 	require.Empty(t, data.OldestElement)
 
@@ -20,7 +20,7 @@ func TestChangeMapShouldChangeSuccessfully(t *testing.T) {
 }
 
 func TestChangeMapWithNilMapShouldEmptyMap(t *testing.T) {
-	data := NewMessagePoolMetricsMap()
+	data := NewQueueMetricsMap()
 
 	data.OldestElement["a"] = int64(123)
 
@@ -32,7 +32,7 @@ func TestChangeMapWithNilMapShouldEmptyMap(t *testing.T) {
 }
 
 func TestChangeMapWithMissingQueueShouldKeepElementAsZero(t *testing.T) {
-	data := NewMessagePoolMetricsMap()
+	data := NewQueueMetricsMap()
 
 	data.OldestElement["a"] = int64(123)
 	data.OldestElement["b"] = int64(432)

internal/messagepool/cache/cache.go → internal/queue/cache/cache.go

@@ -7,7 +7,7 @@ import (
 	"errors"
 	"time"
 
-	"github.com/takenet/deckard/internal/messagepool/entities"
+	"github.com/takenet/deckard/internal/queue/entities"
 )
 
 type Type string

internal/messagepool/cache/cache_suite_test.go → internal/queue/cache/cache_suite_test.go

@@ -8,8 +8,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
-	"github.com/takenet/deckard/internal/messagepool/entities"
-	"github.com/takenet/deckard/internal/messagepool/utils"
+	"github.com/takenet/deckard/internal/queue/entities"
+	"github.com/takenet/deckard/internal/queue/utils"
 )
 
 var ctx, cancel = context.WithCancel(context.Background())

internal/messagepool/cache/memory_cache.go → internal/queue/cache/memory_cache.go

@@ -7,8 +7,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/takenet/deckard/internal/messagepool/entities"
-	"github.com/takenet/deckard/internal/messagepool/utils"
+	"github.com/takenet/deckard/internal/queue/entities"
+	"github.com/takenet/deckard/internal/queue/utils"
 )
 
 // MemoryStorage is an implementation of the Storage Interface using memory.

internal/messagepool/cache/redis_cache.go → internal/queue/cache/redis_cache.go

@@ -13,9 +13,9 @@ import (
 	"github.com/meirf/gopart"
 	"github.com/takenet/deckard/internal/config"
 	"github.com/takenet/deckard/internal/logger"
-	"github.com/takenet/deckard/internal/messagepool/entities"
-	"github.com/takenet/deckard/internal/messagepool/utils"
 	"github.com/takenet/deckard/internal/metrics"
+	"github.com/takenet/deckard/internal/queue/entities"
+	"github.com/takenet/deckard/internal/queue/utils"
 	"go.opentelemetry.io/otel/attribute"
 )
 

internal/messagepool/cache/redis_cache_test.go → internal/queue/cache/redis_cache_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"github.com/takenet/deckard/internal/config"
-	"github.com/takenet/deckard/internal/messagepool/entities"
+	"github.com/takenet/deckard/internal/queue/entities"
 )
 
 func TestRedisCacheIntegration(t *testing.T) {

internal/messagepool/entities/message.go → internal/queue/entities/message.go

@@ -4,7 +4,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/takenet/deckard/internal/messagepool/utils"
+	"github.com/takenet/deckard/internal/queue/utils"
 	"google.golang.org/protobuf/types/known/anypb"
 )
 

internal/messagepool/entities/message_test.go → internal/queue/entities/message_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	"github.com/takenet/deckard/internal/messagepool/utils"
+	"github.com/takenet/deckard/internal/queue/utils"
 )
 
 func TestMaxScore(t *testing.T) {