sql

Documentation

Index

• func New(connector Connector, pattern string, conf *stream.Config, options ...Option) (stream eventstream.Streamer, err error)
• func NewStreamSQL(id string, connector Connector, options ...Option) (eventstream.Streamer, error)
• type Connector
• type Option
  • func WithBlockSize(size int) Option
  • func WithDebug(debug bool) Option
  • func WithFlushIntervals(interval time.Duration) Option
  • func WithLogger(logger *zap.Logger) Option
  • func WithQueryByPattern(pattern, target string, fields any) Option
  • func WithQueryObject(query *Query) Option
  • func WithQueryRawFields(query string, fields any) Option
• type Options
• type Query
  • func NewQueryByPattern(pattern, target string, fieldDescription any) (_ *Query, err error)
  • func NewQueryByRaw(query string, fl any) (queryBuilder *Query, err error)
  • func (q *Query) Extract(msg eventstream.Message) map[string]any
  • func (q *Query) ParamsBy(msg eventstream.Message) (params []any)
  • func (q *Query) QueryString() string
  • func (q *Query) StringByMessage(msg eventstream.Message) string
  • func (q *Query) StringParamsBy(msg eventstream.Message) (params []string)
• type StreamSQL
  • func (s *StreamSQL) Check(ctx context.Context, msg eventstream.Message) bool
  • func (s *StreamSQL) Close() (err error)
  • func (s *StreamSQL) ID() string
  • func (s *StreamSQL) Put(ctx context.Context, msg eventstream.Message) error
  • func (s *StreamSQL) Run(ctx context.Context) error
• type Value
  • func ConvertObjectIntoQueryParams(object any) (values []Value, fields, inserts []string, err error)
  • func MapIntoQueryParams(object any) (values []Value, fieldNames, inserts []string, err error)
  • func PrepareFields(fls any) (values []Value, fields, inserts []string, err error)
  • func PrepareFieldsByArray(fields []string) (values []Value, fieldNames, inserts []string)
  • func PrepareFieldsByString(fls string) (values []Value, fields, inserts []string)

Functions

func New

func New(connector Connector, pattern string, conf *stream.Config, options ...Option) (stream eventstream.Streamer, err error)

New stream for SQL type integrations

func NewStreamSQL

func NewStreamSQL(id string, connector Connector, options ...Option) (eventstream.Streamer, error)

NewStreamSQL creates streamer object for SQL based stream integration

Types

type Connector

type Connector interface {
	Connection() (*sql.DB, error)
}

Connector to DB

type Option

type Option func(opts *Options) error

Option type of the stream

func WithBlockSize

func WithBlockSize(size int) Option

WithBlockSize setup the size of writing block

func WithDebug

func WithDebug(debug bool) Option

WithDebug setup the debug mode ON/OFF

func WithFlushIntervals

func WithFlushIntervals(interval time.Duration) Option

WithFlushIntervals setup interval betwin flashes

func WithLogger

func WithLogger(logger *zap.Logger) Option

WithLogger setup the logging object

func WithQueryByPattern

func WithQueryByPattern(pattern, target string, fields any) Option

WithQueryByPattern setup query object by query pattern

func WithQueryObject

func WithQueryObject(query *Query) Option

WithQueryObject setup query object

func WithQueryRawFields

func WithQueryRawFields(query string, fields any) Option

WithQueryRawFields setup query object by fields parameters

type Options

type Options struct {
	// Debug mode of the stream
	Debug bool

	// BlockSize of buffer to flushing data
	BlockSize int

	// FlushInterval between flushes
	FlushInterval time.Duration

	// QueryBuilder object of data prepare
	QueryBuilder *Query

	// Logger object of module debuging and logs
	Logger *zap.Logger
}

Options of the SQL stream

type Query

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

Query extractor

func NewQueryByPattern

func NewQueryByPattern(pattern, target string, fieldDescription any) (_ *Query, err error)

NewQueryByPattern returns query object

func NewQueryByRaw

func NewQueryByRaw(query string, fl any) (queryBuilder *Query, err error)

NewQueryByRaw returns query object from raw SQL query

func (*Query) Extract

func (q *Query) Extract(msg eventstream.Message) map[string]any

Extract message by special fields and types

func (*Query) ParamsBy

func (q *Query) ParamsBy(msg eventstream.Message) (params []any)

ParamsBy by message

func (*Query) QueryString

func (q *Query) QueryString() string

QueryString - returns the SQL query string

func (*Query) StringByMessage

func (q *Query) StringByMessage(msg eventstream.Message) string

StringByMessage prepare

func (*Query) StringParamsBy

func (q *Query) StringParamsBy(msg eventstream.Message) (params []string)

StringParamsBy by message

type StreamSQL

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

StreamSQL stream

func (*StreamSQL) Check

func (s *StreamSQL) Check(ctx context.Context, msg eventstream.Message) bool

Check message value

func (*StreamSQL) Close

func (s *StreamSQL) Close() (err error)

Close implementation

func (*StreamSQL) ID

func (s *StreamSQL) ID() string

ID returns unical stream identificator

func (*StreamSQL) Put

func (s *StreamSQL) Put(ctx context.Context, msg eventstream.Message) error

Put message to stream

func (*StreamSQL) Run

func (s *StreamSQL) Run(ctx context.Context) error

Run SQL writer daemon

type Value

type Value struct {
	Key       string // Field key in the object
	TargetKey string // Target Key in the database
	Type      eventstream.FieldType
	Length    int
	Format    string
}

Value item

func ConvertObjectIntoQueryParams

func ConvertObjectIntoQueryParams(object any) (values []Value, fields, inserts []string, err error)

ConvertObjectIntoQueryParams returns object which links object Fields and Data Fields

func MapIntoQueryParams

func MapIntoQueryParams(object any) (values []Value, fieldNames, inserts []string, err error)

MapIntoQueryParams returns object which links map fields

func PrepareFields

func PrepareFields(fls any) (values []Value, fields, inserts []string, err error)

PrepareFields matching

func PrepareFieldsByArray

func PrepareFieldsByArray(fields []string) (values []Value, fieldNames, inserts []string)

PrepareFieldsByArray matching and returns raw fields for insert Example: [service=srv:int, name:string] Result: [srv:int, name:string], [service,name], [?,?]

func PrepareFieldsByString

func PrepareFieldsByString(fls string) (values []Value, fields, inserts []string)

PrepareFieldsByString matching and returns raw fields for insert