repli

Documentation

Index

• Constants
• func NewNoopEmitter() *noopEmitter
• func NewNoopReciver() *noopReciver
• func NewStreamEmitter(ctx runtime.Context, logger *log.Logger, tempDir string, maxPayload int32) *streamEmitter
• func NewStreamReciver(ctx runtime.Context, logger *log.Logger, tempDir string, rto time.Duration) *streamReciver
• type Destination
• type Emitter
• type FileIDAndIndex
• type PartialData
• type Reciver
• type RepliData
• type RepliType
• type RequestCurrentFileID
• type RequestCurrentFileIds
• type RequestCurrentIndex
• type RequestFetchData
• type RequestFetchSize
• type ResponseCurrentFileID
• type ResponseCurrentFileIds
• type ResponseCurrentIndex
• type ResponseFetchData
• type ResponseFetchSize
• type Source

Constants

const (
	SubjectCurrentFileID  string = "current_fileID"
	SubjectCurrentFileIds string = "current_fileids"
	SubjectCurrentIndex   string = "current_index"
	SubjectFetchSize      string = "fetch_size"
	SubjectFetchData      string = "fetch_data"
	SubjectRepli          string = "repli"
)

Functions

func NewNoopEmitter

func NewNoopEmitter() *noopEmitter

func NewNoopReciver

func NewNoopReciver() *noopReciver

func NewStreamEmitter

func NewStreamEmitter(ctx runtime.Context, logger *log.Logger, tempDir string, maxPayload int32) *streamEmitter

func NewStreamReciver

func NewStreamReciver(ctx runtime.Context, logger *log.Logger, tempDir string, rto time.Duration) *streamReciver

Types

type Destination

type Destination interface {
	SetCurrentFileID(datafile.FileID) error
	LastFiles() []FileIDAndIndex
	Insert(fileID datafile.FileID, index int64, checksum uint32, key []byte, r io.Reader, expiry time.Time) error
	Delete(key []byte) error
}

type Emitter

type Emitter interface {
	Start(src Source, bindIP string, bindPort int) error
	Stop() error
	EmitInsert(filer indexer.Filer) error
	EmitDelete(key []byte) error
	EmitCurrentFileID(datafile.FileID) error
}

type FileIDAndIndex

type FileIDAndIndex struct {
	FileID datafile.FileID
	Index  int64
}

type PartialData

type PartialData struct {
	Data []byte
	Err  string
}

type Reciver

type Reciver interface {
	Start(dst Destination, serverIP string, serverPort int) error
	Stop() error
}

type RepliData

type RepliData struct {
	Type        RepliType
	FileID      datafile.FileID
	Index       int64
	Size        int64
	Checksum    uint32
	EntryKey    []byte
	EntryValue  []byte
	EntryExpiry time.Time
	PartKeys    []string
}

type RepliType

type RepliType uint8

const (
	RepliInsert RepliType = iota
	RepliDelete
	RepliCurrentFileID
)

type RequestCurrentFileID

type RequestCurrentFileID struct {
}

type RequestCurrentFileIds

type RequestCurrentFileIds struct {
}

type RequestCurrentIndex

type RequestCurrentIndex struct {
	FileID datafile.FileID
}

type RequestFetchData

type RequestFetchData struct {
	FileID datafile.FileID
	Index  int64
	Size   int64
}

type RequestFetchSize

type RequestFetchSize struct {
	FileID datafile.FileID
	Index  int64
}

type ResponseCurrentFileID

type ResponseCurrentFileID struct {
	FileID datafile.FileID
	Err    string
}

type ResponseCurrentFileIds

type ResponseCurrentFileIds struct {
	FileIds []datafile.FileID
	Err     string
}

type ResponseCurrentIndex

type ResponseCurrentIndex struct {
	Index int64
	Err   string
}

type ResponseFetchData

type ResponseFetchData struct {
	Checksum    uint32
	PartKeys    []string
	EntryKey    []byte
	EntryValue  []byte
	EntryExpiry time.Time
	Err         string
}

type ResponseFetchSize

type ResponseFetchSize struct {
	Size int64
	EOF  bool
	Err  string
}

type Source

type Source interface {
	CurrentFileID() datafile.FileID
	FileIds() []datafile.FileID
	LastIndex(datafile.FileID) int64
	Header(fileID datafile.FileID, index int64) (*datafile.Header, datafile.EOFType, error)
	Read(fileID datafile.FileID, index int64, size int64) (*datafile.Entry, error)
}