jx-callback/business/model/dao/dao.go

package dao

import (
	"errors"
	"reflect"
	"runtime/debug"
	"time"

	"git.rosy.net.cn/baseapi/utils"
	"git.rosy.net.cn/jx-callback/globals"
	"github.com/astaxie/beego/adapter/orm"
)

const (
	useGetRowsWhenGetRow      = true
	transactionWarningSeconds = 10 * 60 // 5分钟
)

type DaoDB struct {
	transactionLevel int
	Db               orm.Ormer

	transactionWatchTimer *time.Timer
	beginTransactionStack []byte
}

// func GetRows(db *gorm.DB, inPtr interface{}, sql string, values ...interface{}) (err error) {
// 	if db == nil {
// 		db = gormdb.GetDB()
// 	}
// 	topTypeInfo := reflect.TypeOf(inPtr)
// 	if topTypeInfo.Kind() != reflect.Ptr {
// 		panic("SelectEntities inPtr should be slice ptr (*[]Type)")
// 	}
// 	typeInfo := topTypeInfo.Elem()
// 	if typeInfo.Kind() != reflect.Slice {
// 		panic("SelectEntities inPtr should be slice ptr (*[]Type)")
// 	}
// 	elmType := typeInfo.Elem()

// 	valueInfo := reflect.ValueOf(inPtr)
// 	rows, err := db.Raw(sql, values...).Rows()
// 	if err == nil {
// 		defer rows.Close()
// 		for rows.Next() {
// 			var value reflect.Value
// 			if elmType.Kind() == reflect.Ptr {
// 				value = reflect.New(elmType.Elem())
// 			} else {
// 				value = reflect.New(elmType)
// 			}
// 			db.ScanRows(rows, value.Interface())
// 			if elmType.Kind() != reflect.Ptr {
// 				value = value.Elem()
// 			}
// 			valueInfo.Elem().Set(reflect.Append(valueInfo.Elem(), value))
// 		}
// 		return nil
// 	}
// 	return err
// }
// beego orm的对于传代表字段的字串，数据库字段名（完全匹配，区分大小写），结构体字段名（不区分大小写）都可以

func GetDB() *DaoDB {
	return &DaoDB{Db: orm.NewOrm()}
}

func Begin(db *DaoDB) (err error) {
	if db.transactionLevel == 0 {
		err = db.Db.Begin()
		if err == nil {
			db.startWatchTransaction()
		}
	}
	db.transactionLevel++
	return err
}

func (db *DaoDB) startWatchTransaction() {
	db.beginTransactionStack = debug.Stack()
	db.transactionWatchTimer = utils.AfterFuncWithRecover(transactionWarningSeconds*time.Second, func() {
		globals.SugarLogger.Warnf("Begin Transaction too long, %s", string(db.beginTransactionStack))
		db.transactionWatchTimer = nil
		db.beginTransactionStack = nil
	})
}

func (db *DaoDB) stopWatchTransaction() {
	if db.transactionWatchTimer != nil {
		db.transactionWatchTimer.Stop()
		db.transactionWatchTimer = nil
		db.beginTransactionStack = nil
	}
}

func Commit(db *DaoDB) (err error) {
	if db.transactionLevel == 1 {
		db.stopWatchTransaction()

		err = db.Db.Commit()
		db.transactionLevel = 0
	} else if db.transactionLevel > 1 {
		db.transactionLevel--
	}
	return err
}

func Rollback(db *DaoDB) (err error) {
	if db.transactionLevel > 0 {
		db.stopWatchTransaction()

		err = db.Db.Rollback()
	}
	db.transactionLevel = 0
	return err
}

func GetRow(db *DaoDB, inPtr interface{}, sql string, values ...interface{}) (err error) {
	if db == nil {
		db = GetDB()
	}
	if !useGetRowsWhenGetRow { // beego QueryRow有bug，嵌入的struct不能正常绑定
		err = db.Db.Raw(sql, values).QueryRow(inPtr)
	} else {
		typeInfo := reflect.TypeOf(inPtr)
		if typeInfo.Kind() != reflect.Ptr {
			return errors.New("inPtr must be ptr")
		}
		slice := reflect.New(reflect.SliceOf(typeInfo.Elem()))
		if err = GetRows(db, slice.Interface(), sql, values...); err == nil {
			slice = slice.Elem()
			if slice.Len() > 0 {
				reflect.ValueOf(inPtr).Elem().Set(slice.Index(0))
			} else {
				return orm.ErrNoRows
			}
		}
	}
	return err
}

func GetRows(db *DaoDB, inPtr interface{}, sql string, values ...interface{}) (err error) {
	if db == nil {
		db = GetDB()
	}
	_, err = db.Db.Raw(sql, values).QueryRows(inPtr)
	return err
}

func GetEntity(db *DaoDB, item interface{}, cols ...string) (err error) {
	if db == nil {
		db = GetDB()
	}
	err = utils.CallFuncLogErrorIgnore(func() error {
		return db.Db.Read(item, cols...)
	}, reflect.TypeOf(item).Name(), orm.ErrNoRows)
	return err
}

func UpdateEntity(db *DaoDB, item interface{}, cols ...string) (num int64, err error) {
	if db == nil {
		db = GetDB()
	}
	num, err = db.Db.Update(item, cols...)
	if err != nil && !IsDuplicateError(err) {
		globals.SugarLogger.Errorf("UpdateEntity %s failed with error:%v", reflect.TypeOf(item).Name(), err)
	}
	return num, err
}

func CreateEntity(db *DaoDB, item interface{}) (err error) {
	if db == nil {
		db = GetDB()
	}
	if _, err = db.Db.Insert(item); err != nil && !IsDuplicateError(err) {
		globals.SugarLogger.Errorf("CreateEntity %s failed with error:%v", reflect.TypeOf(item).Name(), err)
	}
	return err
}

// InsertMulti执行成功后ID不会改写成正确的（象Insert一样）
func CreateMultiEntities(db *DaoDB, item interface{}) (err error) {
	if db == nil {
		db = GetDB()
	}
	if _, err = db.Db.InsertMulti(20, item); err != nil && !IsDuplicateError(err) {
		globals.SugarLogger.Errorf("CreateEntity %s failed with error:%v", reflect.TypeOf(item).Name(), err)
	}
	return err
}

func CreateOrUpdate(db *DaoDB, item interface{}, colConflitAndArgs ...string) (err error) {
	if db == nil {
		db = GetDB()
	}
	_, err = db.Db.InsertOrUpdate(item, colConflitAndArgs...)
	return err
}

func DeleteEntity(db *DaoDB, item interface{}, cols ...string) (num int64, err error) {
	if db == nil {
		db = GetDB()
	}
	err = utils.CallFuncLogError(func() error {
		num, err = db.Db.Delete(item, cols...)
		return err
	}, reflect.TypeOf(item).Name())
	return num, err
}

func ExecuteSQL(db *DaoDB, sql string, params ...interface{}) (num int64, err error) {
	if db == nil {
		db = GetDB()
	}
	err = utils.CallFuncLogError(func() error {
		result, err2 := db.Db.Raw(sql, params...).Exec()
		if err2 == nil {
			num, _ = result.RowsAffected()
		}
		return err2
	}, sql)
	return num, err
}

// 此函数要求db在事务中，否则可能导致取得到的是另一个连接的数据
func GetLastTotalRowCount(db *DaoDB) int {
	countInfo := &struct{ Ct int }{}
	if err := GetRow(db, countInfo, "SELECT FOUND_ROWS() ct"); err == nil {
		return countInfo.Ct
	}
	return 0
}