postgres

README

PostgreSQL adaptor

The PostgreSQL adaptor is capable of reading/tailing tables using logical decoding and receiving data for inserts.

Configuration:

pg = postgres({
  "uri": "postgres://127.0.0.1:5432/test"
})

Permissions

Postgres as a transporter source uses Logical Decoding which requires the user account to have superuser or replication permissions.

Run adaptor test

Spin up required containers

You'll need those ports on your local machine: 5432

So make sure to kill anything that might use them (like a local postgres instance)

# From transporter's root folder
version=12
# Pay attention to a WARNING telling you to add a line to /etc/hosts in the following command
scripts/run_db_in_docker.sh postgres $version

Run the tests

# From transporter's root folder
go test -v ./adaptor/postgres/

Tear down containers

Once you're done

TESTDIR=adaptor/postgres scripts/teardown_db_in_docker.sh

Documentation

Index

• Constants
• type Client
  • func NewClient(options ...ClientOptionFunc) (*Client, error)
  • func (c *Client) Close()
  • func (c *Client) Connect() (client.Session, error)
• type ClientOptionFunc
  • func WithURI(uri string) ClientOptionFunc
• type Reader
  • func (r *Reader) Read(resumeMap map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc
• type Session
• type Tailer
  • func (t *Tailer) Read(resumeMap map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc
• type Writer
  • func (w *Writer) Write(msg message.Msg) func(client.Session) (message.Msg, error)

Constants

const (
	// DefaultURI is the default endpoint of Postgres on the local machine.
	// Primarily used when initializing a new Client without a specific URI.
	DefaultURI = "postgres://postgres@transporter-db:5432?sslmode=disable"
)

Types

type Client

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

Client represents a client to the underlying File source.

func NewClient

func NewClient(options ...ClientOptionFunc) (*Client, error)

NewClient creates a default file client

func (*Client) Close

func (c *Client) Close()

Close implements necessary calls to cleanup the underlying *sql.DB

func (*Client) Connect

func (c *Client) Connect() (client.Session, error)

Connect initializes the Postgres connection

type ClientOptionFunc

type ClientOptionFunc func(*Client) error

ClientOptionFunc is a function that configures a Client. It is used in NewClient.

func WithURI

func WithURI(uri string) ClientOptionFunc

WithURI defines the full connection string for the Postgres connection

type Reader

type Reader struct {
}

Reader implements the behavior defined by client.Reader for interfacing with MongoDB.

func (*Reader) Read

func (r *Reader) Read(resumeMap map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc

type Session

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

Session serves as a wrapper for the underlying *sql.DB

type Tailer

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

Tailer implements the behavior defined by client.Tailer for interfacing with the MongoDB oplog.

func (*Tailer) Read

func (t *Tailer) Read(resumeMap map[string]client.MessageSet, filterFn client.NsFilterFunc) client.MessageChanFunc

Tail does the things

type Writer

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

Writer implements client.Writer for use with MongoDB

func (*Writer) Write

func (w *Writer) Write(msg message.Msg) func(client.Session) (message.Msg, error)