map to thread safe

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
 }