rethinkdb

README

RethinkDB adaptor

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

Configuration:

- rethink:
    type: rethinkdb
    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 WithDatabase(db 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 Config
• type ConnectError
  • func (e ConnectError) Error() string
• type InvalidCertError
  • func (e InvalidCertError) Error() string
• type InvalidTimeoutError
  • func (e InvalidTimeoutError) Error() string
• type InvalidURIError
  • func (e InvalidURIError) Error() string
• type Reader
  • func (r *Reader) Read(filterFn client.NsFilterFunc) client.MessageChanFunc
• type RethinkDB
  • func (r *RethinkDB) Connect() error
  • func (r *RethinkDB) Description() string
  • func (r *RethinkDB) Listen() (err error)
  • func (r *RethinkDB) SampleConfig() string
  • func (r *RethinkDB) Start() error
  • func (r *RethinkDB) Stop() error
• type Session
• type VersionError
  • func (e VersionError) Error() string
• type Writer
  • func (w *Writer) Write(msg message.Msg) func(client.Session) 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

	// DefaultDatabase used for the connection options
	DefaultDatabase = "test"
)

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 WithDatabase

func WithDatabase(db string) ClientOptionFunc

WithDatabase configures the database to use for the 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 Config

type Config struct {
	URI       string   `json:"uri" doc:"the uri to connect to, in the form rethink://user:password@host.example:28015/database"`
	Namespace string   `json:"namespace" doc:"rethink namespace to read/write"`
	Timeout   string   `json:"timeout" doc:"timeout for establishing connection, format must be parsable by time.ParseDuration and defaults to 10s"`
	Tail      bool     `json:"tail" doc:"if true, the RethinkDB table will be monitored for changes after copying the namespace"`
	SSL       bool     `json:"ssl" doc:"enable TLS connection"`
	CACerts   []string `json:"cacerts" doc:"array of root CAs to use in order to verify the server certificates"`
}

Config provides custom configuration options for the RethinkDB adapter

type ConnectError

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

ConnectError wraps the underlying error when a failure occurs dialing the database.

func (ConnectError) Error

func (e ConnectError) Error() string

type InvalidCertError

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

InvalidCertError wraps the underlying error when the provided is certificate is not parsable.

func (InvalidCertError) Error

func (e InvalidCertError) Error() string

type InvalidTimeoutError

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

InvalidTimeoutError wraps the underlying error when the provided is not parsable time.ParseDuration.

func (InvalidTimeoutError) Error

func (e InvalidTimeoutError) Error() string

type InvalidURIError

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

InvalidURIError wraps the underlying error when the provided URI is not parsable by mgo.

func (InvalidURIError) Error

func (e InvalidURIError) Error() string

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 {
	// contains filtered or unexported fields
}

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

func (*RethinkDB) Connect

func (r *RethinkDB) Connect() error

Connect tests the connection and if successful, connects to the database

func (*RethinkDB) Description

func (r *RethinkDB) Description() string

Description for rethinkdb adaptor

func (*RethinkDB) Listen

func (r *RethinkDB) Listen() (err error)

Listen start's the adaptor's listener

func (*RethinkDB) SampleConfig

func (r *RethinkDB) SampleConfig() string

SampleConfig for rethinkdb adaptor

func (*RethinkDB) Start

func (r *RethinkDB) Start() error

Start the adaptor as a source

func (*RethinkDB) Stop

func (r *RethinkDB) Stop() error

Stop the adaptor

type Session

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

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

type VersionError

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

VersionError represents any failure in attempting to obtain the version from the provided uri.

func (VersionError) Error

func (e VersionError) Error() string

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) error