mongo

Documentation

Index

• type Cursor
  • func (c *Cursor) CloseWithContext(ctx context.Context) error
  • func (c *Cursor) Err() error
  • func (c *Cursor) Next(ctx context.Context) *notify.Notification
• type MessageStream
  • func NewMessageStream(dbClient *mgo.DbClient, col string, resumeStore ResumeStore, query Query, ...) *MessageStream
  • func (stream *MessageStream) Close() error
  • func (stream *MessageStream) CommitOutput(ctx context.Context, messages []notify.Notification) error
  • func (stream *MessageStream) FetchOutputCursor(ctx context.Context) (notify.OutputCursor, error)
• type NotifyInfo
• type Parser
• type Query
• type Resume
  • func (rm *Resume) Decode(s string) error
  • func (rm Resume) Encode() string
• type ResumeStore

Types

type Cursor

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

func (*Cursor) CloseWithContext

func (c *Cursor) CloseWithContext(ctx context.Context) error

func (*Cursor) Err

func (c *Cursor) Err() error

func (*Cursor) Next

func (c *Cursor) Next(ctx context.Context) *notify.Notification

type MessageStream

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

func NewMessageStream

func NewMessageStream(dbClient *mgo.DbClient, col string, resumeStore ResumeStore, query Query, parser Parser) *MessageStream

• query target is mongo change event data example: { "fullDocument.type": "type-A", }

func (*MessageStream) Close

func (stream *MessageStream) Close() error

func (*MessageStream) CommitOutput

func (stream *MessageStream) CommitOutput(ctx context.Context, messages []notify.Notification) error

func (*MessageStream) FetchOutputCursor

func (stream *MessageStream) FetchOutputCursor(ctx context.Context) (notify.OutputCursor, error)

type NotifyInfo

type NotifyInfo struct {
	CreateTime time.Time
	Payload    interface{}
}

type Parser

type Parser func(ctx context.Context, raw bson.Raw) (*NotifyInfo, error)

type Query

type Query struct {
	Fields map[string]interface{}
}

type Resume

type Resume struct {
	Ts primitive.Timestamp
}

func (*Resume) Decode

func (rm *Resume) Decode(s string) error

func (Resume) Encode

func (rm Resume) Encode() string

type ResumeStore

type ResumeStore interface {
	StoreResume(ctx context.Context, resume Resume) error
	GetResume(ctx context.Context) (resume Resume, err error)
}