more test and test replicator parallel

replicator/replicator.go

@@ -2,21 +2,88 @@ package replicator
 
 import (
 	"context"
-	"io"
+	"time"
 )
 
 type Replicator struct {
-	Src  Driver
-	Dest Driver
+	Src      Driver
+	Dest     Driver
+	Interval time.Duration
 }
 
 func NewReplicator(src, dest Driver) Replicator {
 	return Replicator{
-		Src:  src,
-		Dest: dest,
+		Src:      src,
+		Dest:     dest,
+		Interval: time.Second,
 	}
 }
 
 func (r Replicator) Stream(ctx context.Context) error {
-	return io.EOF
+	last, err := r.getContinuation(ctx)
+	if err != nil {
+		return err
+	}
+	if last == nil {
+		if err := r.fullStream(ctx); err != nil {
+			return err
+		}
+	}
+	return r.deltaStream(ctx)
+}
+
+func (r Replicator) getContinuation(ctx context.Context) (Version, error) {
+	return nil, nil // TODO
+}
+
+func (r Replicator) setContinuation(ctx context.Context, continuation Version) error {
+	return nil // TODO
+}
+
+func (r Replicator) deltaStream(ctx context.Context) error {
+	continuation, err := r.getContinuation(ctx)
+	if err != nil {
+		return err
+	}
+	for ctx.Err() == nil {
+		if err := r._deltaStream(ctx, continuation); err != nil {
+			return err
+		}
+		select {
+		case <-ctx.Done():
+		case <-time.After(r.Interval):
+		}
+	}
+	return ctx.Err()
+}
+
+func (r Replicator) _deltaStream(ctx context.Context, continuation Version) error {
+	last, err := continuation.AsTime()
+	if err != nil {
+		return err
+	}
+	keys, after := r.Src.KeysSince(ctx, last)
+	for keyVersion := range keys {
+		if t, err := keyVersion.Version.AsTime(); err != nil {
+			return err
+		} else if t.After(last) {
+			last = t
+		}
+
+		valueVersion, err := r.Src.Get(ctx, keyVersion.Key)
+		if err != nil {
+			return err
+		}
+
+		if err := r.Dest.Set(ctx, keyVersion.Key, valueVersion.Value, valueVersion.Version); err != nil {
+			return err
+		}
+
+		r.setContinuation(ctx, TimeAsVersion(last))
+	}
+	return *after
+}
+
+func (r Replicator) fullStream(ctx context.Context) error {
+	return r._deltaStream(ctx, nil)
 }

replicator/replicator_test.go

@@ -9,8 +9,11 @@ import (
 )
 
 func TestReplicatorStream(t *testing.T) {
+	t.Parallel()
+
 	key := Key{Namespace: "x/y", Key: "z"}
 	value := Value("hello world")
+	oldVersion := TimeAsVersion(time.Now().Add(-1 * time.Hour))
 	version := TimeAsVersion(time.Now())
 
 	cases := map[string]struct {
@@ -32,7 +35,19 @@ func TestReplicatorStream(t *testing.T) {
 		"one during op moves": {
 			during: func(t *testing.T, r Replicator) {
 				r.Src.Set(nil, key, value, version)
-				time.Sleep(time.Millisecond * 200)
+			},
+			after: func(t *testing.T, r Replicator) {
+				if got, _ := r.Dest.Get(nil, key); !got.Version.Equal(version) || !got.Value.Equal(value) {
+					t.Error(got)
+				}
+			},
+		},
+		"dont overwrite newer received later": {
+			before: func(t *testing.T, r Replicator) {
+				r.Dest.Set(nil, key, value, version)
+			},
+			during: func(t *testing.T, r Replicator) {
+				r.Src.Set(nil, key, value, oldVersion)
 			},
 			after: func(t *testing.T, r Replicator) {
 				if got, _ := r.Dest.Get(nil, key); !got.Version.Equal(version) || !got.Value.Equal(value) {
@@ -46,6 +61,7 @@ func TestReplicatorStream(t *testing.T) {
 		c := d
 		t.Run(name, func(t *testing.T) {
 			replicator := NewReplicator(NewMust(NewMap()), NewMust(NewMap()))
+			replicator.Interval = time.Millisecond * 10
 			ctx, can := context.WithTimeout(context.Background(), time.Second*10)
 			defer can()
 
@@ -55,15 +71,17 @@ func TestReplicatorStream(t *testing.T) {
 
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
-			go func() {
+			go func(t *testing.T) {
 				defer wg.Done()
 				if err := replicator.Stream(ctx); err != nil && !errors.Is(err, ctx.Err()) {
-					t.Fatal(err)
+					t.Error(err)
 				}
-			}()
+			}(t)
+			time.Sleep(time.Millisecond * 150)
 			if c.during != nil {
 				c.during(t, replicator)
 			}
+			time.Sleep(time.Millisecond * 150)
 
 			can()
 			wg.Wait()