incr

Documentation

Index

• Constants
• func BuildUpdateDelteOplog(prefixField string, obj bson.D) (interface{}, error)
• func DiffUpdateOplogToNormal(updateObj bson.D) (interface{}, error)
• func FindFiledPrefix(input bson.D, prefix string) bool
• func GetKey(log bson.D, wanted string) interface{}
• func GetKeyWithIndex(log bson.D, wanted string) (interface{}, int)
• func GetObjectId(log bson.D) (primitive.ObjectID, error)
• func IgnoreError(err error, op string, isFullSyncStage bool) bool
• func RemoveFiled(input bson.D, key string) bson.D
• func StartIncrementalReplication(ctx context.Context, checkpointManager checkpoint.CheckpointManager)
• type OplogReader
  • func NewOplogReader(ckptManager checkpoint.CheckpointManager) *OplogReader
  • func (o *OplogReader) StartReader(ctx context.Context)
  • func (o *OplogReader) StopReader()
• type OplogWriterSingle
  • func NewOplogWriter(fullFinishTs int64, queuedLogs chan *oplog.ChangeLog, ...) *OplogWriterSingle
  • func (ow *OplogWriterSingle) Delete(l *oplog.ChangeLog) error
  • func (ow *OplogWriterSingle) Insert(l *oplog.ChangeLog) error
  • func (ow *OplogWriterSingle) StartWriter(ctx context.Context)
  • func (ow *OplogWriterSingle) StopWriter()
  • func (ow *OplogWriterSingle) Update(l *oplog.ChangeLog, upsert bool) error
  • func (sw *OplogWriterSingle) Upsert(l *oplog.ChangeLog, upsert bool) error
• type Reader
• type Writer

Constants

const (
	VersionMark       = "$v"
	OplogVersionError = "unknown version of OPLog"
	OperationError    = "operation error"
	InsertError       = "insert error"
	UpdateError       = "update error"
	DeleteError       = "delete error"
	UpsertError       = "upsert error"
	DuplicateError    = "duplicate error"
	MatchedCountError = "matched count error"
)

const (
	CursorWaitTime = 5 * time.Second
)

Functions

func BuildUpdateDelteOplog

func BuildUpdateDelteOplog(prefixField string, obj bson.D) (interface{}, error)

func DiffUpdateOplogToNormal

func DiffUpdateOplogToNormal(updateObj bson.D) (interface{}, error)

"o" : { "$v" : 2, "diff" : { "d" : { "count" : false }, "u" : { "name" : "orange" }, "i" : { "c" : 11 } } }

func FindFiledPrefix

func FindFiledPrefix(input bson.D, prefix string) bool

func GetKey

func GetKey(log bson.D, wanted string) interface{}

func GetKeyWithIndex

func GetKeyWithIndex(log bson.D, wanted string) (interface{}, int)

func GetObjectId

func GetObjectId(log bson.D) (primitive.ObjectID, error)

func IgnoreError

func IgnoreError(err error, op string, isFullSyncStage bool) bool

true means error can be ignored https://github.com/mongodb/mongo/blob/master/src/mongo/base/error_codes.yml

func RemoveFiled

func RemoveFiled(input bson.D, key string) bson.D

pay attention: the input bson.D will be modified.

func StartIncrementalReplication

func StartIncrementalReplication(ctx context.Context, checkpointManager checkpoint.CheckpointManager)

Types

type OplogReader

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

func NewOplogReader

func NewOplogReader(ckptManager checkpoint.CheckpointManager) *OplogReader

func (*OplogReader) StartReader

func (o *OplogReader) StartReader(ctx context.Context)

func (*OplogReader) StopReader

func (o *OplogReader) StopReader()

type OplogWriterSingle

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

func NewOplogWriter

func NewOplogWriter(fullFinishTs int64, queuedLogs chan *oplog.ChangeLog, ckptManager checkpoint.CheckpointManager) *OplogWriterSingle

func (*OplogWriterSingle) Delete

func (ow *OplogWriterSingle) Delete(l *oplog.ChangeLog) error

func (*OplogWriterSingle) Insert

func (ow *OplogWriterSingle) Insert(l *oplog.ChangeLog) error

func (*OplogWriterSingle) StartWriter

func (ow *OplogWriterSingle) StartWriter(ctx context.Context)

Start the writer

func (*OplogWriterSingle) StopWriter

func (ow *OplogWriterSingle) StopWriter()

func (*OplogWriterSingle) Update

func (ow *OplogWriterSingle) Update(l *oplog.ChangeLog, upsert bool) error

Update the document

func (*OplogWriterSingle) Upsert

func (sw *OplogWriterSingle) Upsert(l *oplog.ChangeLog, upsert bool) error

Upsert the document

type Reader

type Reader interface {
	StartReader(context.Context)
	StopReader(context.Context)
}

type Writer

type Writer interface {
	StartWriter(context.Context)
	StopWriter(context.Context)
}