Schedules consumer index persistence based on the configuration

Author: ashishkujoy

cmd/server/main.go

@@ -8,7 +8,12 @@ import (
 )
 
 func main() {
-	conf := config.NewConfig("data/segments", "data/metadata", 1024*1024, time.Second*100)
+	conf := config.NewConfig(
+		"data/segments",
+		"data/metadata",
+		1024*1024,
+		time.Second*2,
+	)
 	server, err := netinternal.NewQueueServer(conf, ":50051")
 	if err != nil {
 		log.Fatalf("Failed to create server: %v", err)

internal/consumer/consumer_index.go

@@ -41,17 +41,18 @@ func NewConsumerIndex(config *config.Config) (*ConsumerIndex, error) {
 }
 
 func (ci *ConsumerIndex) schedulePersist() {
+	fmt.Printf("Starting consumer index persistence with interval: %s\n", ci.config.ConsumerIndexSyncInterval())
 	ticker := time.NewTicker(ci.config.ConsumerIndexSyncInterval())
 	go func() {
 		for {
 			select {
 			case <-ticker.C:
-				ci.mu.Lock()
+				fmt.Printf("Starting consumer index persistence\n")
 				err := ci.Persist()
-				ci.mu.Unlock()
 				if err != nil {
 					fmt.Printf("Error persisting consumer index: %v\n", err)
 				}
+				fmt.Printf("Done with consumer index persistence\n")
 			}
 		}
 	}()
@@ -141,12 +142,14 @@ func RestoreConsumerIndex(config *config.Config) (*ConsumerIndex, error) {
 		return nil, err
 	}
 
-	return &ConsumerIndex{
+	c := &ConsumerIndex{
 		writer:  lastIndexFile,
 		indexes: indexes,
 		mu:      &sync.RWMutex{},
 		config:  config,
-	}, nil
+	}
+	go c.schedulePersist()
+	return c, nil
 }
 
 // createIndexFile creates a new index file for the consumer.

internal/consumer/consumer_index_test.go

@@ -23,8 +23,8 @@ func TestReadAndWriteIndex(t *testing.T) {
 	assert.NoError(t, err)
 	defer os.RemoveAll(metadataDir)
 
-	config := config.NewConfig("/tmp", metadataDir, 1234, time.Second*100)
-	index, err := NewConsumerIndex(config)
+	cfg := config.NewConfig("/tmp", metadataDir, 1234, time.Second*100)
+	index, err := NewConsumerIndex(cfg)
 	assert.NoError(t, err)
 
 	index.WriteIndex(1, 10)
@@ -66,26 +66,26 @@ func TestRestoreIndexUsesTheLatestSnapshot(t *testing.T) {
 	assert.NoError(t, err)
 	defer os.RemoveAll(metadataDir)
 
-	config := config.NewConfig("/tmp", metadataDir, 1234, time.Second*100)
+	cfg := config.NewConfig("/tmp", metadataDir, 1234, time.Second*100)
 
-	index1, err := NewConsumerIndex(config)
+	index1, err := NewConsumerIndex(cfg)
 	assert.NoError(t, err)
 
 	index1.WriteIndex(11, 10)
 	index1.WriteIndex(12, 20)
 	index1.WriteIndex(13, 30)
 
-	index1.Close()
+	assert.NoError(t, index1.Close())
 
-	index2, err := RestoreConsumerIndex(config)
+	index2, err := RestoreConsumerIndex(cfg)
 	assert.NoError(t, err)
 
 	index2.WriteIndex(12, 200)
 	index2.WriteIndex(14, 1300)
 
-	index2.Close()
+	assert.NoError(t, index2.Close())
 
-	restoredIndex, err := RestoreConsumerIndex(config)
+	restoredIndex, err := RestoreConsumerIndex(cfg)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 10, restoredIndex.ReadIndex(11))