mongobatch

Documentation

Index

• Constants
• func Decode(s string) (interface{}, error)
• func Encode(v interface{}) (string, error)
• func SetupInitialPosition(cache position_cache.PositionCacheInterface, session *mgo.Session, ...) error
• type Chunk
• type Config
• type IDValue
  • func (idValue IDValue) MarshalJSON() ([]byte, error)
  • func (idValue *IDValue) UnmarshalJSON(value []byte) error
• type PositionValue

Constants

const (
	PlainString  = "string"
	PlainInt     = "int"
	PlainUInt    = "uint"
	BsonObjectID = "bsonID"
)

const Name = "mongo-batch"

Functions

func Decode

func Decode(s string) (interface{}, error)

func Encode

func Encode(v interface{}) (string, error)

func SetupInitialPosition

func SetupInitialPosition(
	cache position_cache.PositionCacheInterface,
	session *mgo.Session,
	router core.Router,
	cfg Config) error

Types

type Chunk

type Chunk struct {
	Database   string   `json:"database" bson:"database"`
	Collection string   `json:"collection" bson:"collection"`
	Seq        int      `json:"seq" bson:"seq"`
	Done       bool     `json:"done" bson:"done"`
	Min        *IDValue `json:"min,omitempty" bson:"min,omitempty"`
	Max        *IDValue `json:"max,omitempty" bson:"max,omitempty"`
	Current    *IDValue `json:"current,omitempty" bson:"current,omitempty"`
	Scanned    int64    `json:"scanned" bson:"scanned"`
}

type Config

type Config struct {
	Source              *config.MongoConnConfig     `mapstructure:"source" toml:"source" json:"source"`
	PositionRepo        *config.GenericPluginConfig `mapstructure:"position-repo" toml:"position-repo" json:"position-repo"`
	BatchSize           int                         `mapstructure:"batch-size"  toml:"batch-size" json:"batch-size"`
	WorkerCnt           int                         `mapstructure:"worker-cnt" toml:"worker-cnt" json:"worker-cnt"`
	ChunkThreshold      int                         `mapstructure:"chunk-threshold"  toml:"chunk-threshold"  json:"chunk-threshold"`
	BatchPerSecondLimit int                         `mapstructure:"batch-per-second-limit" toml:"batch-per-second-limit" json:"batch-per-second-limit"`

	// IgnoreOplogError ignores error with oplog.
	// Some mongo cluster setup may not support oplog.
	IgnoreOplogError bool `mapstructure:"ignore-oplog-error" toml:"ignore-oplog-error" json:"ignore-oplog-error"`
}

type IDValue

type IDValue struct {
	Value interface{} `json:"value"`
	Type  string      `json:"type"`
}

func (IDValue) MarshalJSON

func (idValue IDValue) MarshalJSON() ([]byte, error)

func (*IDValue) UnmarshalJSON

func (idValue *IDValue) UnmarshalJSON(value []byte) error

type PositionValue

type PositionValue struct {
	Start  bson.MongoTimestamp `bson:"start" json:"start"`
	Chunks []Chunk             `bson:"chunks"  json:"chunks"`
}