ckpt

Documentation

Index

• Constants
• type CheckpointContext
  • func (cc *CheckpointContext) String() string
• type CheckpointManager
  • func NewCheckpointManager(name string, startPosition int64) *CheckpointManager
  • func (manager *CheckpointManager) Get() (*CheckpointContext, bool, error)
  • func (manager *CheckpointManager) GetInMemory() *CheckpointContext
  • func (manager *CheckpointManager) SetFetchMethod(method string)
  • func (manager *CheckpointManager) SetOplogDiskFinishTs(ts int64)
  • func (manager *CheckpointManager) SetOplogDiskQueueName(name string)
  • func (manager *CheckpointManager) Update(ts int64) error
• type CheckpointOperation
• type HttpApiCheckpoint
  • func (ckpt *HttpApiCheckpoint) Get() (*CheckpointContext, bool)
  • func (ckpt *HttpApiCheckpoint) Insert(insert *CheckpointContext) error
• type MongoCheckpoint
  • func (ckpt *MongoCheckpoint) Get() (*CheckpointContext, bool)
  • func (ckpt *MongoCheckpoint) Insert(updates *CheckpointContext) error

Constants

const (
	// we can't insert Timestamp(0, 0) that will be treat as Now() inserted
	// into mongo. so we use Timestamp(0, 1)
	InitCheckpoint  = int64(1)
	EmptyCheckpoint = int64(0)
)

const (
	CheckpointName = "name"
)

Types

type CheckpointContext

type CheckpointContext struct {
	Name                   string `bson:"name" json:"name"`
	Timestamp              int64  `bson:"ckpt" json:"ckpt"`
	Version                int    `bson:"version" json:"version"`
	FetchMethod            string `bson:"fetch_method" json:"fetch_method"`
	OplogDiskQueue         string `bson:"oplog_disk_queue" json:"oplog_disk_queue"`
	OplogDiskQueueFinishTs int64  `bson:"oplog_disk_queue_apply_finish_ts" json:"oplog_disk_queue_apply_finish_ts"`
}

func (*CheckpointContext) String

func (cc *CheckpointContext) String() string

type CheckpointManager

type CheckpointManager struct {
	Type string
	// contains filtered or unexported fields
}

func NewCheckpointManager

func NewCheckpointManager(name string, startPosition int64) *CheckpointManager

func (*CheckpointManager) Get

func (manager *CheckpointManager) Get() (*CheckpointContext, bool, error)

get persist checkpoint

func (*CheckpointManager) GetInMemory

func (manager *CheckpointManager) GetInMemory() *CheckpointContext

get in memory checkpoint

func (*CheckpointManager) SetFetchMethod

func (manager *CheckpointManager) SetFetchMethod(method string)

func (*CheckpointManager) SetOplogDiskFinishTs

func (manager *CheckpointManager) SetOplogDiskFinishTs(ts int64)

OplogDiskQueueFinishTs and OplogDiskQueue won't immediate effect, will be inserted in the next Update call.

func (*CheckpointManager) SetOplogDiskQueueName

func (manager *CheckpointManager) SetOplogDiskQueueName(name string)

func (*CheckpointManager) Update

func (manager *CheckpointManager) Update(ts int64) error

Update checkpoint update memory & persistence(db or file)

type CheckpointOperation

type CheckpointOperation interface {
	// read checkpoint from remote storage. and encapsulation
	// with CheckpointContext struct
	// bool means whether exists on remote
	Get() (*CheckpointContext, bool)

	// save checkpoint
	Insert(ckpt *CheckpointContext) error

	// log info
	String() string
}

type HttpApiCheckpoint

type HttpApiCheckpoint struct {
	CheckpointContext

	URL string
}

http

func (*HttpApiCheckpoint) Get

func (ckpt *HttpApiCheckpoint) Get() (*CheckpointContext, bool)

func (*HttpApiCheckpoint) Insert

func (ckpt *HttpApiCheckpoint) Insert(insert *CheckpointContext) error

type MongoCheckpoint

type MongoCheckpoint struct {
	CheckpointContext

	// connection info
	URL       string
	DB, Table string
	// contains filtered or unexported fields
}

mongo

func (*MongoCheckpoint) Get

func (ckpt *MongoCheckpoint) Get() (*CheckpointContext, bool)

func (*MongoCheckpoint) Insert

func (ckpt *MongoCheckpoint) Insert(updates *CheckpointContext) error