BlackForestBytes/goext
19
0
Fork 0
Code Actions Activity

v0.0.518 Improve sq db-listener interface (breaking)
All checks were successful
Build Docker and Deploy / Run goext test-suite (push) Successful in 4m11s
Details

Browse Source
This commit is contained in:
Mike Schwörer
2024-10-05 00:45:55 +02:00
parent 295b3ef793
commit f6b47792a4
6 changed files with 232 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
sq/database.go
View File

@@ -7,6 +7,7 @@ import (
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
"gogs.mikescher.com/BlackForestBytes/goext/langext"
"sync"
"time"
)
type DB interface {
@@ -57,86 +58,118 @@ func (db *database) AddListener(listener Listener) {
func (db *database) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Result, error) {
origsql := sqlstr
t0 := time.Now()
preMeta := PreExecMeta{}
for _, v := range db.lstr {
err := v.PreExec(ctx, nil, &sqlstr, &prep)
err := v.PreExec(ctx, nil, &sqlstr, &prep, preMeta)
if err != nil {
return nil, exerr.Wrap(err, "failed to call SQL pre-exec listener").Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
}
t1 := time.Now()
res, err := db.db.NamedExecContext(ctx, sqlstr, prep)
postMeta := PostExecMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range db.lstr {
v.PostExec(nil, origsql, sqlstr, prep)
v.PostExec(nil, origsql, sqlstr, prep, postMeta)
}
if err != nil {
return nil, exerr.Wrap(err, "Failed to [exec] sql statement").Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
return res, nil
}
func (db *database) Query(ctx context.Context, sqlstr string, prep PP) (*sqlx.Rows, error) {
origsql := sqlstr
t0 := time.Now()
preMeta := PreQueryMeta{}
for _, v := range db.lstr {
err := v.PreQuery(ctx, nil, &sqlstr, &prep)
err := v.PreQuery(ctx, nil, &sqlstr, &prep, preMeta)
if err != nil {
return nil, exerr.Wrap(err, "failed to call SQL pre-query listener").Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
}
t1 := time.Now()
rows, err := sqlx.NamedQueryContext(ctx, db.db, sqlstr, prep)
postMeta := PostQueryMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range db.lstr {
v.PostQuery(nil, origsql, sqlstr, prep)
v.PostQuery(nil, origsql, sqlstr, prep, postMeta)
}
if err != nil {
return nil, exerr.Wrap(err, "Failed to [query] sql statement").Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
return rows, nil
}
func (db *database) Ping(ctx context.Context) error {
t0 := time.Now()
preMeta := PrePingMeta{}
for _, v := range db.lstr {
err := v.PrePing(ctx)
err := v.PrePing(ctx, preMeta)
if err != nil {
return err
}
}
t1 := time.Now()
err := db.db.PingContext(ctx)
postMeta := PostPingMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range db.lstr {
v.PostPing(err)
v.PostPing(err, postMeta)
}
if err != nil {
return exerr.Wrap(err, "Failed to [ping] sql database").Build()
}
return nil
}
func (db *database) BeginTransaction(ctx context.Context, iso sql.IsolationLevel) (Tx, error) {
t0 := time.Now()
db.lock.Lock()
txid := db.txctr
db.txctr += 1 // with overflow !
db.lock.Unlock()
preMeta := PreTxBeginMeta{}
for _, v := range db.lstr {
err := v.PreTxBegin(ctx, txid)
err := v.PreTxBegin(ctx, txid, preMeta)
if err != nil {
return nil, err
}
}
t1 := time.Now()
xtx, err := db.db.BeginTxx(ctx, &sql.TxOptions{Isolation: iso})
if err != nil {
return nil, exerr.Wrap(err, "Failed to start sql transaction").Build()
postMeta := PostTxBeginMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range db.lstr {
v.PostTxBegin(txid, err, postMeta)
}
for _, v := range db.lstr {
v.PostTxBegin(txid, err)
if err != nil {
return nil, exerr.Wrap(err, "Failed to start sql transaction").Build()
}
return NewTransaction(xtx, txid, db), nil

191
sq/listener.go
View File

@@ -1,187 +1,248 @@
package sq
import "context"
import (
"context"
"time"
)
type PrePingMeta struct {
}
type PreTxBeginMeta struct {
}
type PreTxCommitMeta struct {
}
type PreTxRollbackMeta struct {
}
type PreQueryMeta struct {
}
type PreExecMeta struct {
}
type PostPingMeta struct {
Init time.Time
Start time.Time
End time.Time
}
type PostTxBeginMeta struct {
Init time.Time
Start time.Time
End time.Time
}
type PostTxCommitMeta struct {
Init time.Time
Start time.Time
End time.Time
ExecCounter int
QueryCounter int
}
type PostTxRollbackMeta struct {
Init time.Time
Start time.Time
End time.Time
ExecCounter int
QueryCounter int
}
type PostQueryMeta struct {
Init time.Time
Start time.Time
End time.Time
}
type PostExecMeta struct {
Init time.Time
Start time.Time
End time.Time
}
type Listener interface {
PrePing(ctx context.Context) error
PreTxBegin(ctx context.Context, txid uint16) error
PreTxCommit(txid uint16) error
PreTxRollback(txid uint16) error
PreQuery(ctx context.Context, txID *uint16, sql *string, params *PP) error
PreExec(ctx context.Context, txID *uint16, sql *string, params *PP) error
PrePing(ctx context.Context, meta PrePingMeta) error
PreTxBegin(ctx context.Context, txid uint16, meta PreTxBeginMeta) error
PreTxCommit(txid uint16, meta PreTxCommitMeta) error
PreTxRollback(txid uint16, meta PreTxRollbackMeta) error
PreQuery(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreQueryMeta) error
PreExec(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreExecMeta) 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)
PostPing(result error, meta PostPingMeta)
PostTxBegin(txid uint16, result error, meta PostTxBeginMeta)
PostTxCommit(txid uint16, result error, meta PostTxCommitMeta)
PostTxRollback(txid uint16, result error, meta PostTxRollbackMeta)
PostQuery(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostQueryMeta)
PostExec(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostExecMeta)
}
type genListener struct {
prePing func(ctx context.Context) error
preTxBegin func(ctx context.Context, txid uint16) error
preTxCommit func(txid uint16) error
preTxRollback func(txid uint16) error
preQuery func(ctx context.Context, txID *uint16, sql *string, params *PP) error
preExec func(ctx context.Context, txID *uint16, sql *string, params *PP) error
postPing func(result error)
postTxBegin func(txid uint16, result error)
postTxCommit func(txid uint16, result error)
postTxRollback func(txid uint16, result error)
postQuery func(txID *uint16, sqlOriginal string, sqlReal string, params PP)
postExec func(txID *uint16, sqlOriginal string, sqlReal string, params PP)
prePing func(ctx context.Context, meta PrePingMeta) error
preTxBegin func(ctx context.Context, txid uint16, meta PreTxBeginMeta) error
preTxCommit func(txid uint16, meta PreTxCommitMeta) error
preTxRollback func(txid uint16, meta PreTxRollbackMeta) error
preQuery func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreQueryMeta) error
preExec func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreExecMeta) error
postPing func(result error, meta PostPingMeta)
postTxBegin func(txid uint16, result error, meta PostTxBeginMeta)
postTxCommit func(txid uint16, result error, meta PostTxCommitMeta)
postTxRollback func(txid uint16, result error, meta PostTxRollbackMeta)
postQuery func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostQueryMeta)
postExec func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostExecMeta)
}
func (g genListener) PrePing(ctx context.Context) error {
func (g genListener) PrePing(ctx context.Context, meta PrePingMeta) error {
if g.prePing != nil {
return g.prePing(ctx)
return g.prePing(ctx, meta)
} else {
return nil
}
}
func (g genListener) PreTxBegin(ctx context.Context, txid uint16) error {
func (g genListener) PreTxBegin(ctx context.Context, txid uint16, meta PreTxBeginMeta) error {
if g.preTxBegin != nil {
return g.preTxBegin(ctx, txid)
return g.preTxBegin(ctx, txid, meta)
} else {
return nil
}
}
func (g genListener) PreTxCommit(txid uint16) error {
func (g genListener) PreTxCommit(txid uint16, meta PreTxCommitMeta) error {
if g.preTxCommit != nil {
return g.preTxCommit(txid)
return g.preTxCommit(txid, meta)
} else {
return nil
}
}
func (g genListener) PreTxRollback(txid uint16) error {
func (g genListener) PreTxRollback(txid uint16, meta PreTxRollbackMeta) error {
if g.preTxRollback != nil {
return g.preTxRollback(txid)
return g.preTxRollback(txid, meta)
} else {
return nil
}
}
func (g genListener) PreQuery(ctx context.Context, txID *uint16, sql *string, params *PP) error {
func (g genListener) PreQuery(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreQueryMeta) error {
if g.preQuery != nil {
return g.preQuery(ctx, txID, sql, params)
return g.preQuery(ctx, txID, sql, params, meta)
} else {
return nil
}
}
func (g genListener) PreExec(ctx context.Context, txID *uint16, sql *string, params *PP) error {
func (g genListener) PreExec(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreExecMeta) error {
if g.preExec != nil {
return g.preExec(ctx, txID, sql, params)
return g.preExec(ctx, txID, sql, params, meta)
} else {
return nil
}
}
func (g genListener) PostPing(result error) {
func (g genListener) PostPing(result error, meta PostPingMeta) {
if g.postPing != nil {
g.postPing(result)
g.postPing(result, meta)
}
}
func (g genListener) PostTxBegin(txid uint16, result error) {
func (g genListener) PostTxBegin(txid uint16, result error, meta PostTxBeginMeta) {
if g.postTxBegin != nil {
g.postTxBegin(txid, result)
g.postTxBegin(txid, result, meta)
}
}
func (g genListener) PostTxCommit(txid uint16, result error) {
func (g genListener) PostTxCommit(txid uint16, result error, meta PostTxCommitMeta) {
if g.postTxCommit != nil {
g.postTxCommit(txid, result)
g.postTxCommit(txid, result, meta)
}
}
func (g genListener) PostTxRollback(txid uint16, result error) {
func (g genListener) PostTxRollback(txid uint16, result error, meta PostTxRollbackMeta) {
if g.postTxRollback != nil {
g.postTxRollback(txid, result)
g.postTxRollback(txid, result, meta)
}
}
func (g genListener) PostQuery(txID *uint16, sqlOriginal string, sqlReal string, params PP) {
func (g genListener) PostQuery(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostQueryMeta) {
if g.postQuery != nil {
g.postQuery(txID, sqlOriginal, sqlReal, params)
g.postQuery(txID, sqlOriginal, sqlReal, params, meta)
}
}
func (g genListener) PostExec(txID *uint16, sqlOriginal string, sqlReal string, params PP) {
func (g genListener) PostExec(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostExecMeta) {
if g.postExec != nil {
g.postExec(txID, sqlOriginal, sqlReal, params)
g.postExec(txID, sqlOriginal, sqlReal, params, meta)
}
}
func NewPrePingListener(f func(ctx context.Context) error) Listener {
func NewPrePingListener(f func(ctx context.Context, meta PrePingMeta) error) Listener {
return genListener{prePing: f}
}
func NewPreTxBeginListener(f func(ctx context.Context, txid uint16) error) Listener {
func NewPreTxBeginListener(f func(ctx context.Context, txid uint16, meta PreTxBeginMeta) error) Listener {
return genListener{preTxBegin: f}
}
func NewPreTxCommitListener(f func(txid uint16) error) Listener {
func NewPreTxCommitListener(f func(txid uint16, meta PreTxCommitMeta) error) Listener {
return genListener{preTxCommit: f}
}
func NewPreTxRollbackListener(f func(txid uint16) error) Listener {
func NewPreTxRollbackListener(f func(txid uint16, meta PreTxRollbackMeta) error) Listener {
return genListener{preTxRollback: f}
}
func NewPreQueryListener(f func(ctx context.Context, txID *uint16, sql *string, params *PP) error) Listener {
func NewPreQueryListener(f func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreQueryMeta) error) Listener {
return genListener{preQuery: f}
}
func NewPreExecListener(f func(ctx context.Context, txID *uint16, sql *string, params *PP) error) Listener {
func NewPreExecListener(f func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreExecMeta) error) Listener {
return genListener{preExec: f}
}
func NewPreListener(f func(ctx context.Context, cmdtype string, txID *uint16, sql *string, params *PP) error) Listener {
return genListener{
preExec: func(ctx context.Context, txID *uint16, sql *string, params *PP) error {
preExec: func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreExecMeta) error {
return f(ctx, "EXEC", txID, sql, params)
},
preQuery: func(ctx context.Context, txID *uint16, sql *string, params *PP) error {
preQuery: func(ctx context.Context, txID *uint16, sql *string, params *PP, meta PreQueryMeta) error {
return f(ctx, "QUERY", txID, sql, params)
},
}
}
func NewPostPingListener(f func(result error)) Listener {
func NewPostPingListener(f func(result error, meta PostPingMeta)) Listener {
return genListener{postPing: f}
}
func NewPostTxBeginListener(f func(txid uint16, result error)) Listener {
func NewPostTxBeginListener(f func(txid uint16, result error, meta PostTxBeginMeta)) Listener {
return genListener{postTxBegin: f}
}
func NewPostTxCommitListener(f func(txid uint16, result error)) Listener {
func NewPostTxCommitListener(f func(txid uint16, result error, meta PostTxCommitMeta)) Listener {
return genListener{postTxCommit: f}
}
func NewPostTxRollbackListener(f func(txid uint16, result error)) Listener {
func NewPostTxRollbackListener(f func(txid uint16, result error, meta PostTxRollbackMeta)) Listener {
return genListener{postTxRollback: f}
}
func NewPostQueryListener(f func(txID *uint16, sqlOriginal string, sqlReal string, params PP)) Listener {
func NewPostQueryListener(f func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostQueryMeta)) Listener {
return genListener{postQuery: f}
}
func NewPostExecListener(f func(txID *uint16, sqlOriginal string, sqlReal string, params PP)) Listener {
func NewPostExecListener(f func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostExecMeta)) Listener {
return genListener{postExec: f}
}
func NewPostListener(f func(cmdtype string, txID *uint16, sqlOriginal string, sqlReal string, params PP)) Listener {
return genListener{
postExec: func(txID *uint16, sqlOriginal string, sqlReal string, params PP) {
postExec: func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostExecMeta) {
f("EXEC", txID, sqlOriginal, sqlReal, params)
},
postQuery: func(txID *uint16, sqlOriginal string, sqlReal string, params PP) {
postQuery: func(txID *uint16, sqlOriginal string, sqlReal string, params PP, meta PostQueryMeta) {
f("QUERY", txID, sqlOriginal, sqlReal, params)
},
}

47
sq/transaction.go
View File

@@ -6,6 +6,7 @@ import (
"github.com/jmoiron/sqlx"
"gogs.mikescher.com/BlackForestBytes/goext/exerr"
"gogs.mikescher.com/BlackForestBytes/goext/langext"
"time"
)
type TxStatus string
@@ -46,42 +47,56 @@ func NewTransaction(xtx *sqlx.Tx, txid uint16, db *database) Tx {
}
func (tx *transaction) Rollback() error {
t0 := time.Now()
preMeta := PreTxRollbackMeta{}
for _, v := range tx.db.lstr {
err := v.PreTxRollback(tx.id)
err := v.PreTxRollback(tx.id, preMeta)
if err != nil {
return exerr.Wrap(err, "failed to call SQL pre-rollback listener").Int("tx.id", int(tx.id)).Build()
}
}
t1 := time.Now()
result := tx.tx.Rollback()
if result == nil {
tx.status = TxStatusRollback
}
postMeta := PostTxRollbackMeta{Init: t0, Start: t1, End: time.Now(), ExecCounter: tx.execCtr, QueryCounter: tx.queryCtr}
for _, v := range tx.db.lstr {
v.PostTxRollback(tx.id, result)
v.PostTxRollback(tx.id, result, postMeta)
}
return result
}
func (tx *transaction) Commit() error {
t0 := time.Now()
preMeta := PreTxCommitMeta{}
for _, v := range tx.db.lstr {
err := v.PreTxCommit(tx.id)
err := v.PreTxCommit(tx.id, preMeta)
if err != nil {
return exerr.Wrap(err, "failed to call SQL pre-commit listener").Int("tx.id", int(tx.id)).Build()
}
}
t1 := time.Now()
result := tx.tx.Commit()
if result == nil {
tx.status = TxStatusComitted
}
postMeta := PostTxCommitMeta{Init: t0, Start: t1, End: time.Now(), ExecCounter: tx.execCtr, QueryCounter: tx.queryCtr}
for _, v := range tx.db.lstr {
v.PostTxRollback(tx.id, result)
v.PostTxCommit(tx.id, result, postMeta)
}
return result
@@ -89,21 +104,29 @@ func (tx *transaction) Commit() error {
func (tx *transaction) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Result, error) {
origsql := sqlstr
t0 := time.Now()
preMeta := PreExecMeta{}
for _, v := range tx.db.lstr {
err := v.PreExec(ctx, langext.Ptr(tx.id), &sqlstr, &prep)
err := v.PreExec(ctx, langext.Ptr(tx.id), &sqlstr, &prep, preMeta)
if err != nil {
return nil, exerr.Wrap(err, "failed to call SQL pre-exec listener").Int("tx.id", int(tx.id)).Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
}
t1 := time.Now()
res, err := tx.tx.NamedExecContext(ctx, sqlstr, prep)
tx.execCtr++
if tx.status == TxStatusInitial && err == nil {
tx.status = TxStatusActive
}
postMeta := PostExecMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range tx.db.lstr {
v.PostExec(langext.Ptr(tx.id), origsql, sqlstr, prep)
v.PostExec(langext.Ptr(tx.id), origsql, sqlstr, prep, postMeta)
}
if err != nil {
@@ -114,21 +137,29 @@ func (tx *transaction) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Re
func (tx *transaction) Query(ctx context.Context, sqlstr string, prep PP) (*sqlx.Rows, error) {
origsql := sqlstr
t0 := time.Now()
preMeta := PreQueryMeta{}
for _, v := range tx.db.lstr {
err := v.PreQuery(ctx, langext.Ptr(tx.id), &sqlstr, &prep)
err := v.PreQuery(ctx, langext.Ptr(tx.id), &sqlstr, &prep, preMeta)
if err != nil {
return nil, exerr.Wrap(err, "failed to call SQL pre-query listener").Int("tx.id", int(tx.id)).Str("original_sql", origsql).Str("sql", sqlstr).Any("sql_params", prep).Build()
}
}
t1 := time.Now()
rows, err := sqlx.NamedQueryContext(ctx, tx.tx, sqlstr, prep)
tx.queryCtr++
if tx.status == TxStatusInitial && err == nil {
tx.status = TxStatusActive
}
postMeta := PostQueryMeta{Init: t0, Start: t1, End: time.Now()}
for _, v := range tx.db.lstr {
v.PostQuery(langext.Ptr(tx.id), origsql, sqlstr, prep)
v.PostQuery(langext.Ptr(tx.id), origsql, sqlstr, prep, postMeta)
}
if err != nil {