sql

package

v1.1.0 Latest Latest Go to latest Published: Oct 17, 2018 License: MIT Imports: 4 Imported by: 0

Details

This section is empty.

This section is empty.

This section is empty.

type InsertFunc func(*sql.Tx, *streams.Message) error

InsertFunc represents a callback to handle processing a Message on the Sink.

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

Sink represents a SQL sink processor.

func NewSink(db *sql.DB, fn InsertFunc, opts ...SinkFunc) (*Sink, error)

NewSink creates a new batch sql insert sink.

func (p *Sink) Close() error

Close closes the processor.

func (p *Sink) Process(msg *streams.Message) error

Process processes the stream record.

func (p *Sink) WithPipe(pipe streams.Pipe)

WithPipe sets the pipe on the Processor.

type SinkFunc func(*Sink)

SinkFunc represents a function that configures the Sink.

func WithBatchFrequency(freq time.Duration) SinkFunc

WithBatchFrequency configures the frequency to send a batch on the Sink.

func WithBatchMessages(messages int) SinkFunc

WithBatchMessages configures the number of messages to send in a batch on the Sink.

func WithBeginFn(fn TxFunc) SinkFunc

WithBeginFn sets the transaction start callback on the Sink.

func WithCommitFn(fn TxFunc) SinkFunc

WithCommitFn sets the transaction commit callback on the Sink.

type TxFunc func(*sql.Tx) error

TxFunc represents a function that receives a sql transaction.