dao

Documentation

Index

• type DB
  • func New(ctx context.Context, cfg config.Config) (*DB, error)
  • func (db *DB) CanSnapshot(namespace string, table string) (bool, error)
  • func (db *DB) Connect(ctx context.Context) error
  • func (db *DB) DatabaseName() string
  • func (db *DB) GetAvroSchema(fingerprint string) (avro.Schema, error)
  • func (db *DB) GetCurrLSN() (lsn pglogrepl.LSN, err error)
  • func (db *DB) GetSchema(fingerprint string) (creek.SchemaMsg, error)
  • func (db *DB) PersistSchemaFromRelation(relation Relation) (string, error)
  • func (db *DB) Snapshot(namespace string, table string) (header creek.SnapshotHeader, reader *SnapshotReader, err error)
  • func (db *DB) StartReplication(tables []string, publicationName string, publicationSlot string) (*Replication, error)
• type Relation
• type Replication
  • func (r *Replication) Close()
  • func (r *Replication) Done() <-chan struct{}
  • func (r *Replication) Next() (creek.WAL, error)
  • func (r *Replication) NextSchema() (creek.SchemaMsg, error)
  • func (r *Replication) SchemaStream() <-chan creek.SchemaMsg
  • func (r *Replication) Stream() <-chan creek.WAL
• type SnapshotReader
  • func (s *SnapshotReader) Error() error
  • func (s *SnapshotReader) Rows() <-chan map[string]interface{}
  • func (s *SnapshotReader) Schema() avro.Schema

Types

type DB

type DB struct {
	InitialLSN pglogrepl.LSN
	// contains filtered or unexported fields
}

func New

func New(ctx context.Context, cfg config.Config) (*DB, error)

func (*DB) CanSnapshot

func (db *DB) CanSnapshot(namespace string, table string) (bool, error)

func (*DB) Connect

func (db *DB) Connect(ctx context.Context) error

func (*DB) DatabaseName

func (db *DB) DatabaseName() string

func (*DB) GetAvroSchema

func (db *DB) GetAvroSchema(fingerprint string) (avro.Schema, error)

func (*DB) GetCurrLSN

func (db *DB) GetCurrLSN() (lsn pglogrepl.LSN, err error)

func (*DB) GetSchema

func (db *DB) GetSchema(fingerprint string) (creek.SchemaMsg, error)

func (*DB) PersistSchemaFromRelation

func (db *DB) PersistSchemaFromRelation(relation Relation) (string, error)

func (*DB) Snapshot

func (db *DB) Snapshot(namespace string, table string) (header creek.SnapshotHeader, reader *SnapshotReader, err error)

func (*DB) StartReplication

func (db *DB) StartReplication(tables []string, publicationName string, publicationSlot string) (*Replication, error)

type Relation

type Relation struct {
	Msg       *pglogrepl.RelationMessage
	Schema    *pgtypeavro.Record // Schema for all fields
	KeySchema *pgtypeavro.Record // Schema for key fields
	// contains filtered or unexported fields
}

Idk, maybe this will be useful in future to have some wrapper functions around relations

type Replication

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

func (*Replication) Close

func (r *Replication) Close()

func (*Replication) Done

func (r *Replication) Done() <-chan struct{}

func (*Replication) Next

func (r *Replication) Next() (creek.WAL, error)

func (*Replication) NextSchema

func (r *Replication) NextSchema() (creek.SchemaMsg, error)

func (*Replication) SchemaStream

func (r *Replication) SchemaStream() <-chan creek.SchemaMsg

func (*Replication) Stream

func (r *Replication) Stream() <-chan creek.WAL

type SnapshotReader

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

func (*SnapshotReader) Error

func (s *SnapshotReader) Error() error

func (*SnapshotReader) Rows

func (s *SnapshotReader) Rows() <-chan map[string]interface{}

func (*SnapshotReader) Schema

func (s *SnapshotReader) Schema() avro.Schema