mongoreplay

type Buffer struct {
	Ctx context.Context

	Config *BufferConfig
	// contains filtered or unexported fields
}

type BufferConfig struct {
	Capacity int `json:"capacity"`
}

type Controller struct {
	Ctx context.Context

	SourceCollection *OplogCollection
	DestCollection   *OplogCollection

	StageTracker *StageTracker
	// contains filtered or unexported fields
}

type Dumper struct {
	Ctx           context.Context
	Config        *DumperConfig
	SrcCollection *OplogCollection
	DstCollection *OplogCollection

	LastResumeToken *ResumeTokenStore

	DumperCloseCh chan bool
	DumperCh      chan *MessageN
	// contains filtered or unexported fields
}

type DumperConfig struct {
	FetchCountThreshold int `json:"fetch_count_threshold"`
}

type Filter struct {
	FilterKey   string `json:"filter_key"`
	FilterValue string `json:"filter_value"`
	FilterType  string `json:"filter_type"`
}

type FlusherFunc func(msg *MessageN) (err error)

type MessageN struct {
	CollectionPath string                 `json:"ns"`
	FullDocument   map[string]interface{} `json:"o"`
	OperationType  OperationTypeT         `json:"op"`
	Timestamp      primitive.Timestamp    `json:"ts"`
}

type MongoConfig struct {
	Username string `json:"username"`
	Password string `json:"password"`
	Host     string `json:"host"`
	Port     string `json:"port"`
	DbName   string `json:"db_name"`
}

type NoopStageExecutor struct{}

type OperationTypeT string

type Oplog struct {
	Ctx        context.Context
	CancelFunc context.CancelFunc

	SrcCollections map[string]*OplogCollection

	DstCollections map[string]*OplogCollection
	// contains filtered or unexported fields
}

type OplogCollection struct {
	Name            string            `json:"name"`
	Filters         []Filter          `json:"filters"`
	MongoCollection *mongo.Collection `json:"mongo_collection"`
	MongoDatabase   *mongo.Database   `json:"mongo_database"`
}

type OplogConfig struct {
	DbName      string            `json:"db_name"`
	Collections []OplogCollection `json:"collections"`
}

type OplogTailer struct {
	Ctx context.Context

	Collection *OplogCollection

	FetchCountThreshold int

	WatchThreshold            int
	WatchCount                int
	ShouldHonorWatchThreshold bool

	CtrlrCh chan *MessageN
}

type QueryFunc func(*MessageN) error

type QueryGenerator struct {
	Ctx        context.Context
	Collection *mongo.Collection
	// contains filtered or unexported fields
}

type ResumeTokenStore struct {
	Timestamp primitive.Timestamp `json:"timestamp"`
}

type Seeder struct {
	Count      int
	Collection *OplogCollection

	ShouldClean bool
	// contains filtered or unexported fields
}

type Stage struct {
	StartTime       time.Time `json:"start_time"`
	StopTime        time.Time `json:"stop_time"`
	LastHeartbeatAt time.Time `json:"last_heartbeat_at"`

	StageType StageTypeT `json:"stage_type"`
	Status    StateTypeT `json:"status"`

	Metadata map[string]interface{} `json:"metadata"`
}

type StageExecutor interface {
	Run(...interface{}) error
}

type StageFunction func(context.Context, *OplogCollection, *OplogCollection) (StageExecutor, error)

type StageTracker struct {
	Ctx       context.Context
	CurrStage StageTypeT
	Stages    map[StageTypeT]*Stage

	SrcCollection *OplogCollection
	DstCollection *OplogCollection
	// contains filtered or unexported fields
}

type StageTypeT uint8

type StateTypeT uint8

type TailerManager struct {
	Ctx context.Context

	SourceCollection *OplogCollection
	DestCollection   *OplogCollection

	LastResumeToken *ResumeTokenStore
	// contains filtered or unexported fields
}