v0.0.41

dataext/bufferedReadCloser.go

@@ -1,56 +1,157 @@
 package dataext
 
-import "io"
+import (
+	"errors"
+	"io"
+)
+
+type brcMode int
+
+const (
+	modeSourceReading  = 0
+	modeSourceFinished = 1
+	modeBufferReading  = 2
+	modeBufferFinished = 3
+)
 
 type BufferedReadCloser interface {
 	io.ReadCloser
 	BufferedAll() ([]byte, error)
+	Reset() error
 }
 
 type bufferedReadCloser struct {
-	buffer   []byte
+	buffer []byte
-	inner    io.ReadCloser
+	inner  io.ReadCloser
-	finished bool
+	mode   brcMode
-}
+	off    int
-
-func (b *bufferedReadCloser) Read(p []byte) (int, error) {
-
-	n, err := b.inner.Read(p)
-	if n > 0 {
-		b.buffer = append(b.buffer, p[0:n]...)
-	}
-
-	if err == io.EOF {
-		b.finished = true
-	}
-
-	return n, err
 }
 
 func NewBufferedReadCloser(sub io.ReadCloser) BufferedReadCloser {
 	return &bufferedReadCloser{
-		buffer:   make([]byte, 0, 1024),
+		buffer: make([]byte, 0, 1024),
-		inner:    sub,
+		inner:  sub,
-		finished: false,
+		mode:   modeSourceReading,
+		off:    0,
+	}
+}
+
+func (b *bufferedReadCloser) Read(p []byte) (int, error) {
+	switch b.mode {
+	case modeSourceReading:
+		n, err := b.inner.Read(p)
+		if n > 0 {
+			b.buffer = append(b.buffer, p[0:n]...)
+		}
+
+		if err == io.EOF {
+			b.mode = modeSourceFinished
+		}
+
+		return n, err
+
+	case modeSourceFinished:
+		return 0, io.EOF
+
+	case modeBufferReading:
+
+		if len(b.buffer) <= b.off {
+			b.mode = modeBufferFinished
+			if len(p) == 0 {
+				return 0, nil
+			}
+			return 0, io.EOF
+		}
+
+		n := copy(p, b.buffer[b.off:])
+		b.off += n
+		return n, nil
+
+	case modeBufferFinished:
+		return 0, io.EOF
+
+	default:
+		return 0, errors.New("object in undefined status")
 	}
 }
 
 func (b *bufferedReadCloser) Close() error {
-	err := b.inner.Close()
+	switch b.mode {
-	if err != nil {
+	case modeSourceReading:
-		b.finished = true
+		_, err := b.BufferedAll()
+		if err != nil {
+			return err
+		}
+		err = b.inner.Close()
+		if err != nil {
+			return err
+		}
+		b.mode = modeSourceFinished
+		return nil
+
+	case modeSourceFinished:
+		return nil
+
+	case modeBufferReading:
+		b.mode = modeBufferFinished
+		return nil
+
+	case modeBufferFinished:
+		return nil
+
+	default:
+		return errors.New("object in undefined status")
 	}
-	return err
+
 }
 
 func (b *bufferedReadCloser) BufferedAll() ([]byte, error) {
-	arr := make([]byte, 1024)
+	switch b.mode {
-	for !b.finished {
+	case modeSourceReading:
-		_, err := b.Read(arr)
+		arr := make([]byte, 1024)
-		if err != nil && err != io.EOF {
+		for b.mode == modeSourceReading {
-			return nil, err
+			_, err := b.Read(arr)
+			if err != nil && err != io.EOF {
+				return nil, err
+			}
 		}
-	}
+		return b.buffer, nil
 
-	return b.buffer, nil
+	case modeSourceFinished:
+		return b.buffer, nil
+
+	case modeBufferReading:
+		return b.buffer, nil
+
+	case modeBufferFinished:
+		return b.buffer, nil
+
+	default:
+		return nil, errors.New("object in undefined status")
+	}
+}
+
+func (b *bufferedReadCloser) Reset() error {
+	switch b.mode {
+	case modeSourceReading:
+		fallthrough
+	case modeSourceFinished:
+		err := b.Close()
+		if err != nil {
+			return err
+		}
+		b.mode = modeBufferReading
+		b.off = 0
+		return nil
+
+	case modeBufferReading:
+		fallthrough
+	case modeBufferFinished:
+		b.mode = modeBufferReading
+		b.off = 0
+		return nil
+
+	default:
+		return errors.New("object in undefined status")
+	}
 }

langext/sort.go

@@ -0,0 +1,39 @@
+package langext
+
+import "sort"
+
+func Sort[T OrderedConstraint](arr []T) {
+	sort.Slice(arr, func(i1, i2 int) bool {
+		return arr[i1] < arr[i2]
+	})
+}
+
+func SortStable[T OrderedConstraint](arr []T) {
+	sort.SliceStable(arr, func(i1, i2 int) bool {
+		return arr[i1] < arr[i2]
+	})
+}
+
+func IsSorted[T OrderedConstraint](arr []T) bool {
+	return sort.SliceIsSorted(arr, func(i1, i2 int) bool {
+		return arr[i1] < arr[i2]
+	})
+}
+
+func SortSlice[T any](arr []T, less func(v1, v2 T) bool) {
+	sort.Slice(arr, func(i1, i2 int) bool {
+		return less(arr[i1], arr[i2])
+	})
+}
+
+func SortSliceStable[T any](arr []T, less func(v1, v2 T) bool) {
+	sort.SliceStable(arr, func(i1, i2 int) bool {
+		return less(arr[i1], arr[i2])
+	})
+}
+
+func IsSliceSorted[T any](arr []T, less func(v1, v2 T) bool) bool {
+	return sort.SliceIsSorted(arr, func(i1, i2 int) bool {
+		return less(arr[i1], arr[i2])
+	})
+}

sq/database.go

@@ -12,14 +12,14 @@ type DB interface {
 	Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows, error)
 	Ping(ctx context.Context) error
 	BeginTransaction(ctx context.Context, iso sql.IsolationLevel) (Tx, error)
-	SetListener(listener Listener)
+	AddListener(listener Listener)
 }
 
 type database struct {
 	db    *sqlx.DB
 	txctr uint16
 	lock  sync.Mutex
-	lstr  Listener
+	lstr  []Listener
 }
 
 func NewDB(db *sqlx.DB) DB {
@@ -27,31 +27,50 @@ func NewDB(db *sqlx.DB) DB {
 		db:    db,
 		txctr: 0,
 		lock:  sync.Mutex{},
+		lstr:  make([]Listener, 0),
 	}
 }
 
-func (db *database) SetListener(listener Listener) {
+func (db *database) AddListener(listener Listener) {
-	db.lstr = listener
+	db.lstr = append(db.lstr, listener)
 }
 
-func (db *database) Exec(ctx context.Context, sql string, prep PP) (sql.Result, error) {
+func (db *database) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Result, error) {
-	if db.lstr != nil {
+	origsql := sqlstr
-		db.lstr.OnExec(nil, sql, &prep)
+	for _, v := range db.lstr {
+		err := v.PreExec(ctx, nil, &sqlstr, &prep)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	res, err := db.db.NamedExecContext(ctx, sqlstr, prep)
+
+	for _, v := range db.lstr {
+		v.PostExec(nil, origsql, sqlstr, prep)
 	}
 
-	res, err := db.db.NamedExecContext(ctx, sql, prep)
 	if err != nil {
 		return nil, err
 	}
 	return res, nil
 }
 
-func (db *database) Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows, error) {
+func (db *database) Query(ctx context.Context, sqlstr string, prep PP) (*sqlx.Rows, error) {
-	if db.lstr != nil {
+	origsql := sqlstr
-		db.lstr.OnQuery(nil, sql, &prep)
+	for _, v := range db.lstr {
+		err := v.PreQuery(ctx, nil, &sqlstr, &prep)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	rows, err := sqlx.NamedQueryContext(ctx, db.db, sqlstr, prep)
+
+	for _, v := range db.lstr {
+		v.PostQuery(nil, origsql, sqlstr, prep)
 	}
 
-	rows, err := db.db.NamedQueryContext(ctx, sql, prep)
 	if err != nil {
 		return nil, err
 	}
@@ -59,11 +78,19 @@ func (db *database) Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows,
 }
 
 func (db *database) Ping(ctx context.Context) error {
-	if db.lstr != nil {
+	for _, v := range db.lstr {
-		db.lstr.OnPing()
+		err := v.PrePing(ctx)
+		if err != nil {
+			return err
+		}
 	}
 
 	err := db.db.PingContext(ctx)
+
+	for _, v := range db.lstr {
+		v.PostPing(err)
+	}
+
 	if err != nil {
 		return err
 	}
@@ -76,8 +103,11 @@ func (db *database) BeginTransaction(ctx context.Context, iso sql.IsolationLevel
 	db.txctr += 1 // with overflow !
 	db.lock.Unlock()
 
-	if db.lstr != nil {
+	for _, v := range db.lstr {
-		db.lstr.OnTxBegin(txid)
+		err := v.PreTxBegin(ctx, txid)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	xtx, err := db.db.BeginTxx(ctx, &sql.TxOptions{Isolation: iso})
@@ -85,5 +115,9 @@ func (db *database) BeginTransaction(ctx context.Context, iso sql.IsolationLevel
 		return nil, err
 	}
 
+	for _, v := range db.lstr {
+		v.PostTxBegin(txid, err)
+	}
+
 	return NewTransaction(xtx, txid, db.lstr), nil
 }

sq/listener.go

@@ -1,10 +1,19 @@
 package sq
 
+import "context"
+
 type Listener interface {
-	OnQuery(txID *uint16, sql string, params *PP)
+	PrePing(ctx context.Context) error
-	OnExec(txID *uint16, sql string, params *PP)
+	PreTxBegin(ctx context.Context, txid uint16) error
-	OnPing()
+	PreTxCommit(txid uint16) error
-	OnTxBegin(txid uint16)
+	PreTxRollback(txid uint16) error
-	OnTxCommit(txid uint16)
+	PreQuery(ctx context.Context, txID *uint16, sql *string, params *PP) error
-	OnTxRollback(txid uint16)
+	PreExec(ctx context.Context, txID *uint16, sql *string, params *PP) error
+
+	PostPing(result error)
+	PostTxBegin(txid uint16, result error)
+	PostTxCommit(txid uint16, result error)
+	PostTxRollback(txid uint16, result error)
+	PostQuery(txID *uint16, sqlOriginal string, sqlReal string, params PP)
+	PostExec(txID *uint16, sqlOriginal string, sqlReal string, params PP)
 }

sq/transaction.go

@@ -17,10 +17,10 @@ type Tx interface {
 type transaction struct {
 	tx   *sqlx.Tx
 	id   uint16
-	lstr Listener
+	lstr []Listener
 }
 
-func NewTransaction(xtx *sqlx.Tx, txid uint16, lstr Listener) Tx {
+func NewTransaction(xtx *sqlx.Tx, txid uint16, lstr []Listener) Tx {
 	return &transaction{
 		tx:   xtx,
 		id:   txid,
@@ -29,39 +29,75 @@ func NewTransaction(xtx *sqlx.Tx, txid uint16, lstr Listener) Tx {
 }
 
 func (tx *transaction) Rollback() error {
-	if tx.lstr != nil {
+	for _, v := range tx.lstr {
-		tx.lstr.OnTxRollback(tx.id)
+		err := v.PreTxRollback(tx.id)
+		if err != nil {
+			return err
+		}
 	}
 
-	return tx.tx.Rollback()
+	result := tx.tx.Rollback()
+
+	for _, v := range tx.lstr {
+		v.PostTxRollback(tx.id, result)
+	}
+
+	return result
 }
 
 func (tx *transaction) Commit() error {
-	if tx.lstr != nil {
+	for _, v := range tx.lstr {
-		tx.lstr.OnTxCommit(tx.id)
+		err := v.PreTxCommit(tx.id)
+		if err != nil {
+			return err
+		}
 	}
 
-	return tx.tx.Commit()
+	result := tx.tx.Commit()
+
+	for _, v := range tx.lstr {
+		v.PostTxRollback(tx.id, result)
+	}
+
+	return result
 }
 
-func (tx *transaction) Exec(ctx context.Context, sql string, prep PP) (sql.Result, error) {
+func (tx *transaction) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Result, error) {
-	if tx.lstr != nil {
+	origsql := sqlstr
-		tx.lstr.OnExec(langext.Ptr(tx.id), sql, &prep)
+	for _, v := range tx.lstr {
+		err := v.PreExec(ctx, langext.Ptr(tx.id), &sqlstr, &prep)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	res, err := tx.tx.NamedExecContext(ctx, sqlstr, prep)
+
+	for _, v := range tx.lstr {
+		v.PostExec(langext.Ptr(tx.id), origsql, sqlstr, prep)
 	}
 
-	res, err := tx.tx.NamedExecContext(ctx, sql, prep)
 	if err != nil {
 		return nil, err
 	}
 	return res, nil
 }
 
-func (tx *transaction) Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows, error) {
+func (tx *transaction) Query(ctx context.Context, sqlstr string, prep PP) (*sqlx.Rows, error) {
-	if tx.lstr != nil {
+	origsql := sqlstr
-		tx.lstr.OnQuery(langext.Ptr(tx.id), sql, &prep)
+	for _, v := range tx.lstr {
+		err := v.PreQuery(ctx, langext.Ptr(tx.id), &sqlstr, &prep)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	rows, err := sqlx.NamedQueryContext(ctx, tx.tx, sqlstr, prep)
+
+	for _, v := range tx.lstr {
+		v.PostQuery(langext.Ptr(tx.id), origsql, sqlstr, prep)
 	}
 
-	rows, err := sqlx.NamedQueryContext(ctx, tx.tx, sql, prep)
 	if err != nil {
 		return nil, err
 	}