mongo

Documentation

Index

• func GetClient() *mongo.Client
• func GetTimeoutLimit() time.Duration
• func InitClient(opt *Options) error
• func SetClient(cli *mongo.Client)
• type CollectionInfo
• type Database
• type DummyStreamHandler
  • func (h *DummyStreamHandler) GetCollName() string
  • func (h *DummyStreamHandler) GetDbName() string
  • func (h *DummyStreamHandler) GetOpTypes() []string
  • func (h *DummyStreamHandler) OnChange(*StreamObject) error
• type NS
• type Options
  • func NewDefaultOpt() *Options
• type StreamHandler
• type StreamMonitor
  • func NewDefaultMonitor(opt *Options) (*StreamMonitor, error)
  • func (m *StreamMonitor) AddHandler(handlers ...StreamHandler) error
  • func (m *StreamMonitor) GetHandler(stream *StreamObject) (StreamHandler, bool)
  • func (m *StreamMonitor) SetStore(store cdcstore.MongoPosInterface)
  • func (m *StreamMonitor) SetWatcher() error
  • func (m *StreamMonitor) StartMonitor() error
• type StreamObject
• type UpdateDescription
• type WatchId

Functions

func GetClient

func GetClient() *mongo.Client

func GetTimeoutLimit

func GetTimeoutLimit() time.Duration

func InitClient

func InitClient(opt *Options) error

func SetClient

func SetClient(cli *mongo.Client)

Types

type CollectionInfo

type CollectionInfo struct {
	Name    string   `yaml:"name"`
	Actions []string `yaml:"actions"`
}

type Database

type Database struct {
	Name        string            `yaml:"name"`
	Collections []*CollectionInfo `yaml:"collections"`
}

type DummyStreamHandler

type DummyStreamHandler struct {
	DbName   string
	CollName string
	OpTypes  []string
}

func (*DummyStreamHandler) GetCollName

func (h *DummyStreamHandler) GetCollName() string

func (*DummyStreamHandler) GetDbName

func (h *DummyStreamHandler) GetDbName() string

func (*DummyStreamHandler) GetOpTypes

func (h *DummyStreamHandler) GetOpTypes() []string

func (*DummyStreamHandler) OnChange

func (h *DummyStreamHandler) OnChange(*StreamObject) error

type NS

type NS struct {
	Database   string `bson:"db"`
	Collection string `bson:"coll"`
}

NS Changed database

type Options

type Options struct {
	Addr      string      `yaml:"addr"`
	User      string      `yaml:"user"`
	Pwd       string      `yaml:"pwd"`
	Auth      bool        `yaml:"auth"`
	Direct    bool        `yaml:"direct"`
	PoolSize  uint64      `yaml:"pool_size"`
	Timeout   uint64      `yaml:"timeout"`
	Databases []*Database `yaml:"databases"`
	// contains filtered or unexported fields
}

func NewDefaultOpt

func NewDefaultOpt() *Options

type StreamHandler

type StreamHandler interface {
	GetDbName() string
	GetCollName() string
	GetOpTypes() []string
	OnChange(*StreamObject) error
}

type StreamMonitor

type StreamMonitor struct {
	cdcstore.MongoPosInterface
	// contains filtered or unexported fields
}

func NewDefaultMonitor

func NewDefaultMonitor(opt *Options) (*StreamMonitor, error)

func (*StreamMonitor) AddHandler

func (m *StreamMonitor) AddHandler(handlers ...StreamHandler) error

func (*StreamMonitor) GetHandler

func (m *StreamMonitor) GetHandler(stream *StreamObject) (StreamHandler, bool)

func (*StreamMonitor) SetStore

func (m *StreamMonitor) SetStore(store cdcstore.MongoPosInterface)

func (*StreamMonitor) SetWatcher

func (m *StreamMonitor) SetWatcher() error

func (*StreamMonitor) StartMonitor

func (m *StreamMonitor) StartMonitor() error

type StreamObject

type StreamObject struct {
	Id                *WatchId `bson:"_id"`
	OperationType     string
	FullDocument      bson.Raw
	Ns                NS
	UpdateDescription *UpdateDescription
	DocumentKey       map[string]interface{}
}

type UpdateDescription

type UpdateDescription struct {
	RemoveFields []string               `bson:"removeFields"`
	UpdateFields map[string]interface{} `bson:"updateFields"`
}

UpdateDescription 修改的document字段和值

type WatchId

type WatchId struct {
	Data string `bson:"_data"`
}

WatchId for resume token Specifies the logical starting point for the new change stream