sink

Documentation

Index

• Variables
• func NewSinker(lgr log.Logger, cfg *s3_provider.S3Destination, mtrcs metrics.Registry, ...) (*sinker, error)
• type FileCache
  • func (f *FileCache) Add(item *abstract.ChangeItem) error
  • func (f *FileCache) AddCopy(item abstract.ChangeItem) error
  • func (f *FileCache) Clear()
  • func (f *FileCache) Empty() bool
  • func (f *FileCache) ExtractLsns() []uint64
  • func (f *FileCache) IsSnapshotFileCache() bool
  • func (f *FileCache) LSNRange() (uint64, uint64)
  • func (f *FileCache) Split(intervals []ObjectRange, maxCacheSize uint64) []*FileCache
• type ObjectRange
  • func NewObjectRange(from, to uint64) ObjectRange
• type Snapshot
• type SnapshotGzip
  • func NewSnapshotGzip() *SnapshotGzip
  • func (b *SnapshotGzip) Close()
  • func (b *SnapshotGzip) FeedChannel() chan<- []byte
  • func (b *SnapshotGzip) Read(buf []byte) (n int, err error)
• type SnapshotRaw
  • func NewSnapshotRaw() *SnapshotRaw
  • func (b *SnapshotRaw) Close()
  • func (b *SnapshotRaw) FeedChannel() chan<- []byte
  • func (b *SnapshotRaw) Read(buf []byte) (n int, err error)

Variables

var (
	FatalAWSCodes = set.New("InvalidAccessKeyId")
)

Functions

func NewSinker

func NewSinker(lgr log.Logger, cfg *s3_provider.S3Destination, mtrcs metrics.Registry, cp coordinator.Coordinator, transferID string) (*sinker, error)

Types

type FileCache

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

func (*FileCache) Add

func (f *FileCache) Add(item *abstract.ChangeItem) error

func (*FileCache) AddCopy

func (f *FileCache) AddCopy(item abstract.ChangeItem) error

extra copy, but works fine with range, useful for tests

func (*FileCache) Clear

func (f *FileCache) Clear()

func (*FileCache) Empty

func (f *FileCache) Empty() bool

func (*FileCache) ExtractLsns

func (f *FileCache) ExtractLsns() []uint64

func (*FileCache) IsSnapshotFileCache

func (f *FileCache) IsSnapshotFileCache() bool

func (*FileCache) LSNRange

func (f *FileCache) LSNRange() (uint64, uint64)

func (*FileCache) Split

func (f *FileCache) Split(intervals []ObjectRange, maxCacheSize uint64) []*FileCache

Split file cache into file cache parts. Each cache part has items that contain in one of then given intervals and with consecutive LSNs and size that le than maxCacheSize NB intervals range is expected to be sorted

type ObjectRange

type ObjectRange struct {
	From uint64 `json:"from"`
	To   uint64 `json:"to"`
}

func NewObjectRange

func NewObjectRange(from, to uint64) ObjectRange

type Snapshot

type Snapshot interface {
	Read(buf []byte) (n int, err error)
	FeedChannel() chan<- []byte
	Close()
}

type SnapshotGzip

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

func NewSnapshotGzip

func NewSnapshotGzip() *SnapshotGzip

func (*SnapshotGzip) Close

func (b *SnapshotGzip) Close()

func (*SnapshotGzip) FeedChannel

func (b *SnapshotGzip) FeedChannel() chan<- []byte

func (*SnapshotGzip) Read

func (b *SnapshotGzip) Read(buf []byte) (n int, err error)

type SnapshotRaw

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

func NewSnapshotRaw

func NewSnapshotRaw() *SnapshotRaw

func (*SnapshotRaw) Close

func (b *SnapshotRaw) Close()

func (*SnapshotRaw) FeedChannel

func (b *SnapshotRaw) FeedChannel() chan<- []byte

func (*SnapshotRaw) Read

func (b *SnapshotRaw) Read(buf []byte) (n int, err error)