full

Documentation

Index

• Constants
• func StartFullReplication(ctx context.Context, checkpointManager checkpoint.CheckpointManager, ...)
• type DocumentReader
  • func NewDocumentReader(database string, collection string, source *mong.Mong, batch int, ...) *DocumentReader
  • func (r *DocumentReader) ReportResult(result WriteResult)
  • func (r *DocumentReader) SetProgress(progress *SyncProgress)
  • func (r *DocumentReader) StartSync(ctx context.Context) error
• type DocumentWriter
  • func NewDocumentWriter(database string, collection string, target *mong.Mong) *DocumentWriter
  • func (r *DocumentWriter) SetProgress(progress *SyncProgress)
  • func (r *DocumentWriter) WriteDocuments(docs []*bson.Raw) (WriteResult, error)
• type QpsLimit
  • func NewQpsLimit(maxQps uint) *QpsLimit
  • func (q *QpsLimit) Incr(incr int)
  • func (q *QpsLimit) Reset()
  • func (q *QpsLimit) Wait()
• type SyncProgress
  • func NewSyncProgress(database string, collection string) *SyncProgress
  • func (f *SyncProgress) Increment(incr int)
  • func (f *SyncProgress) Progress() float64
  • func (f *SyncProgress) SetTotal(total uint64)
• type WriteResult

Constants

const (
	MAX_BUFFER_BYTE_SIZE = 12 * 1024 * 1024
)

Functions

func StartFullReplication

func StartFullReplication(ctx context.Context, checkpointManager checkpoint.CheckpointManager,
	dbAndCollections map[string][]string)

Types

type DocumentReader

type DocumentReader struct {
	// The database name
	Database string
	// The name of the collection
	Collection string
	// The batch size
	Batch int
	// The source database
	Source *mong.Mong
	// The document writer
	Writer *DocumentWriter
	// Progression state
	Progress *SyncProgress
}

func NewDocumentReader

func NewDocumentReader(database string, collection string, source *mong.Mong, batch int, writer *DocumentWriter) *DocumentReader

func (*DocumentReader) ReportResult

func (r *DocumentReader) ReportResult(result WriteResult)

Report the result of the write operation

func (*DocumentReader) SetProgress

func (r *DocumentReader) SetProgress(progress *SyncProgress)

Set the total count of documents to sync

func (*DocumentReader) StartSync

func (r *DocumentReader) StartSync(ctx context.Context) error

Read a batch of documents from the source

type DocumentWriter

type DocumentWriter struct {
	// The database name
	Database string
	// The name of the collection
	Collection string
	// The source database
	Target *mong.Mong
	// Progression state
	Progress *SyncProgress
}

func NewDocumentWriter

func NewDocumentWriter(database string, collection string, target *mong.Mong) *DocumentWriter

func (*DocumentWriter) SetProgress

func (r *DocumentWriter) SetProgress(progress *SyncProgress)

Set the total count of documents to sync

func (*DocumentWriter) WriteDocuments

func (r *DocumentWriter) WriteDocuments(docs []*bson.Raw) (WriteResult, error)

Sync the documents to the target

type QpsLimit

type QpsLimit struct {
	Start  time.Time
	MaxQps uint
	// contains filtered or unexported fields
}

func NewQpsLimit

func NewQpsLimit(maxQps uint) *QpsLimit

func (*QpsLimit) Incr

func (q *QpsLimit) Incr(incr int)

func (*QpsLimit) Reset

func (q *QpsLimit) Reset()

func (*QpsLimit) Wait

func (q *QpsLimit) Wait()

type SyncProgress

type SyncProgress struct {
	Database   string
	Collection string
	// contains filtered or unexported fields
}

func NewSyncProgress

func NewSyncProgress(database string, collection string) *SyncProgress

func (*SyncProgress) Increment

func (f *SyncProgress) Increment(incr int)

func (*SyncProgress) Progress

func (f *SyncProgress) Progress() float64

func (*SyncProgress) SetTotal

func (f *SyncProgress) SetTotal(total uint64)

type WriteResult

type WriteResult struct {
	InsertedCount           int
	UpdatedCount            int
	SkippedOnDuplicateCount int
	ErrorCount              int
}