diff

Documentation

Index

• Constants
• Variables
• func Diff(referenceValues, currentValues <-chan KeyValue, changes chan Change, ...) error
• func DiffIndexIndex(referenceIndex Index, currentIndex Index, changes chan Change, ...) error
• func DiffIndexStream(referenceIndex Index, currentValues <-chan KeyValue, changes chan Change, ...) error
• func DiffStreamIndex(referenceValues <-chan KeyValue, currentIndex Index, changes chan Change, ...) error
• func DiffStreamReference(referenceValues, currentValues <-chan KeyValue, changes chan Change, ...) error
• type Change
  • func (c Change) String() string
• type ChangeType
• type CompareResult
• type Index
  • func NewIndex(recordValues bool) Index
• type Indexer
• type KeyValue
• type MemoryIndex
  • func (i *MemoryIndex) Cleanup() (err error)
  • func (i *MemoryIndex) Compare(kv KeyValue) (CompareResult, error)
  • func (i *MemoryIndex) DoesRecordValues() bool
  • func (i *MemoryIndex) Index(kvs <-chan KeyValue, resumeKey <-chan []byte) (err error)
  • func (i *MemoryIndex) KeyValues() <-chan KeyValue
  • func (i *MemoryIndex) KeysNotSeen() <-chan []byte
  • func (i *MemoryIndex) ResumeKey() ([]byte, error)
  • func (i *MemoryIndex) Value(key []byte) []byte
• type SyncIndex

Constants

const (
	Unchanged = iota
	Created
	Modified
	Deleted
)

Variables

var (
	Debug = false

	ErrMustSupportKeysNotSeen = errors.New("referenceIndex must support KeysNotSeen")
	ErrMustRecordValues       = errors.New("referenceIndex must record values")
)

Functions

func Diff

func Diff(referenceValues, currentValues <-chan KeyValue, changes chan Change, cancel <-chan bool) error

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 faster and less memory consumming alternatives if you can provide better garanties from your stores.

func DiffIndexIndex

func DiffIndexIndex(referenceIndex Index, currentIndex Index, changes chan Change, cancel <-chan bool) error

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 DiffIndexStream

func DiffIndexStream(referenceIndex Index, currentValues <-chan KeyValue, changes chan Change, cancel <-chan bool) error

Compares a store (currentValues) with a reference (referenceIndex), streaming the reference.

The referenceIndex is the indexed target store. It MUST store the values AND support KeysNotSeen.

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 DiffStreamIndex

func DiffStreamIndex(referenceValues <-chan KeyValue, currentIndex Index, changes chan Change, cancel <-chan bool) error

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 DiffStreamReference

func DiffStreamReference(referenceValues, currentValues <-chan KeyValue, changes chan Change, cancel <-chan bool) error

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.

Types

type Change

type Change struct {
	Type ChangeType
	Key  []byte
	// The new value (Created and Modified events only)
	Value []byte
}

func (Change) String

func (c Change) String() string

type ChangeType

type ChangeType int

type CompareResult

type CompareResult int

const (
	MissingKey CompareResult = iota
	ModifiedKey
	UnchangedKey
)

type Index

type Index interface {
	Indexer
	SyncIndex
}

func NewIndex

func NewIndex(recordValues bool) Index

type Indexer

type Indexer interface {
	// Index records new values and ONE new resume key value (if given).
	// If resumeKey is not nil, a value is expected.
	Index(kvs <-chan KeyValue, resumeKey <-chan []byte) error

	// ResumeKey returns the current resume key value
	ResumeKey() ([]byte, error)
}

Indexer represents indexing functions

type KeyValue

type KeyValue struct {
	Key   []byte
	Value []byte
}

type MemoryIndex

type MemoryIndex struct {
	// contains filtered or unexported fields
}

func (*MemoryIndex) Cleanup

func (i *MemoryIndex) Cleanup() (err error)

func (*MemoryIndex) Compare

func (i *MemoryIndex) Compare(kv KeyValue) (CompareResult, error)

func (*MemoryIndex) DoesRecordValues

func (i *MemoryIndex) DoesRecordValues() bool

func (*MemoryIndex) Index

func (i *MemoryIndex) Index(kvs <-chan KeyValue, resumeKey <-chan []byte) (err error)

func (*MemoryIndex) KeyValues

func (i *MemoryIndex) KeyValues() <-chan KeyValue

func (*MemoryIndex) KeysNotSeen

func (i *MemoryIndex) KeysNotSeen() <-chan []byte

func (*MemoryIndex) ResumeKey

func (i *MemoryIndex) ResumeKey() ([]byte, error)

func (*MemoryIndex) Value

func (i *MemoryIndex) Value(key []byte) []byte

type SyncIndex

type SyncIndex interface {
	Cleanup() error

	Compare(kv KeyValue) (CompareResult, error)
	Value(key []byte) []byte

	KeyValues() <-chan KeyValue
	DoesRecordValues() bool

	// KeysNotSeen streams the keys that were not seen in the previous Compare calls.
	// Return nil if the index did not record this information.
	KeysNotSeen() <-chan []byte
}