Documentation
¶
Index ¶
- func GetClient(ctx context.Context, uri string) (*mongo.Client, error)
- type MongoDBSource
- func (m *MongoDBSource) Ack(ctx context.Context, msg hermod.Message) error
- func (m *MongoDBSource) Close() error
- func (m *MongoDBSource) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
- func (m *MongoDBSource) DiscoverDatabases(ctx context.Context) ([]string, error)
- func (m *MongoDBSource) DiscoverTables(ctx context.Context) ([]string, error)
- func (m *MongoDBSource) GetState() map[string]string
- func (m *MongoDBSource) IsReady(ctx context.Context) error
- func (m *MongoDBSource) Ping(ctx context.Context) error
- func (m *MongoDBSource) Read(ctx context.Context) (hermod.Message, error)
- func (m *MongoDBSource) Sample(ctx context.Context, table string) (hermod.Message, error)
- func (m *MongoDBSource) SetInitialLoad(enabled bool)
- func (m *MongoDBSource) SetLogger(logger hermod.Logger)
- func (m *MongoDBSource) SetState(state map[string]string)
- func (m *MongoDBSource) Snapshot(ctx context.Context, tables ...string) error
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func GetClient ¶
GetClient returns the shared client for uri, dialling it on first use.
It takes a context because the verification ping below has to be bounded by whatever the caller allowed. It previously built its own 10s context from Background and ignored the caller's entirely, so a source Read or a readiness Ping with a two-second budget still waited ten — and the driver's own server-selection timeout could stretch that further.
Types ¶
type MongoDBSource ¶
type MongoDBSource struct {
// contains filtered or unexported fields
}
MongoDBSource implements the hermod.Source interface for MongoDB Change Streams.
func NewMongoDBSource ¶
func NewMongoDBSource(uri, database, collection string, useCDC bool) *MongoDBSource
func (*MongoDBSource) Close ¶
func (m *MongoDBSource) Close() error
func (*MongoDBSource) DiscoverColumns ¶
func (m *MongoDBSource) DiscoverColumns(ctx context.Context, table string) ([]hermod.ColumnInfo, error)
func (*MongoDBSource) DiscoverDatabases ¶
func (m *MongoDBSource) DiscoverDatabases(ctx context.Context) ([]string, error)
func (*MongoDBSource) DiscoverTables ¶
func (m *MongoDBSource) DiscoverTables(ctx context.Context) ([]string, error)
func (*MongoDBSource) GetState ¶
func (m *MongoDBSource) GetState() map[string]string
func (*MongoDBSource) SetInitialLoad ¶
func (m *MongoDBSource) SetInitialLoad(enabled bool)
SetInitialLoad asks for a one-time backfill of the watched collection before the change stream is tailed.
It runs only when the source has no record of having run before, so enabling it on a workflow that is already streaming does nothing until that record is cleared. See initialLoadComplete for what the record is and why the resume token cannot serve as one.
func (*MongoDBSource) SetLogger ¶
func (m *MongoDBSource) SetLogger(logger hermod.Logger)
SetLogger installs the logger the engine propagates to sources that accept one. The backfill reports what it could not carry across through it.
func (*MongoDBSource) SetState ¶
func (m *MongoDBSource) SetState(state map[string]string)
Source Files