Adds queue service

Author: ashishkujoy

internal/queue/queueServerice_test.go

@@ -0,0 +1,40 @@
+package queueinternal
+
+import (
+	"ashishkujoy/queue/internal/config"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEnqueue(t *testing.T) {
+	segmentPath := createTempDir("testEnqueue/segments")
+	defer os.RemoveAll(segmentPath)
+	metaDataPath := createTempDir("testEnqueue/metadata")
+	defer os.RemoveAll(metaDataPath)
+	config := config.NewConfigWithMetadataPath(segmentPath, metaDataPath, 1024)
+
+	queueService, err := NewQueueService(config)
+	assert.NoError(t, err)
+
+	queueService.Enqueue([]byte("Hello World"))
+	queueService.Enqueue([]byte("Hello World 1"))
+	queueService.Enqueue([]byte("Hello World 2"))
+	queueService.Enqueue([]byte("Hello World 3"))
+
+	data, _ := queueService.Dequeue(1)
+	assert.Equal(t, []byte("Hello World"), data)
+
+	data, _ = queueService.Dequeue(1)
+	assert.Equal(t, []byte("Hello World 1"), data)
+
+	data, _ = queueService.Dequeue(1)
+	assert.Equal(t, []byte("Hello World 2"), data)
+
+	data, _ = queueService.Dequeue(2)
+	assert.Equal(t, []byte("Hello World"), data)
+
+	data, _ = queueService.Dequeue(1)
+	assert.Equal(t, []byte("Hello World 3"), data)
+}

internal/queue/queueService.go

@@ -0,0 +1,42 @@
+package queueinternal
+
+import (
+	"ashishkujoy/queue/internal/config"
+	"ashishkujoy/queue/internal/consumer"
+)
+
+type QueueService struct {
+	queue         *Queue
+	consumerIndex *consumer.ConsumerIndex
+}
+
+func NewQueueService(config *config.Config) (*QueueService, error) {
+	queue, err := NewQueue(config)
+	if err != nil {
+		return nil, err
+	}
+	consumerIndex, err := consumer.NewConsumerIndex(config)
+	if err != nil {
+		return nil, err
+	}
+
+	return &QueueService{
+		queue:         queue,
+		consumerIndex: consumerIndex,
+	}, nil
+}
+
+func (qs *QueueService) Enqueue(data []byte) error {
+	_, err := qs.queue.Enqueue(data)
+	return err
+}
+
+func (qs *QueueService) Dequeue(consumerId int) ([]byte, error) {
+	index := qs.consumerIndex.ReadIndex(consumerId)
+	data, err := qs.queue.Dequeue(index + 1)
+	if err != nil {
+		return nil, err
+	}
+	qs.consumerIndex.WriteIndex(consumerId, index+1)
+	return data, nil
+}