listener

Documentation

Index

• Constants
• func NewRepository(conn *pgx.Conn) *repositoryImpl
• type ActionData
• type ActionKind
• type Begin
• type BinaryParser
  • func NewBinaryParser(byteOrder binary.ByteOrder) *BinaryParser
  • func (p *BinaryParser) ParseWalMessage(msg []byte, tx *WalTransaction) error
• type Column
  • func (c *Column) AssertValue(src []byte)
• type Commit
• type DataType
• type Delete
• type Event
  • func (e Event) GetSubjectName(prefix string) string
  • func (v Event) MarshalEasyJSON(w *jwriter.Writer)
  • func (v Event) MarshalJSON() ([]byte, error)
  • func (v *Event) UnmarshalEasyJSON(l *jlexer.Lexer)
  • func (v *Event) UnmarshalJSON(data []byte) error
• type Insert
• type Listener
  • func NewWalListener(cfg *config.Config, repo repository, repl replication, publ publisher, ...) *Listener
  • func (l *Listener) AckWalMessage(lsn uint64) error
  • func (l *Listener) Process() error
  • func (l *Listener) SendPeriodicHeartbeats(ctx context.Context)
  • func (l *Listener) SendStandbyStatus() error
  • func (l *Listener) Stop() error
  • func (l *Listener) Stream(ctx context.Context)
• type NatsPublisher
  • func NewNatsPublisher(conn stan.Conn) *NatsPublisher
  • func (n NatsPublisher) Close() error
  • func (n NatsPublisher) Publish(subject string, event Event) error
• type Origin
• type Relation
• type RelationColumn
• type RelationData
• type TupleData
• type Update
• type WalTransaction
  • func NewWalTransaction() *WalTransaction
  • func (w *WalTransaction) Clear()
  • func (w WalTransaction) CreateActionData(relationID int32, rows []TupleData, kind ActionKind) (a ActionData, err error)
  • func (w *WalTransaction) CreateEventsWithFilter(tableMap map[string][]string) []Event

Constants

const (
	ErrPostgresConnection    = "db connection error"
	ErrReplicationConnection = "replication connection error"
	ErrNatsConnection        = "nats connection error"
	ErrPublishEvent          = "publish message error"
	ErrUnmarshalMsg          = "unmarshal wal message error"
	ErrAckWalMessage         = "acknowledge wal message error"
	ErrSendStandbyStatus     = "send standby status error"
)

Constants with error text message

const (
	StartServiceMessage = "service was started"
	StopServiceMessage  = "service was stopped"
)

Service info message.

const (
	CommitMsgType   byte = 'C'
	BeginMsgType    byte = 'B'
	OriginMsgType   byte = 'O'
	RelationMsgType byte = 'R'
	TypeMsgType     byte = 'Y'
	InsertMsgType   byte = 'I'
	UpdateMsgType   byte = 'U'
	DeleteMsgType   byte = 'D'

	NewTupleDataType byte = 'N'
	TextDataType     byte = 't'
	NullDataType     byte = 'n'
	ToastDataType    byte = 'u'
)

Functions

func NewRepository

func NewRepository(conn *pgx.Conn) *repositoryImpl

Types

type ActionData

type ActionData struct {
	Schema  string
	Table   string
	Kind    ActionKind
	Columns []Column
}

type ActionKind

type ActionKind string

const (
	ActionKindInsert ActionKind = "INSERT"
	ActionKindUpdate ActionKind = "UPDATE"
	ActionKindDelete ActionKind = "DELETE"
)

kind of wall message.

type Begin

type Begin struct {
	// Identifies the message as a begin message.
	LSN int64
	// Commit timestamp of the transaction.
	Timestamp time.Time
	// 	Xid of the transaction.
	XID int32
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type BinaryParser

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

func NewBinaryParser

func NewBinaryParser(byteOrder binary.ByteOrder) *BinaryParser

func (*BinaryParser) ParseWalMessage

func (p *BinaryParser) ParseWalMessage(msg []byte, tx *WalTransaction) error

type Column

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

func (*Column) AssertValue

func (c *Column) AssertValue(src []byte)

type Commit

type Commit struct {
	// Flags; currently unused (must be 0).
	Flags int8
	// The LSN of the commit.
	LSN int64
	// The end LSN of the transaction.
	TransactionLSN int64
	// Commit timestamp of the transaction.
	Timestamp time.Time
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type DataType

type DataType struct {
	// ID of the data type.
	ID int32
	// Namespace (empty string for pg_catalog).
	Namespace string
	// name of the data type.
	Name string
}

type Delete

type Delete struct {
	/// ID of the relation corresponding to the ID in the relation message.
	RelationID int32
	// Identifies the following TupleData submessage as a key.
	KeyTuple bool
	// Identifies the following TupleData message as a old tuple.
	OldTuple bool
	// TupleData message part representing the contents of new tuple.
	Row []TupleData
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type Event

type Event struct {
	ID        uuid.UUID              `json:"id"`
	Schema    string                 `json:"schema"`
	Table     string                 `json:"table"`
	Action    string                 `json:"action"`
	Data      map[string]interface{} `json:"data"`
	EventTime time.Time              `json:"commitTime"`
}

Event event structure for publishing to the NATS server.

func (Event) GetSubjectName

func (e Event) GetSubjectName(prefix string) string

GetSubjectName creates subject name from the prefix, schema and table name.

func (Event) MarshalEasyJSON

func (v Event) MarshalEasyJSON(w *jwriter.Writer)

MarshalEasyJSON supports easyjson.Marshaler interface

func (Event) MarshalJSON

func (v Event) MarshalJSON() ([]byte, error)

MarshalJSON supports json.Marshaler interface

func (*Event) UnmarshalEasyJSON

func (v *Event) UnmarshalEasyJSON(l *jlexer.Lexer)

UnmarshalEasyJSON supports easyjson.Unmarshaler interface

func (*Event) UnmarshalJSON

func (v *Event) UnmarshalJSON(data []byte) error

UnmarshalJSON supports json.Unmarshaler interface

type Insert

type Insert struct {
	/// ID of the relation corresponding to the ID in the relation message.
	RelationID int32
	// Identifies the following TupleData message as a new tuple.
	NewTuple bool
	// TupleData message part representing the contents of new tuple.
	Row []TupleData
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type Listener

type Listener struct {
	LSN uint64
	// contains filtered or unexported fields
}

Listener main service struct.

func NewWalListener

func NewWalListener(
	cfg *config.Config,
	repo repository,
	repl replication,
	publ publisher,
	parser parser,
) *Listener

func (*Listener) AckWalMessage

func (l *Listener) AckWalMessage(lsn uint64) error

AckWalMessage acknowledge received wal message.

func (*Listener) Process

func (l *Listener) Process() error

Process is main service entry point.

func (*Listener) SendPeriodicHeartbeats

func (l *Listener) SendPeriodicHeartbeats(ctx context.Context)

SendPeriodicHeartbeats send periodic keep alive hearbeats to the server.

func (*Listener) SendStandbyStatus

func (l *Listener) SendStandbyStatus() error

SendStandbyStatus sends a `StandbyStatus` object with the current RestartLSN value to the server.

func (*Listener) Stop

func (l *Listener) Stop() error

Stop is a finalizer function.

func (*Listener) Stream

func (l *Listener) Stream(ctx context.Context)

Stream receive event from PostgreSQL. Accept message, apply filter and publish it in NATS server.

type NatsPublisher

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

func NewNatsPublisher

func NewNatsPublisher(conn stan.Conn) *NatsPublisher

func (NatsPublisher) Close

func (n NatsPublisher) Close() error

func (NatsPublisher) Publish

func (n NatsPublisher) Publish(subject string, event Event) error

type Origin

type Origin struct {
	// The LSN of the commit on the origin server.
	LSN int64
	// name of the origin.
	Name string
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type Relation

type Relation struct {
	// ID of the relation.
	ID int32
	// Namespace (empty string for pg_catalog).
	Namespace string
	// Relation name.
	Name string
	// Replica identity setting for the relation (same as relreplident in pg_class).
	Replica int8
	Columns []RelationColumn
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type RelationColumn

type RelationColumn struct {
	// Flags for the column which marks the column as part of the key.
	Key bool
	// name of the column.
	Name string
	// ID of the column's data type.
	TypeID int32
	// valueType modifier of the column (atttypmod).
	ModifierType int32
}

type RelationData

type RelationData struct {
	Schema  string
	Table   string
	Columns []Column
}

type TupleData

type TupleData struct {
	Value []byte
}

type Update

type Update struct {
	/// ID of the relation corresponding to the ID in the relation message.
	RelationID int32
	// Identifies the following TupleData submessage as a key.
	KeyTuple bool
	// Identifies the following TupleData message as a old tuple.
	OldTuple bool
	// Identifies the following TupleData message as a new tuple.
	NewTuple bool
	// TupleData message part representing the contents of new tuple.
	Row []TupleData
	// TupleData message part representing the contents of the old tuple or primary key.
	//Only present if the previous 'O' or 'K' part is present.
	OldRow []TupleData
}

Logical Replication Message Formats. https://postgrespro.ru/docs/postgrespro/10/protocol-logicalrep-message-formats#

type WalTransaction

type WalTransaction struct {
	LSN           int64
	BeginTime     *time.Time
	CommitTime    *time.Time
	RelationStore map[int32]RelationData
	Actions       []ActionData
}

func NewWalTransaction

func NewWalTransaction() *WalTransaction

func (*WalTransaction) Clear

func (w *WalTransaction) Clear()

func (WalTransaction) CreateActionData

func (w WalTransaction) CreateActionData(
	relationID int32,
	rows []TupleData,
	kind ActionKind,
) (a ActionData, err error)

func (*WalTransaction) CreateEventsWithFilter

func (w *WalTransaction) CreateEventsWithFilter(
	tableMap map[string][]string) []Event

CreateEventsWithFilter filter wal message by table, action and create events for each value.