Adds a test to verify queue behaviour after restore

Author: ashishkujoy

internal/queue/queue_service.go

@@ -45,3 +45,11 @@ func (qs *QueueService) RevertDequeue(consumerId int) {
 	index := qs.consumerIndex.ReadIndex(consumerId)
 	qs.consumerIndex.WriteIndex(consumerId, index-1)
 }
+
+func (qs *QueueService) Close() error {
+	err := qs.queue.Close()
+	if err != nil {
+		return err
+	}
+	return qs.consumerIndex.Close()
+}

internal/queue/queue_service_test.go

@@ -24,19 +24,53 @@ func TestEnqueue(t *testing.T) {
 	assert.NoError(t, queueService.Enqueue([]byte("Hello World 2")))
 	assert.NoError(t, queueService.Enqueue([]byte("Hello World 3")))
 
-	_, err = queueService.Dequeue(1)
+	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)
+}
+
+func TestDequeueFromARestoredQueue(t *testing.T) {
+	segmentPath := createTempDir("testEnqueue/segments")
+	metaDataPath := createTempDir("testEnqueue/metadata")
+	defer os.RemoveAll(segmentPath)
+	defer os.RemoveAll(metaDataPath)
+	cfg := config.NewConfig(segmentPath, metaDataPath, 1024, time.Second)
+
+	queueService, err := NewQueueService(cfg)
 	assert.NoError(t, err)
-	//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)
+
+	assert.NoError(t, queueService.Enqueue([]byte("Hello World")))
+	assert.NoError(t, queueService.Enqueue([]byte("Hello World 1")))
+	assert.NoError(t, queueService.Enqueue([]byte("Hello World 2")))
+	assert.NoError(t, 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)
+
+	assert.NoError(t, queueService.Close())
+	queueService, err = NewQueueService(cfg)
+	assert.NoError(t, err)
+
+	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)
 }