BlackForestBytes/goext
20
0
Fork 0
Code Actions

v0.0.40

Browse Source
This commit is contained in:
Mike Schwörer 2022-12-21 15:34:59 +01:00
parent fff609db4a
commit e3b8d2cc0f
Signed by: Mikescher
GPG Key ID: D3C7172E0A70F8CF
3 changed files with 115 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
sq/database.go
View File

@ -12,14 +12,14 @@ type DB interface {
Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows, error) Query(ctx context.Context, sql string, prep PP) (*sqlx.Rows, error)
Ping(ctx context.Context) error Ping(ctx context.Context) error
BeginTransaction(ctx context.Context, iso sql.IsolationLevel) (Tx, error) BeginTransaction(ctx context.Context, iso sql.IsolationLevel) (Tx, error)
SetListener(listener Listener) AddListener(listener Listener)
} }
type database struct { type database struct {
db *sqlx.DB db *sqlx.DB
txctr uint16 txctr uint16
lock sync.Mutex lock sync.Mutex
lstr Listener lstr []Listener
} }
func NewDB(db *sqlx.DB) DB { func NewDB(db *sqlx.DB) DB {
@ -27,31 +27,50 @@ func NewDB(db *sqlx.DB) DB {
db: db, db: db,
txctr: 0, txctr: 0,
lock: sync.Mutex{}, 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(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 { if err != nil {
return nil, err return nil, err
} }
return res, nil 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(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 { if err != nil {
return nil, err 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 { func (db *database) Ping(ctx context.Context) error {
if db.lstr != nil { for _, v := range db.lstr {
db.lstr.OnPing() err := v.PrePing()
if err != nil {
return err
}
} }
err := db.db.PingContext(ctx) err := db.db.PingContext(ctx)
for _, v := range db.lstr {
v.PostPing(err)
}
if err != nil { if err != nil {
return err return err
} }
@ -76,8 +103,11 @@ func (db *database) BeginTransaction(ctx context.Context, iso sql.IsolationLevel
db.txctr += 1 // with overflow ! db.txctr += 1 // with overflow !
db.lock.Unlock() db.lock.Unlock()
if db.lstr != nil { for _, v := range db.lstr {
db.lstr.OnTxBegin(txid) err := v.PreTxBegin(txid)
if err != nil {
return nil, err
}
} }
xtx, err := db.db.BeginTxx(ctx, &sql.TxOptions{Isolation: iso}) 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 return nil, err
} }
for _, v := range db.lstr {
v.PostTxBegin(txid, err)
}
return NewTransaction(xtx, txid, db.lstr), nil return NewTransaction(xtx, txid, db.lstr), nil
} }

19
sq/listener.go
View File

@ -1,10 +1,17 @@
package sq package sq
type Listener interface { type Listener interface {
OnQuery(txID *uint16, sql *string, params *PP) PrePing() error
OnExec(txID *uint16, sql *string, params *PP) PreTxBegin(txid uint16) error
OnPing() PreTxCommit(txid uint16) error
OnTxBegin(txid uint16) PreTxRollback(txid uint16) error
OnTxCommit(txid uint16) PreQuery(txID *uint16, sql *string, params *PP) error
OnTxRollback(txid uint16) PreExec(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)
} }

68
sq/transaction.go
View File

@ -17,10 +17,10 @@ type Tx interface {
type transaction struct { type transaction struct {
tx *sqlx.Tx tx *sqlx.Tx
id uint16 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{ return &transaction{
tx: xtx, tx: xtx,
id: txid, id: txid,
@ -29,39 +29,75 @@ func NewTransaction(xtx *sqlx.Tx, txid uint16, lstr Listener) Tx {
} }
func (tx *transaction) Rollback() error { 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 { 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)
} }
func (tx *transaction) Exec(ctx context.Context, sql string, prep PP) (sql.Result, error) { return result
if tx.lstr != nil { }
tx.lstr.OnExec(langext.Ptr(tx.id), &sql, &prep)
func (tx *transaction) Exec(ctx context.Context, sqlstr string, prep PP) (sql.Result, error) {
origsql := sqlstr
for _, v := range tx.lstr {
err := v.PreExec(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 { if err != nil {
return nil, err return nil, err
} }
return res, nil 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(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 { if err != nil {
return nil, err return nil, err
} }