From b215a81c59722fa739a3bc69f6eaa5019007cc98 Mon Sep 17 00:00:00 2001
From: bel <bel@bel.bel>
Date: Sun, 5 Nov 2023 13:48:48 -0700
Subject: [PATCH] map to thread safe

---
 replicator/map.go | 25 ++++++++++++++++++++-----
 1 file changed, 20 insertions(+), 5 deletions(-)

diff --git a/replicator/map.go b/replicator/map.go
index 1654301..0ce4366 100644
--- a/replicator/map.go
+++ b/replicator/map.go
@@ -3,13 +3,20 @@ package replicator
 import (
 	"context"
 	"io"
+	"sync"
 	"time"
 )
 
-type Map map[Key]ValueVersion
+type Map struct {
+	m    map[Key]ValueVersion
+	lock *sync.RWMutex
+}
 
 func NewMap() Map {
-	return make(Map)
+	return Map{
+		m:    make(map[Key]ValueVersion),
+		lock: &sync.RWMutex{},
+	}
 }
 
 func (m Map) KeysSince(ctx context.Context, t time.Time) (chan KeyVersion, *error) {
@@ -17,9 +24,11 @@ func (m Map) KeysSince(ctx context.Context, t time.Time) (chan KeyVersion, *erro
 	var err error
 	go func() {
 		defer close(result)
-		for k := range m {
+		m.lock.RLock()
+		defer m.lock.RUnlock()
+		for k := range m.m {
 			select {
-			case result <- KeyVersion{Key: k, Version: m[k].Version}:
+			case result <- KeyVersion{Key: k, Version: m.m[k].Version}:
 			case <-ctx.Done():
 				err = ctx.Err()
 				return
@@ -30,14 +39,20 @@ func (m Map) KeysSince(ctx context.Context, t time.Time) (chan KeyVersion, *erro
 }
 
 func (m Map) Get(_ context.Context, k Key) (ValueVersion, error) {
+	m.lock.RLock()
+	defer m.lock.RUnlock()
 	return ValueVersion{}, io.EOF
 }
 
 func (m Map) Set(_ context.Context, key Key, value Value, version Version) error {
-	m[key] = ValueVersion{Value: value, Version: version}
+	m.lock.Lock()
+	defer m.lock.Unlock()
+	m.m[key] = ValueVersion{Value: value, Version: version}
 	return nil
 }
 
 func (m Map) Del(_ context.Context, k Key, v Version) error {
+	m.lock.Lock()
+	defer m.lock.Unlock()
 	return io.EOF
 }