postgres

Documentation

Index

• type CDCConfig
• type CDCFactory
  • func (f *CDCFactory) Create(config adapters.SourceConfig) (adapters.FactSource, error)
  • func (f *CDCFactory) GetConfigSchema() adapters.ConfigSchema
  • func (f *CDCFactory) ValidateConfig(config adapters.SourceConfig) error
• type CDCSource
  • func NewCDCSource(config *CDCConfig) (*CDCSource, error)
  • func (c *CDCSource) GetMetadata() adapters.SourceMetadata
  • func (c *CDCSource) GetSourceSchema() *adapters.Schema
  • func (c *CDCSource) HealthCheck() error
  • func (c *CDCSource) Start(ctx context.Context) error
  • func (c *CDCSource) Stop(ctx context.Context) error
  • func (c *CDCSource) Subscribe(ctx context.Context, factTypes []string) (<-chan *adapters.TypedFact, error)
• type ChangeEvent
• type ChangeTransformer
  • func NewChangeTransformer(config *CDCConfig) *ChangeTransformer
  • func (t *ChangeTransformer) TransformChange(change *ChangeEvent) (*adapters.TypedFact, error)
• type PollerConfig
• type PostgresPollerFactory
  • func (f *PostgresPollerFactory) Create(config adapters.SourceConfig) (adapters.FactSource, error)
  • func (f *PostgresPollerFactory) GetConfigSchema() adapters.ConfigSchema
  • func (f *PostgresPollerFactory) ValidateConfig(config adapters.SourceConfig) error
• type PostgresPollerSource
  • func NewPostgresPollerSource(sourceID string, config PollerConfig) (*PostgresPollerSource, error)
  • func (p *PostgresPollerSource) GetMetadata() adapters.SourceMetadata
  • func (p *PostgresPollerSource) GetSourceSchema() *adapters.Schema
  • func (p *PostgresPollerSource) HealthCheck() error
  • func (p *PostgresPollerSource) Start(ctx context.Context) error
  • func (p *PostgresPollerSource) Stop(ctx context.Context) error
  • func (p *PostgresPollerSource) Subscribe(ctx context.Context, factTypes []string) (<-chan *adapters.TypedFact, error)

Types

type CDCConfig

type CDCConfig struct {
	SourceID             string                    `json:"source_id" yaml:"source_id"`
	SourceType           string                    `json:"source_type" yaml:"source_type"`
	ConnectionString     string                    `json:"connection_string" yaml:"connection_string"`
	SlotName             string                    `json:"slot_name" yaml:"slot_name"`
	PublicationName      string                    `json:"publication_name" yaml:"publication_name"`
	Plugin               string                    `json:"plugin" yaml:"plugin"`
	CreateSlot           bool                      `json:"create_slot" yaml:"create_slot"`
	Tables               []string                  `json:"tables" yaml:"tables"`
	Operations           []string                  `json:"operations" yaml:"operations"` // INSERT, UPDATE, DELETE
	SchemaMapping        map[string]string         `json:"schema_mapping" yaml:"schema_mapping"`
	StartLSN             string                    `json:"start_lsn" yaml:"start_lsn"`
	BatchSize            int                       `json:"batch_size" yaml:"batch_size"`
	HeartbeatIntervalSec int                       `json:"heartbeat_interval_sec" yaml:"heartbeat_interval_sec"`
	BufferSize           int                       `json:"buffer_size" yaml:"buffer_size"`
	PollInterval         time.Duration             `json:"poll_interval" yaml:"poll_interval"`
	MaxChanges           int                       `json:"max_changes" yaml:"max_changes"`
	Transforms           []adapters.Transformation `json:"transforms" yaml:"transforms"`
}

CDCConfig holds PostgreSQL CDC configuration

type CDCFactory

type CDCFactory struct{}

Factory for PostgreSQL CDC sources

func (*CDCFactory) Create

func (f *CDCFactory) Create(config adapters.SourceConfig) (adapters.FactSource, error)

func (*CDCFactory) GetConfigSchema

func (f *CDCFactory) GetConfigSchema() adapters.ConfigSchema

func (*CDCFactory) ValidateConfig

func (f *CDCFactory) ValidateConfig(config adapters.SourceConfig) error

type CDCSource

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

CDCSource implements Change Data Capture for PostgreSQL

func NewCDCSource

func NewCDCSource(config *CDCConfig) (*CDCSource, error)

NewCDCSource creates a new PostgreSQL CDC source

func (*CDCSource) GetMetadata

func (c *CDCSource) GetMetadata() adapters.SourceMetadata

func (*CDCSource) GetSourceSchema

func (c *CDCSource) GetSourceSchema() *adapters.Schema

func (*CDCSource) HealthCheck

func (c *CDCSource) HealthCheck() error

func (*CDCSource) Start

func (c *CDCSource) Start(ctx context.Context) error

func (*CDCSource) Stop

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

func (*CDCSource) Subscribe

func (c *CDCSource) Subscribe(ctx context.Context, factTypes []string) (<-chan *adapters.TypedFact, error)

type ChangeEvent

type ChangeEvent struct {
	Operation string                 `json:"operation"`
	Schema    string                 `json:"schema"`
	Table     string                 `json:"table"`
	Before    map[string]interface{} `json:"before,omitempty"`
	After     map[string]interface{} `json:"after,omitempty"`
	LSN       string                 `json:"lsn"`
	Timestamp time.Time              `json:"timestamp"`
	TxID      uint32                 `json:"tx_id"`
}

ChangeEvent represents a database change event

type ChangeTransformer

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

ChangeTransformer transforms database changes to TypedFacts

func NewChangeTransformer

func NewChangeTransformer(config *CDCConfig) *ChangeTransformer

func (*ChangeTransformer) TransformChange

func (t *ChangeTransformer) TransformChange(change *ChangeEvent) (*adapters.TypedFact, error)

type PollerConfig

type PollerConfig struct {
	ConnectionString string      `json:"connection_string" yaml:"connection_string"`
	Query            string      `json:"query" yaml:"query"`
	IntervalSeconds  int         `json:"interval_seconds" yaml:"interval_seconds"`
	TimestampColumn  string      `json:"timestamp_column" yaml:"timestamp_column"`
	TieBreakColumn   string      `json:"tie_break_column" yaml:"tie_break_column"`
	ProcessedLedger  string      `json:"processed_ledger_table" yaml:"processed_ledger_table"`
	StartTimestamp   string      `json:"start_timestamp" yaml:"start_timestamp"`
	StartTieBreak    interface{} `json:"start_tie_break" yaml:"start_tie_break"`
	SchemaName       string      `json:"schema_name" yaml:"schema_name"`
	MaxRows          int         `json:"max_rows" yaml:"max_rows"`
}

PollerConfig holds configuration for the PostgreSQL poller.

type PostgresPollerFactory

type PostgresPollerFactory struct{}

PostgresPollerFactory creates PostgreSQL poller sources.

func (*PostgresPollerFactory) Create

func (f *PostgresPollerFactory) Create(config adapters.SourceConfig) (adapters.FactSource, error)

func (*PostgresPollerFactory) GetConfigSchema

func (f *PostgresPollerFactory) GetConfigSchema() adapters.ConfigSchema

func (*PostgresPollerFactory) ValidateConfig

func (f *PostgresPollerFactory) ValidateConfig(config adapters.SourceConfig) error

type PostgresPollerSource

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

PostgresPollerSource polls PostgreSQL database at regular intervals.

func NewPostgresPollerSource

func NewPostgresPollerSource(sourceID string, config PollerConfig) (*PostgresPollerSource, error)

NewPostgresPollerSource creates a new PostgreSQL poller source.

func (*PostgresPollerSource) GetMetadata

func (p *PostgresPollerSource) GetMetadata() adapters.SourceMetadata

func (*PostgresPollerSource) GetSourceSchema

func (p *PostgresPollerSource) GetSourceSchema() *adapters.Schema

func (*PostgresPollerSource) HealthCheck

func (p *PostgresPollerSource) HealthCheck() error

func (*PostgresPollerSource) Start

func (p *PostgresPollerSource) Start(ctx context.Context) error

func (*PostgresPollerSource) Stop

func (p *PostgresPollerSource) Stop(ctx context.Context) error

func (*PostgresPollerSource) Subscribe

func (p *PostgresPollerSource) Subscribe(ctx context.Context, factTypes []string) (<-chan *adapters.TypedFact, error)