pg

Documentation

Index

• func CreatePublicationIfNotExists(ctx context.Context, conn *pgconn.PgConn, pubName string, pubFor string, ...) error
• func CreateReplicationSlotIfNotExists(ctx context.Context, conn *pgconn.PgConn, slotName string) (pglogrepl.CreateReplicationSlotResult, error)
• func DropPublicationIfExists(ctx context.Context, conn *pgconn.PgConn, pubName string) error
• func DropReplicationSlotIfExists(ctx context.Context, conn *pgconn.PgConn, pubName string) error
• func StartReplication(ctx context.Context, conn *pgconn.PgConn, slotName string, lsn string, ...) error
• type FnProcessData
• type WALBlob
• type WALRow
  • func (row WALRow) ColumnByNameToString(name string) string
  • func (row WALRow) ColumnByPosToString(j int) string
  • func (row WALRow) GetID() string
  • func (row WALRow) GetOldID() string
  • func (row *WALRow) SetValuesFromResultReader(reader *pgconn.ResultReader)
  • func (row WALRow) ToJSON() string
• type WALRowOldKey
• type WALRowPrimaryKey

Functions

func CreatePublicationIfNotExists

func CreatePublicationIfNotExists(ctx context.Context, conn *pgconn.PgConn, pubName string, pubFor string, pubWith string) error

CreatePublicationIfNotExists creates publication on database if not already exists

func CreateReplicationSlotIfNotExists

func CreateReplicationSlotIfNotExists(ctx context.Context, conn *pgconn.PgConn, slotName string) (pglogrepl.CreateReplicationSlotResult, error)

CreateReplicationSlotIfNotExists creates replication slot on database if not already exists

func DropPublicationIfExists

func DropPublicationIfExists(ctx context.Context, conn *pgconn.PgConn, pubName string) error

DropPublicationIfExists drops publication if it exists

func DropReplicationSlotIfExists

func DropReplicationSlotIfExists(ctx context.Context, conn *pgconn.PgConn, pubName string) error

DropReplicationSlotIfExists drops replication connection if it exists

func StartReplication

func StartReplication(ctx context.Context, conn *pgconn.PgConn, slotName string, lsn string, processor *[]*FnProcessData, config []string) error

StartReplication starts listening WAL events from database

Types

type FnProcessData

type FnProcessData func(string, *WALBlob)

FnProcessData function type for processing WAL entries

type WALBlob

type WALBlob struct {
	Change []WALRow `json:"change"`
}

WALBlob presents data change coming from database {"change":[{"kind":"delete","schema":"public","table":"customer","pk":{"pknames":["id"],"pktypes":["integer"]},"oldkeys":{"keynames":["id","name","created","enabled","country"],"keytypes":["integer","character varying","timestamp(0) without time zone","boolean","character varying(2)"],"keyvalues":[11685,"testi 1","2020-10-09 12:56:32",true,"FI"]}},{"kind":"delete","schema":"public","table":"customer","pk":{"pknames":["id"],"pktypes":["integer"]},"oldkeys":{"keynames":["id","name","created","enabled","country"],"keytypes":["integer","character varying","timestamp(0) without time zone","boolean","character varying(2)"],"keyvalues":[11686,"testi 2","2020-10-10 08:19:19",true,"FI"]}}]} Example JSON:

{
	"change":
	[
		{
			"kind":"insert",
			"schema":"public",
			"table":"t",
			"columnnames":["id","name"],
			"columntypes":["integer","text"],
			"columnvalues":[1000,"cat"]
			"pk":
			{
				"pknames":["id"],
				"pktypes:["integer"]
			},
			"oldkeys":
			{
				"keynames":["id","name","created","enabled","country"],
				"keytypes":["integer","character varying","timestamp(0) without time zone","boolean","character varying(2)"],
				"keyvalues":[11685,"testi 1","2020-10-09 12:56:32",true,"FI"]
			}
		}
	]
}

type WALRow

type WALRow struct {
	Kind         string           `json:"kind"`
	Schema       string           `json:"schema"`
	Table        string           `json:"table"`
	ColumnNames  []string         `json:"columnnames"`
	ColumnTypes  []string         `json:"columntypes"`
	ColumnValues []interface{}    `json:"columnvalues"`
	PrimaryKey   WALRowPrimaryKey `json:"pk"`
	OldKey       WALRowOldKey     `json:"oldkeys"`
}

WALRow is invidual change coming from database

func (WALRow) ColumnByNameToString

func (row WALRow) ColumnByNameToString(name string) string

ColumnByNameToString returns column value by name as string

func (WALRow) ColumnByPosToString

func (row WALRow) ColumnByPosToString(j int) string

ColumnByPosToString returns column value by position as string

func (WALRow) GetID

func (row WALRow) GetID() string

GetID returns computed ID value for WALRow

func (WALRow) GetOldID

func (row WALRow) GetOldID() string

GetOldID returns computed previous ID value for WALRow

func (*WALRow) SetValuesFromResultReader

func (row *WALRow) SetValuesFromResultReader(reader *pgconn.ResultReader)

SetValuesFromResultReader overrides row values with values coming from result reader

func (WALRow) ToJSON

func (row WALRow) ToJSON() string

ToJSON returns normalized version of WALRow as JSON

type WALRowOldKey

type WALRowOldKey struct {
	KeyNames  []string      `json:"keynames"`
	KeyTypes  []string      `json:"keytypes"`
	KeyValues []interface{} `json:"keyvalues"`
}

WALRowOldKey has old primary key information about the row

type WALRowPrimaryKey

type WALRowPrimaryKey struct {
	ColumnNames []string `json:"pknames"`
	ColumnTypes []string `json:"pktypes"`
}

WALRowPrimaryKey has primary key information about the row