sink

Documentation

Index

• type CSVSink
  • func NewCSVSink(w io.Writer) *CSVSink
  • func NewCSVSinkWithDelimiter(w io.Writer, delimiter rune) *CSVSink
  • func (s *CSVSink) Close(ctx context.Context) error
  • func (s *CSVSink) Flush(ctx context.Context) error
  • func (s *CSVSink) WriteRow(ctx context.Context, values ...interface{}) error
• type ConcurrentSink
  • func NewConcurrentSink(downStreamBuilder func(idx int) Sink, concurrency int) *ConcurrentSink
  • func (c *ConcurrentSink) Close(ctx context.Context) error
  • func (c *ConcurrentSink) Flush(ctx context.Context) error
  • func (c *ConcurrentSink) WriteRow(ctx context.Context, values ...interface{}) error
• type SQLSink
  • func NewSQLSink(db *sql.DB, hint string, retryCount int, retryInterval time.Duration) *SQLSink
  • func (s *SQLSink) Close(ctx context.Context) error
  • func (s *SQLSink) Flush(ctx context.Context) error
  • func (s *SQLSink) WriteRow(ctx context.Context, values ...interface{}) error
• type Sink

Types

type CSVSink

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

CSVSink writes values to a file in CSV format.

func NewCSVSink

func NewCSVSink(w io.Writer) *CSVSink

NewCSVSink creates a sink that writes values to an io.Writer in CSV format.

func NewCSVSinkWithDelimiter

func NewCSVSinkWithDelimiter(w io.Writer, delimiter rune) *CSVSink

NewCSVSinkWithDelimiter creates a sink that writes values to an io.Writer in CSV format, using a customized delimiter.

func (*CSVSink) Close

func (s *CSVSink) Close(ctx context.Context) error

Close closes the underlying io.Writer if it is an io.WriteCloser.

func (*CSVSink) Flush

func (s *CSVSink) Flush(ctx context.Context) error

Flush writes any buffered data to the underlying io.Writer.

func (*CSVSink) WriteRow

func (s *CSVSink) WriteRow(ctx context.Context, values ...interface{}) error

WriteRow writes a row to the underlying io.Writer. The writing attempt may be deferred until reaching a batch.

type ConcurrentSink

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

ConcurrentSink inserts values to one of the downstream sinks. The insert will be blocked if all downstream sinks are working.

WARN: Although this sink can transform serial Writes to multiple down stream sinks, this sink itself is not concurrent safe. You must not call WriteRow and Flush concurrently.

func NewConcurrentSink

func NewConcurrentSink(downStreamBuilder func(idx int) Sink, concurrency int) *ConcurrentSink

func (*ConcurrentSink) Close

func (c *ConcurrentSink) Close(ctx context.Context) error

Close closes all downstream sinks concurrently, wait all sinks to be closed and returns the first error encountered.

WARN: Close() will wait until all existing write ops are finished.

func (*ConcurrentSink) Flush

func (c *ConcurrentSink) Flush(ctx context.Context) error

Flush flushes all downstream sinks concurrently, wait all sinks to be flushed and returns the first error encountered.

WARN: Flush() will wait until all existing write ops are finished.

func (*ConcurrentSink) WriteRow

func (c *ConcurrentSink) WriteRow(ctx context.Context, values ...interface{}) error

type SQLSink

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

SQLSink inserts values to a database in batch.

func NewSQLSink

func NewSQLSink(db *sql.DB, hint string, retryCount int, retryInterval time.Duration) *SQLSink

NewSQLSink creates a sink that inserts values to a database in batch.

func (*SQLSink) Close

func (s *SQLSink) Close(ctx context.Context) error

func (*SQLSink) Flush

func (s *SQLSink) Flush(ctx context.Context) error

Flush writes any buffered data to the db.

func (*SQLSink) WriteRow

func (s *SQLSink) WriteRow(ctx context.Context, values ...interface{}) error

WriteRow writes a row to the database. The writing attempt may be deferred until reaching a batch.

type Sink

type Sink interface {
	WriteRow(ctx context.Context, values ...interface{}) error
	Flush(ctx context.Context) error
	Close(ctx context.Context) error
}