sql

Documentation

Index

• type ExecFunc
  • func (fn ExecFunc) Exec(tx *sql.Tx, msg streams.Message) error
• type Executor
• type Sink
  • func NewSink(db *sql.DB, batch int, exec Executor) (*Sink, error)
  • func (p *Sink) Close() error
  • func (p *Sink) Commit(ctx context.Context) error
  • func (p *Sink) Process(msg streams.Message) error
  • func (p *Sink) WithPipe(pipe streams.Pipe)
• type Transaction

Types

type ExecFunc

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

ExecFunc represents a function implementing an Executor.

func (ExecFunc) Exec

func (fn ExecFunc) Exec(tx *sql.Tx, msg streams.Message) error

Exec executes a query on the given transaction.

type Executor

type Executor interface {
	// Exec executes a query on the given transaction.
	Exec(*sql.Tx, streams.Message) error
}

Executor represents a SQL query executor.

type Sink

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

Sink represents a SQL sink processor.

func NewSink

func NewSink(db *sql.DB, batch int, exec Executor) (*Sink, error)

NewSink creates a new batch sql insert sink.

func (*Sink) Close

func (p *Sink) Close() error

Close closes the processor.

func (*Sink) Commit

func (p *Sink) Commit(ctx context.Context) error

Commit commits a processors batch.

func (*Sink) Process

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

Process processes the stream record.

func (*Sink) WithPipe

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

WithPipe sets the pipe on the Processor.

type Transaction

type Transaction interface {
	// Begin handles the start of a SQL transaction.
	Begin(*sql.Tx) error
	// Commit handles the commit of a SQL transaction.
	Commit(*sql.Tx) error
}

Transaction represents a SQL transaction handler.