source

Documentation

Index

• Constants
• func NewSource() sdk.Source
• type Config
  • func ParseConfig(cfg map[string]string) (Config, error)
• type Iterator
• type Source
  • func (s *Source) Ack(ctx context.Context, position sdk.Position) error
  • func (s *Source) Configure(_ context.Context, cfgRaw map[string]string) (err error)
  • func (s *Source) Open(ctx context.Context, sdkPosition sdk.Position) (err error)
  • func (s *Source) Parameters() map[string]sdk.Parameter
  • func (s *Source) Read(ctx context.Context) (sdk.Record, error)
  • func (s *Source) Teardown(ctx context.Context) error

Constants

const (
	// ConfigKeyOrderingColumn is a config name for an ordering column.
	ConfigKeyOrderingColumn = "orderingColumn"
	// ConfigKeyKeyColumn is a config name for an key column.
	ConfigKeyKeyColumn = "keyColumn"
	// ConfigKeyColumns is a config name for columns.
	ConfigKeyColumns = "columns"
	// ConfigKeyBatchSize is a config name for a batch size.
	ConfigKeyBatchSize = "batchSize"
	// ConfigKeySnapshot is a config name for a snapshot field.
	ConfigKeySnapshot = "snapshot"
)

Functions

func NewSource

func NewSource() sdk.Source

NewSource creates new instance of the Source.

Types

type Config

type Config struct {
	config.Config

	// OrderingColumn is a name of a column that the connector will use for ordering rows.
	OrderingColumn string `key:"orderingColumn" validate:"required,max=64"`
	// KeyColumn is a column name that records should use for their Key fields.
	// Max length is 64, see [MySQL Identifier Length Limits].
	//
	// [MySQL Identifier Length Limits]: https://dev.mysql.com/doc/refman/8.0/en/identifier-length.html
	KeyColumn string `key:"keyColumn" validate:"max=64"`
	// Columns is a comma separated list of column names that should be included in each Record's payload.
	Columns []string `key:"columns" validate:"contains_or_default=OrderingColumn,dive,max=64"`
	// BatchSize is a size of rows batch.
	BatchSize int `key:"batchSize" validate:"gte=1,lte=100000"`
	// Snapshot determines whether or not the connector will take a snapshot
	// of the entire collection before starting CDC mode.
	Snapshot bool `key:"snapshot"`
}

Config holds source specific configurable values.

func ParseConfig

func ParseConfig(cfg map[string]string) (Config, error)

ParseConfig maps the incoming map to the Config and validates it.

type Iterator

type Iterator interface {
	HasNext(ctx context.Context) (bool, error)
	Next(ctx context.Context) (sdk.Record, error)
	Stop(ctx context.Context) error
}

Iterator defines an Iterator interface needed for the Source.

type Source

type Source struct {
	sdk.UnimplementedSource
	// contains filtered or unexported fields
}

Source is a Vitess source plugin.

func (*Source) Ack

func (s *Source) Ack(ctx context.Context, position sdk.Position) error

Ack does nothing. We don't need acks for the Snapshot or CDC iterators. It just returns nil here in order to pass the acceptance tests properly.

func (*Source) Configure

func (s *Source) Configure(_ context.Context, cfgRaw map[string]string) (err error)

Configure parses and initializes the config.

func (*Source) Open

func (s *Source) Open(ctx context.Context, sdkPosition sdk.Position) (err error)

Open makes sure everything is prepared to read records.

func (*Source) Parameters

func (s *Source) Parameters() map[string]sdk.Parameter

Parameters is a map of named Parameters that describe how to configure the Source.

func (*Source) Read

func (s *Source) Read(ctx context.Context) (sdk.Record, error)

Read fetches a record from an iterator. If there's no record will return sdk.ErrBackoffRetry.

func (*Source) Teardown

func (s *Source) Teardown(ctx context.Context) error

Teardown closes connections, stops iterator.