iterator

Documentation

Index

• Variables
• type CDCIterator
  • func NewCDCIterator(ctx context.Context, db *sqlx.DB, table, trackingTable, key string, ...) (*CDCIterator, error)
  • func (i *CDCIterator) Ack(_ context.Context, pos *position.Position) error
  • func (i *CDCIterator) ClearTrackingTable(ctx context.Context)
  • func (i *CDCIterator) DeleteRows(ctx context.Context, ids []any) error
  • func (i *CDCIterator) HasNext(ctx context.Context) (bool, error)
  • func (i *CDCIterator) Next(ctx context.Context) (sdk.Record, error)
  • func (i *CDCIterator) Stop(ctx context.Context) error
• type CombinedIterator
  • func NewCombinedIterator(ctx context.Context, db *sqlx.DB, conn, table, key, orderingColumn string, ...) (*CombinedIterator, error)
  • func (c *CombinedIterator) Ack(ctx context.Context, rp sdk.Position) error
  • func (c *CombinedIterator) HasNext(ctx context.Context) (bool, error)
  • func (c *CombinedIterator) Next(ctx context.Context) (sdk.Record, error)
  • func (c *CombinedIterator) SetupCDC(ctx context.Context, db *sqlx.DB) error
  • func (c *CombinedIterator) Stop(ctx context.Context) error
• type SnapshotIterator
  • func NewSnapshotIterator(ctx context.Context, db *sqlx.DB, table, orderingColumn, key string, ...) (*SnapshotIterator, error)
  • func (i *SnapshotIterator) Ack(ctx context.Context, p sdk.Position) error
  • func (i *SnapshotIterator) HasNext(ctx context.Context) (bool, error)
  • func (i *SnapshotIterator) Next(ctx context.Context) (sdk.Record, error)
  • func (i *SnapshotIterator) Stop(context.Context) error

Variables

var (
	ErrKeyIsNotExist            = errors.New("key is not exist")
	ErrOrderingColumnIsNotExist = errors.New("ordering column is not exist")
	ErrWrongTrackingIDType      = errors.New("tracking id wrong type")
	ErrNoInitializedIterator    = errors.New("not initialized iterator")
	ErrUnknownOperatorType      = errors.New("unknown iterator type")
)

Types

type CDCIterator

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

CDCIterator - cdc iterator.

func NewCDCIterator

func NewCDCIterator(
	ctx context.Context,
	db *sqlx.DB,
	table, trackingTable, key string,
	columns []string,
	batchSize int,
	position *position.Position,
	columnTypes map[string]string,
) (*CDCIterator, error)

NewCDCIterator create new cdc iterator.

func (*CDCIterator) Ack

func (i *CDCIterator) Ack(_ context.Context, pos *position.Position) error

Ack check if record with position was recorded.

func (*CDCIterator) ClearTrackingTable

func (i *CDCIterator) ClearTrackingTable(ctx context.Context)

ClearTrackingTable remove recorded rows from tracking table.

func (*CDCIterator) DeleteRows

func (i *CDCIterator) DeleteRows(ctx context.Context, ids []any) error

func (*CDCIterator) HasNext

func (i *CDCIterator) HasNext(ctx context.Context) (bool, error)

HasNext check ability to get next record.

func (*CDCIterator) Next

func (i *CDCIterator) Next(ctx context.Context) (sdk.Record, error)

Next get new record.

func (*CDCIterator) Stop

func (i *CDCIterator) Stop(ctx context.Context) error

Stop shutdown iterator.

type CombinedIterator

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

CombinedIterator combined iterator.

func NewCombinedIterator

func NewCombinedIterator(
	ctx context.Context,
	db *sqlx.DB,
	conn, table, key, orderingColumn string,
	columns []string,
	batchSize int,
	snapshot bool,
	sdkPosition sdk.Position,
) (*CombinedIterator, error)

NewCombinedIterator - create new iterator.

func (*CombinedIterator) Ack

func (c *CombinedIterator) Ack(ctx context.Context, rp sdk.Position) error

Ack collect tracking ids for removing.

func (*CombinedIterator) HasNext

func (c *CombinedIterator) HasNext(ctx context.Context) (bool, error)

HasNext returns a bool indicating whether the iterator has the next record to return or not. If the underlying snapshot iterator returns false, the combined iterator will try to switch to the cdc iterator.

func (*CombinedIterator) Next

func (c *CombinedIterator) Next(ctx context.Context) (sdk.Record, error)

Next returns the next record.

func (*CombinedIterator) SetupCDC

func (c *CombinedIterator) SetupCDC(ctx context.Context, db *sqlx.DB) error

SetupCDC - create tracking table, add columns, add triggers, set identity column.

func (*CombinedIterator) Stop

func (c *CombinedIterator) Stop(ctx context.Context) error

Stop the underlying iterators.

type SnapshotIterator

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

SnapshotIterator - snapshot iterator.

func NewSnapshotIterator

func NewSnapshotIterator(
	ctx context.Context,
	db *sqlx.DB,
	table, orderingColumn, key string,
	columns []string,
	batchSize int,
	position *position.Position,
	columnTypes map[string]string,
) (*SnapshotIterator, error)

func (*SnapshotIterator) Ack

func (i *SnapshotIterator) Ack(ctx context.Context, p sdk.Position) error

Ack check if record with position was recorded.

func (*SnapshotIterator) HasNext

func (i *SnapshotIterator) HasNext(ctx context.Context) (bool, error)

HasNext check ability to get next record.

func (*SnapshotIterator) Next

func (i *SnapshotIterator) Next(ctx context.Context) (sdk.Record, error)

Next get new record.

func (*SnapshotIterator) Stop

func (i *SnapshotIterator) Stop(context.Context) error

Stop shutdown iterator.