reader

Documentation

Index

• Constants
• Variables
• func GetVersion(db int) (uint64, bool)
• func KVVersionGenerator(so ServerOptions, inc <-chan []RedisKV, outc chan<- VersionData)
• func KeyReader(db int, inp <-chan []string, out chan<- []RedisKV)
• func KeyUpdatePublisher(db int, listener <-chan RedisUpdates, out chan<- []RedisKV)
• func ReadVersionDelta(so ServerOptions, start <-chan string, out chan<- VersionData, ...)
• func RedisExecutor(URL string)
• func RedisSubscriber(URL string, dbs []int, dbtype map[int]string, listeners []chan<- string)
• func Scan(so ServerOptions, out chan<- VersionData)
• func ScanVersion(so ServerOptions, version uint64, out chan<- VersionData)
• func WriteKeysWithVersionDB(co ClientOptions, msg *sendrecv.Msg, group *sync.WaitGroup)
• func WriteKeysWithVersionSquashDB(co ClientOptions, msg *sendrecv.Msg, group *sync.WaitGroup)
• type ClientOptions
• type DeltaReceiver
  • func (r *DeltaReceiver) Init(so ServerOptions, dbt int, ver uint64)
  • func (r *DeltaReceiver) SubscribeKeys(sname string) <-chan string
  • func (r *DeltaReceiver) SubscribeVersions(sname string) <-chan VersionData
• type GlobalOptions
  • func (o *GlobalOptions) Dump()
• type KHV
• type PubSubStr
  • func (subs *PubSubStr) AddSource(source <-chan string, sname string)
  • func (subs *PubSubStr) AddSubscriber(subscriber chan<- string, name string)
  • func (subs *PubSubStr) InitPubSub(source <-chan string, sname string)
• type PubSubVersion
  • func (subs *PubSubVersion) AddSource(source <-chan VersionData, sname string)
  • func (subs *PubSubVersion) AddSubscriber(subscriber chan<- VersionData, name string)
  • func (subs *PubSubVersion) InitPubSub(source <-chan VersionData, sname string)
• type ReadOperation
• type ReceiverData
  • func (r *ReceiverData) Add(k string, v string)
  • func (r *ReceiverData) Consume(buf *bytes.Buffer, maxChunk int) uint32
  • func (r *ReceiverData) Delete(k string)
  • func (r *ReceiverData) Init(DB int)
• type RedisKV
• type RedisOperation
• type RedisUpdates
• type RetriveOperation
• type ServerOptions
• type TypeConverter
• type VersionData

Constants

const RedisBuf = 10

const RedisExchangeKey = "version"

Variables

var ExecutorGo chan RedisOperation

Functions

func GetVersion

func GetVersion(db int) (uint64, bool)

func KVVersionGenerator

func KVVersionGenerator(so ServerOptions, inc <-chan []RedisKV, outc chan<- VersionData)

func KeyReader

func KeyReader(db int, inp <-chan []string, out chan<- []RedisKV)

func KeyUpdatePublisher

func KeyUpdatePublisher(db int, listener <-chan RedisUpdates, out chan<- []RedisKV)

func ReadVersionDelta

func ReadVersionDelta(so ServerOptions, start <-chan string, out chan<- VersionData, initVersion uint64)

func RedisExecutor

func RedisExecutor(URL string)

func RedisSubscriber

func RedisSubscriber(URL string, dbs []int, dbtype map[int]string, listeners []chan<- string)

func Scan

func Scan(so ServerOptions, out chan<- VersionData)

func ScanVersion

func ScanVersion(so ServerOptions, version uint64, out chan<- VersionData)

func WriteKeysWithVersionDB

func WriteKeysWithVersionDB(co ClientOptions, msg *sendrecv.Msg, group *sync.WaitGroup)

func WriteKeysWithVersionSquashDB

func WriteKeysWithVersionSquashDB(co ClientOptions, msg *sendrecv.Msg, group *sync.WaitGroup)

Types

type ClientOptions

type ClientOptions struct {
	DB int
	//InitVersion int64
	Plain  bool
	Global *GlobalOptions
}

type DeltaReceiver

type DeltaReceiver struct {
	DB           int
	DBtype       int
	StartVersion uint64
	//
	DrainVer <-chan VersionData // for final database replication
	//
	//
	NotifyVer chan<- string // for versionable database replication

	DrainString chan<- string // for non-versionless databases replication
	// contains filtered or unexported fields
}

func (*DeltaReceiver) Init

func (r *DeltaReceiver) Init(so ServerOptions, dbt int, ver uint64)

func (*DeltaReceiver) SubscribeKeys

func (r *DeltaReceiver) SubscribeKeys(sname string) <-chan string

func (*DeltaReceiver) SubscribeVersions

func (r *DeltaReceiver) SubscribeVersions(sname string) <-chan VersionData

type GlobalOptions

type GlobalOptions struct {
	LocalVersion bool
	RedisURL     *string
}

func (*GlobalOptions) Dump

func (o *GlobalOptions) Dump()

type KHV

type KHV struct {
	Key  string
	Hash uint64
	Size uint32
}

type PubSubStr

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

func (*PubSubStr) AddSource

func (subs *PubSubStr) AddSource(source <-chan string, sname string)

Adds source in M to N sources to drains

func (*PubSubStr) AddSubscriber

func (subs *PubSubStr) AddSubscriber(subscriber chan<- string, name string)

func (*PubSubStr) InitPubSub

func (subs *PubSubStr) InitPubSub(source <-chan string, sname string)

type PubSubVersion

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

func (*PubSubVersion) AddSource

func (subs *PubSubVersion) AddSource(source <-chan VersionData, sname string)

Adds source in M to N sources to drains

func (*PubSubVersion) AddSubscriber

func (subs *PubSubVersion) AddSubscriber(subscriber chan<- VersionData, name string)

func (*PubSubVersion) InitPubSub

func (subs *PubSubVersion) InitPubSub(source <-chan VersionData, sname string)

type ReadOperation

type ReadOperation func(int64, string, redis.Conn) (error, RedisKV)

type ReceiverData

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

func (*ReceiverData) Add

func (r *ReceiverData) Add(k string, v string)

func (*ReceiverData) Consume

func (r *ReceiverData) Consume(buf *bytes.Buffer, maxChunk int) uint32

func (*ReceiverData) Delete

func (r *ReceiverData) Delete(k string)

func (*ReceiverData) Init

func (r *ReceiverData) Init(DB int)

type RedisKV

type RedisKV = *sendrecv.Msg_SendValues

type RedisOperation

type RedisOperation func(redis.Conn) error

Factory is a function to create new connections.

type RedisUpdates

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

type RetriveOperation

type RetriveOperation func(string, redis.Conn) error

type ServerOptions

type ServerOptions struct {
	DB    int
	Index int
	//InitVersion int64
	Plain  bool
	Global *GlobalOptions
}

type TypeConverter

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

type VersionData

type VersionData struct {
	Version     uint64
	DeltaFor    uint64
	VersionData []RedisKV
}