rethinkdb

README

RethinkDB adaptor

The RethinkDB adaptor is capable of reading/tailing tables and receiving data for inserts.

Configuration:

r = rethink({
  "uri": "rethink://127.0.0.1:28015/",
  // "timeout": "30s",
  // "tail": false,
  // "ssl": false,
  // "cacerts": ["/path/to/cert.pem"]
})

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 WithCACerts(certs []string) ClientOptionFunc
  • func WithReadTimeout(timeout string) ClientOptionFunc
  • func WithSSL(ssl bool) ClientOptionFunc
  • func WithSessionTimeout(timeout string) ClientOptionFunc
  • func WithURI(uri string) ClientOptionFunc
  • func WithWriteTimeout(timeout string) ClientOptionFunc
• type Reader
  • func (r *Reader) Read(filterFn client.NsFilterFunc) client.MessageChanFunc
• type RethinkDB
  • func (r *RethinkDB) Client() (client.Client, error)
  • func (r *RethinkDB) Description() string
  • func (r *RethinkDB) Reader() (client.Reader, error)
  • func (r *RethinkDB) SampleConfig() string
  • func (r *RethinkDB) Writer(done chan struct{}, wg *sync.WaitGroup) (client.Writer, error)
• type Session
• type Writer
  • func (w *Writer) Write(msg message.Msg) func(client.Session) (message.Msg, error)

Constants

const (
	// DefaultURI is the default endpoint for RethinkDB on the local machine.
	// Primarily used when initializing a new Client without a specific URI.
	DefaultURI = "rethinkdb://127.0.0.1:28015/test"

	// DefaultTimeout is the default time.Duration used if one is not provided for options
	// that pertain to timeouts.
	DefaultTimeout = 10 * time.Second
)

Types

type Client

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

Client creates and holds the session to RethinkDB

func NewClient

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

NewClient creates a new client to work with RethinkDB.

The caller can configure the new client by passing configuration options to the func.

Example:

client, err := NewClient(
  WithURI("rethinkdb://localhost:28015"),
  WithTimeout("30s"))

If no URI is configured, it uses DefaultURI.

An error is also returned when some configuration option is invalid

func (*Client) Close

func (c *Client) Close()

Close fulfills the Closer interface and takes care of cleaning up the rethink.Session

func (*Client) Connect

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

Connect wraps the underlying session to the RethinkDB database

type ClientOptionFunc

type ClientOptionFunc func(*Client) error

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

func WithCACerts

func WithCACerts(certs []string) ClientOptionFunc

WithCACerts configures the RootCAs for the underlying TLS connection

func WithReadTimeout

func WithReadTimeout(timeout string) ClientOptionFunc

WithReadTimeout overrides the DefaultTimeout and should be parseable by time.ParseDuration

func WithSSL

func WithSSL(ssl bool) ClientOptionFunc

WithSSL configures the database connection to connect via TLS.

func WithSessionTimeout

func WithSessionTimeout(timeout string) ClientOptionFunc

WithSessionTimeout overrides the DefaultTimeout and should be parseable by time.ParseDuration

func WithURI

func WithURI(uri string) ClientOptionFunc

WithURI defines the full connection string of the RethinkDB database.

func WithWriteTimeout

func WithWriteTimeout(timeout string) ClientOptionFunc

WithWriteTimeout overrides the DefaultTimeout and should be parseable by time.ParseDuration

type Reader

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

Reader fulfills the client.Reader interface for use with both copying and tailing a RethinkDB database.

func (*Reader) Read

func (r *Reader) Read(filterFn client.NsFilterFunc) client.MessageChanFunc

type RethinkDB

type RethinkDB struct {
	adaptor.BaseConfig
	Tail    bool     `json:"tail"`
	SSL     bool     `json:"ssl"`
	CACerts []string `json:"cacerts"`
}

RethinkDB is an adaptor that writes metrics to rethinkdb (http://rethinkdb.com/) An open-source distributed database

func (*RethinkDB) Client

func (r *RethinkDB) Client() (client.Client, error)

func (*RethinkDB) Description

func (r *RethinkDB) Description() string

Description for rethinkdb adaptor

func (*RethinkDB) Reader

func (r *RethinkDB) Reader() (client.Reader, error)

func (*RethinkDB) SampleConfig

func (r *RethinkDB) SampleConfig() string

SampleConfig for rethinkdb adaptor

func (*RethinkDB) Writer

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

type Session

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

Session contains an instance of the rethink.Session for use by Readers/Writers

type Writer

type Writer struct {
	*sync.Mutex
	// contains filtered or unexported fields
}

Writer implements client.Writer for use with RethinkDB

func (*Writer) Write

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