initial commit

Author: Mikaël Cluseau

change.go

@@ -0,0 +1,39 @@
+package diff
+
+import (
+	"fmt"
+)
+
+type ChangeType int
+
+const (
+	Unchanged = iota
+	Created
+	Modified
+	Deleted
+)
+
+type Change struct {
+	Type ChangeType
+	Key  []byte
+	// The new value (Created and Modified events only)
+	Value []byte
+}
+
+func (c Change) String() string {
+	var s string
+	switch c.Type {
+	case Unchanged:
+		s = "U"
+	case Created:
+		s = "C"
+	case Modified:
+		s = "M"
+	case Deleted:
+		s = "D"
+	default:
+		panic(fmt.Errorf("Invalid change type: %d", c.Type))
+	}
+
+	return fmt.Sprintf("%s %q %q", s, c.Key, c.Value)
+}

diff.go

@@ -0,0 +1,158 @@
+package diff
+
+import (
+	"sync"
+)
+
+// Compares a store (currentValues) with a reference (referenceValues), streaming the reference.
+//
+// The referenceValues channel provide values in the reference store. It will be indexed.
+//
+// The currentValues channel provide values in the target store. It will be indexed.
+//
+// The changes channel will receive the changes, including Unchanged.
+//
+// See other diff implementations for less faster and less memory consumming alternatives if
+// you can provide better garanties from your stores.
+func Diff(referenceValues, currentValues <-chan KeyValue, changes chan Change) {
+	referenceIndex := NewIndex(true)
+	currentIndex := NewIndex(false)
+
+	wg := sync.WaitGroup{}
+	wg.Add(2)
+
+	go func() {
+		for kv := range referenceValues {
+			referenceIndex.Index(kv)
+		}
+		wg.Done()
+	}()
+
+	go func() {
+		for kv := range currentValues {
+			currentIndex.Index(kv)
+		}
+		wg.Done()
+	}()
+
+	wg.Wait()
+
+	DiffIndexIndex(referenceIndex, currentIndex, changes)
+}
+
+// Compares a store (currentValues) with a reference (referenceValues), streaming the reference.
+//
+// The referenceValues channel provide values in the reference store. It MUST NOT produce duplicate keys.
+//
+// The currentValues channel provide values in the target store. It will be indexed.
+//
+// The changes channel will receive the changes, including Unchanged.
+func DiffStreamReference(referenceValues, currentValues <-chan KeyValue, changes chan Change) {
+	currentIndex := NewIndex(false)
+
+	for kv := range currentValues {
+		currentIndex.Index(kv)
+	}
+
+	DiffStreamIndex(referenceValues, currentIndex, changes)
+}
+
+// Compares a store (currentIndex) with a reference (referenceValues), streaming the reference.
+//
+// The referenceValues channel provide values in the reference store. It MUST NOT produce duplicate keys.
+//
+// The currentIndex is the indexed target store.
+//
+// The changes channel will receive the changes, including Unchanged.
+func DiffStreamIndex(referenceValues <-chan KeyValue, currentIndex *Index, changes chan Change) {
+	for kv := range referenceValues {
+		kv := kv
+
+		switch currentIndex.Compare(kv) {
+		case MissingKey:
+			changes <- Change{
+				Type:  Created,
+				Key:   kv.Key,
+				Value: kv.Value,
+			}
+
+		case ModifiedKey:
+			changes <- Change{
+				Type:  Modified,
+				Key:   kv.Key,
+				Value: kv.Value,
+			}
+
+		case UnchangedKey:
+			changes <- Change{
+				Type: Unchanged,
+				Key:  kv.Key,
+			}
+
+		}
+	}
+
+	for key := range currentIndex.KeysNotSeen() {
+		changes <- Change{
+			Type: Deleted,
+			Key:  key,
+		}
+	}
+}
+
+// Compares a store (currentValues) with a reference (referenceIndex), streaming the reference.
+//
+// The referenceIndex is the indexed target store. It MUST store the values.
+//
+// The currentValues channel provide values in the reference store. It MUST NOT produce duplicate keys.
+//
+// The changes channel will receive the changes, including Unchanged.
+func DiffIndexStream(referenceIndex *Index, currentValues <-chan KeyValue, changes chan Change) {
+	if !referenceIndex.recordValues {
+		panic("referenceIndex must record values")
+	}
+	for kv := range currentValues {
+		kv := kv
+
+		switch referenceIndex.Compare(kv) {
+		case MissingKey:
+			changes <- Change{
+				Type: Deleted,
+				Key:  kv.Key,
+			}
+
+		case ModifiedKey:
+			changes <- Change{
+				Type:  Modified,
+				Key:   kv.Key,
+				Value: kv.Value,
+			}
+
+		case UnchangedKey:
+			changes <- Change{
+				Type: Unchanged,
+				Key:  kv.Key,
+			}
+
+		}
+	}
+
+	for key := range referenceIndex.KeysNotSeen() {
+		changes <- Change{
+			Type:  Created,
+			Key:   key,
+			Value: referenceIndex.Value(key),
+		}
+	}
+}
+
+// Compares a store (currentValues) with a reference (referenceIndex), streaming the reference.
+//
+// The referenceIndex is the indexed target store. It MUST store the values.
+//
+// The currentIndex is the indexed target store.
+//
+// The changes channel will receive the changes, including Unchanged.
+func DiffIndexIndex(referenceIndex *Index, currentIndex *Index, changes chan Change) {
+	DiffStreamIndex(referenceIndex.KeyValues(), currentIndex, changes)
+}

diff_test.go

@@ -0,0 +1,144 @@
+package diff
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestDiffAllChangeTypes(t *testing.T) {
+	testChanges(t, []KeyValue{
+		kvS("k1", "v1"),
+		kvS("k2", "v2"),
+		kvS("k3", "v3.1"),
+	}, []KeyValue{
+		kvS("k1", "v1"),
+		kvS("k3", "v3"),
+		kvS("k4", "v4"),
+	}, []Change{
+		{Unchanged, []byte("k1"), nil},
+		{Created, []byte("k2"), []byte("v2")},
+		{Modified, []byte("k3"), []byte("v3.1")},
+		{Deleted, []byte("k4"), nil},
+	})
+}
+
+func TestDiffEmpty(t *testing.T) {
+	testChanges(t, []KeyValue{}, []KeyValue{}, []Change{})
+}
+
+func TestDiffCreate(t *testing.T) {
+	testChanges(t, []KeyValue{
+		kvS("k1", "v1"),
+	}, []KeyValue{}, []Change{
+		{Created, []byte("k1"), []byte("v1")},
+	})
+}
+
+func TestDiffSame(t *testing.T) {
+	testChanges(t, []KeyValue{
+		kvS("k1", "v1"),
+	}, []KeyValue{
+		kvS("k1", "v1"),
+	}, []Change{
+		{Unchanged, []byte("k1"), nil},
+	})
+}
+
+func TestDiffModify(t *testing.T) {
+	testChanges(t, []KeyValue{
+		kvS("k1", "v1.1"),
+	}, []KeyValue{
+		kvS("k1", "v1"),
+	}, []Change{
+		{Modified, []byte("k1"), []byte("v1.1")},
+	})
+}
+
+func TestDiffDelete(t *testing.T) {
+	testChanges(t, []KeyValue{}, []KeyValue{
+		kvS("k1", "v1.1"),
+	}, []Change{
+		{Deleted, []byte("k1"), nil},
+	})
+}
+
+func TestDiffIndexStream(t *testing.T) {
+	changes := make(chan Change, 10)
+
+	refIndex := NewIndex(true)
+	refIndex.Index(kvS("k2", "v2"))
+
+	go func() {
+		DiffIndexStream(refIndex, stream([]KeyValue{
+			kvS("k1", "v1"),
+		}), changes)
+		close(changes)
+	}()
+
+	expectChanges(t, changes, []Change{
+		{Deleted, []byte("k1"), nil},
+		{Created, []byte("k2"), []byte("v2")},
+	})
+}
+
+func stream(kvs []KeyValue) <-chan KeyValue {
+	c := make(chan KeyValue, 1)
+	go func() {
+		for _, kv := range kvs {
+			c <- kv
+		}
+		close(c)
+	}()
+	return c
+}
+
+func kvS(key, value string) KeyValue {
+	return KeyValue{[]byte(key), []byte(value)}
+}
+
+func testChanges(t *testing.T, ref, current []KeyValue, exp []Change) {
+	t.Helper()
+
+	changes := make(chan Change, 10)
+
+	go func() {
+		Diff(stream(ref), stream(current), changes)
+		close(changes)
+	}()
+
+	expectChanges(t, changes, exp)
+}
+
+func expectChanges(t *testing.T, changes <-chan Change, exp []Change) {
+	t.Helper()
+
+	for change := range changes {
+		found := false
+		for idx, expChange := range exp {
+			if eq(change, expChange) {
+				exp = append(exp[0:idx], exp[idx+1:]...)
+				found = true
+			}
+		}
+		if !found {
+			t.Error("unexpected change: ", change)
+		}
+	}
+
+	for _, noFound := range exp {
+		t.Error("expected change not found: ", noFound)
+	}
+}
+
+func eq(c1, c2 Change) bool {
+	if c1.Type != c2.Type {
+		return false
+	}
+	if !bytes.Equal(c1.Key, c2.Key) {
+		return false
+	}
+	if !bytes.Equal(c1.Value, c2.Value) {
+		return false
+	}
+	return true
+}