mongo_watch

README

mongo_watch

MongoDB Runtime Change Stream Filter

Documentation

Index

• func NewNoDecodersForFilterError(db, coll string) error
• type Change
  • func (ch *Change) DocumentKeyID() interface{}
• type DecodeFunc
• type Decoder
• type Filter
• type FilterFunc
• type NoDecodersForFilterError
  • func IsNoDecodersForFilterError(err error) (*NoDecodersForFilterError, bool)
  • func (e *NoDecodersForFilterError) Error() string
• type OpType
• type Option
  • func WithDecodeFunc(db, coll string, decoder DecodeFunc) Option
  • func WithDecoder(decoder Decoder) Option
• type Watch
  • func New(ctx context.Context, client *mongo.Client, ops ...Option) *Watch
  • func (wa *Watch) Errors() <-chan error
  • func (wa *Watch) Subscribe(f ...Filter) func()

Functions

func NewNoDecodersForFilterError

func NewNoDecodersForFilterError(db, coll string) error

Types

type Change

type Change struct {
	ID struct {
		Data string `bson:"_data"`
	} `bson:"_id"`

	OperationType OpType `bson:"operationType"`

	FullDocument bson.Raw `bson:"fullDocument"`

	NS struct {
		DB   string `bson:"db"`
		Coll string `bson:"coll"`
	} `bson:"ns"`

	To *struct {
		DB   string `bson:"db"`
		Coll string `bson:"coll"`
	} `bson:"to"`

	DocumentKey bson.Raw `bson:"documentKey"`

	UpdateDescription *struct {
		UpdatedFields bson.Raw `bson:"updatedFields"`
		RemovedFields []string `bson:"removedFields"`
	} `bson:"updateDescription"`

	ClusterTime primitive.Timestamp `bson:"clusterTime"`

	TxnNumber int64 `bson:"txnNumber"`

	LSID bson.Raw `bson:"lsid"`
}

https://docs.mongodb.com/manual/reference/change-events/

func (*Change) DocumentKeyID

func (ch *Change) DocumentKeyID() interface{}

type DecodeFunc

type DecodeFunc func(doc bson.Raw) (interface{}, error)

type Decoder

type Decoder func() (db, coll string, decode DecodeFunc)

type Filter

type Filter func() (db, coll string, filter FilterFunc)

type FilterFunc

type FilterFunc func(op OpType, v interface{}, change Change)

type NoDecodersForFilterError

type NoDecodersForFilterError struct {
	DB   string
	Coll string
}

func IsNoDecodersForFilterError

func IsNoDecodersForFilterError(err error) (*NoDecodersForFilterError, bool)

func (*NoDecodersForFilterError) Error

func (e *NoDecodersForFilterError) Error() string

type OpType

type OpType string

const (
	OpInsert  OpType = "insert"
	OpDelete  OpType = "delete"
	OpReplace OpType = "replace"
	OpUpdate  OpType = "update"
)

type Option

type Option func(wa *Watch)

func WithDecodeFunc

func WithDecodeFunc(db, coll string, decoder DecodeFunc) Option

func WithDecoder

func WithDecoder(decoder Decoder) Option

type Watch

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

func New

func New(ctx context.Context, client *mongo.Client, ops ...Option) *Watch

func (*Watch) Errors

func (wa *Watch) Errors() <-chan error

func (*Watch) Subscribe

func (wa *Watch) Subscribe(f ...Filter) func()