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"
})

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 Postgres
  • func (p *Postgres) Client() (client.Client, error)
  • func (p *Postgres) Description() string
  • func (p *Postgres) Reader() (client.Reader, error)
  • func (p *Postgres) SampleConfig() string
  • func (p *Postgres) Writer(done chan struct{}, wg *sync.WaitGroup) (client.Writer, error)
• type Reader
  • func (r *Reader) Read(filterFn client.NsFilterFunc) client.MessageChanFunc
• type Session
• type Tailer
  • func (t *Tailer) Read(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://127.0.0.1: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 Postgres

type Postgres struct {
	adaptor.BaseConfig
	Debug           bool   `json:"debug" doc:"display debug information"`
	Tail            bool   `json:"tail" doc:"if tail is true, then the postgres source will tail the oplog after copying the namespace"`
	ReplicationSlot string `json:"replication_slot" doc:"required if tail is true; sets the replication slot to use for logical decoding"`
}

Postgres is an adaptor to read / write to postgres. it works as a source by copying files, and then optionally tailing the oplog

func (*Postgres) Client

func (p *Postgres) Client() (client.Client, error)

func (*Postgres) Description

func (p *Postgres) Description() string

Description for postgres adaptor

func (*Postgres) Reader

func (p *Postgres) Reader() (client.Reader, error)

func (*Postgres) SampleConfig

func (p *Postgres) SampleConfig() string

SampleConfig for postgres adaptor

func (*Postgres) Writer

func (p *Postgres) Writer(done chan struct{}, wg *sync.WaitGroup) (client.Writer, error)

type Reader

type Reader struct {
}

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

func (*Reader) Read

func (r *Reader) Read(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(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)