docsyncer

Documentation

Index

• Constants
• Variables
• func FlushCheckpoint(ckptMap map[string]bson.MongoTimestamp) error
• func GenerateCollExecutorId() int
• func GenerateDocExecutorId() int
• func GetAllNamespace(sources []*utils.MongoSource) (map[utils.NS]bool, error)
• func IsShardingToSharding(fromIsSharding bool, toConn *utils.MongoConn) bool
• func LoadCheckpoint() (map[string]bson.MongoTimestamp, error)
• func StartDropDestCollection(nsSet map[utils.NS]bool, toConn *utils.MongoConn, ...) (map[string]bool, error)
• func StartIndexSync(indexMap map[utils.NS][]mgo.Index, toUrl string, nsExistedSet map[string]bool, ...) (syncError error)
• func StartNamespaceSpecSyncForSharding(csUrl string, toConn *utils.MongoConn, nsExistedSet map[string]bool, ...) error
• type CollectionExecutor
  • func NewCollectionExecutor(id int, mongoUrl string, ns utils.NS) *CollectionExecutor
  • func (colExecutor *CollectionExecutor) Start() error
  • func (colExecutor *CollectionExecutor) Sync(docs []*bson.Raw)
  • func (colExecutor *CollectionExecutor) Wait() error
• type DBSyncer
  • func NewDBSyncer(replset string, fromMongoUrl string, toMongoUrl string, ...) *DBSyncer
  • func (syncer *DBSyncer) GetIndexMap() map[utils.NS][]mgo.Index
  • func (syncer *DBSyncer) Start() (syncError error)
• type DocExecutor
  • func NewDocExecutor(id int, colExecutor *CollectionExecutor, session *mgo.Session) *DocExecutor
• type DocumentReader
  • func NewDocumentReader(src string, ns utils.NS) *DocumentReader
  • func (reader *DocumentReader) Close()
  • func (reader *DocumentReader) GetIndexes() ([]mgo.Index, error)
  • func (reader *DocumentReader) NextDoc() (doc *bson.Raw, err error)

Constants

const (
	MAX_BUFFER_BYTE_SIZE = 16 * 1024 * 1024
)

Variables

var GlobalCollExecutorId int32 = -1

var GlobalDocExecutorId int32 = -1

Functions

func FlushCheckpoint

func FlushCheckpoint(ckptMap map[string]bson.MongoTimestamp) error

func GenerateCollExecutorId

func GenerateCollExecutorId() int

func GenerateDocExecutorId

func GenerateDocExecutorId() int

func GetAllNamespace

func GetAllNamespace(sources []*utils.MongoSource) (map[utils.NS]bool, error)

func IsShardingToSharding

func IsShardingToSharding(fromIsSharding bool, toConn *utils.MongoConn) bool

func LoadCheckpoint

func LoadCheckpoint() (map[string]bson.MongoTimestamp, error)

func StartDropDestCollection

func StartDropDestCollection(nsSet map[utils.NS]bool, toConn *utils.MongoConn,
	nsTrans *transform.NamespaceTransform) (map[string]bool, error)

func StartIndexSync

func StartIndexSync(indexMap map[utils.NS][]mgo.Index, toUrl string,
	nsExistedSet map[string]bool, nsTrans *transform.NamespaceTransform) (syncError error)

func StartNamespaceSpecSyncForSharding

func StartNamespaceSpecSyncForSharding(csUrl string, toConn *utils.MongoConn,
	nsExistedSet map[string]bool, nsTrans *transform.NamespaceTransform) error

Types

type CollectionExecutor

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

func NewCollectionExecutor

func NewCollectionExecutor(id int, mongoUrl string, ns utils.NS) *CollectionExecutor

func (*CollectionExecutor) Start

func (colExecutor *CollectionExecutor) Start() error

func (*CollectionExecutor) Sync

func (colExecutor *CollectionExecutor) Sync(docs []*bson.Raw)

func (*CollectionExecutor) Wait

func (colExecutor *CollectionExecutor) Wait() error

type DBSyncer

type DBSyncer struct {

	// source mongodb url
	FromMongoUrl string
	// destination mongodb url
	ToMongoUrl string
	// contains filtered or unexported fields
}

func NewDBSyncer

func NewDBSyncer(
	replset string,
	fromMongoUrl string,
	toMongoUrl string,
	nsTrans *transform.NamespaceTransform,
	orphanFilter *filter.OrphanFilter) *DBSyncer

func (*DBSyncer) GetIndexMap

func (syncer *DBSyncer) GetIndexMap() map[utils.NS][]mgo.Index

func (*DBSyncer) Start

func (syncer *DBSyncer) Start() (syncError error)

type DocExecutor

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

func NewDocExecutor

func NewDocExecutor(id int, colExecutor *CollectionExecutor, session *mgo.Session) *DocExecutor

type DocumentReader

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

func NewDocumentReader

func NewDocumentReader(src string, ns utils.NS) *DocumentReader

NewDocumentReader creates reader with mongodb url

func (*DocumentReader) Close

func (reader *DocumentReader) Close()

func (*DocumentReader) GetIndexes

func (reader *DocumentReader) GetIndexes() ([]mgo.Index, error)

func (*DocumentReader) NextDoc

func (reader *DocumentReader) NextDoc() (doc *bson.Raw, err error)

NextDoc returns an document by raw bytes which is []byte