sql

Jan 18, 2023

Changes in this version

type Option

+ func WithQuery(query string, qOpts ...QueryOption) Option

type Query

+ func NewQuery(query string, opts ...QueryOption) (*Query, error)

+ type QueryConfig struct

+ FieldObject any

+ IterateBy string

+ Target string

+ type QueryOption func(*QueryConfig)

+ func QWithIterateBy(iterateBy string) QueryOption

+ func QWithMessageTmpl(fields any) QueryOption

+ func QWithTarget(target string) QueryOption

Jan 17, 2023

Changes in this version

type Option

+ func WithLogger(logger *zap.Logger) Option

+ type Options struct

+ BlockSize int

+ Debug bool

+ FlushInterval time.Duration

+ Logger *zap.Logger

+ QueryBuilder *Query

type Query

+ func (q *Query) QueryString() string

type Value

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

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

Oct 15, 2019

Changes in this version

type Option

+ func WithQueryByPattern(pattern, target string, fields interface{}) Option

+ func WithQueryObject(query *Query) Option

+ func WithQueryRawFields(query string, fields interface{}) Option

Jul 29, 2019

Changes in this version

+ func New(connector Connector, conf *storage.StreamConfig, pattern string) (stream eventstream.Streamer, err error)

+ func NewStreamSQL(id string, connector Connector, query Query, options ...Option) (_ eventstream.Streamer, err error)

+ func NewStreamSQLByRaw(id string, connector Connector, query string, fields interface{}, ...) (eventstream.Streamer, error)

+ type Connector interface

+ Connection func() (*sql.DB, error)

+ type Option func(stream *StreamSQL)

+ func WithBlockSize(size int) Option

+ func WithDebug(debug bool) Option

+ func WithFlushIntervals(interval time.Duration) Option

+ type Query struct

+ Q string

+ Values []Value

+ func NewQueryByPattern(pattern, target string, fl interface{}) (_ *Query, err error)

+ func NewQueryByRaw(query string, fl interface{}) (q *Query, err error)

+ func (q *Query) Extract(msg eventstream.Message) map[string]interface{}

+ func (q *Query) ParamsBy(msg eventstream.Message) (params []interface{})

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

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

+ type StreamSQL struct

+ func (s *StreamSQL) Check(msg eventstream.Message) bool

+ func (s *StreamSQL) Close() error

+ func (s *StreamSQL) ID() string

+ func (s *StreamSQL) Put(msg eventstream.Message) error

+ func (s *StreamSQL) Run() error

+ type Value struct

+ Format string

+ Key string

+ Length int

+ TargetKey string

+ Type eventstream.FieldType

+ func PrepareFields(fls interface{}) (values []Value, fields, inserts []string, err error)

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

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