From e64899e3e8c27615453308dc56abe4709a5d0ed6 Mon Sep 17 00:00:00 2001
From: aozhiwei <aozhiwei@kingsome.cn>
Date: Sun, 16 Jun 2024 12:57:35 +0800
Subject: [PATCH] 1

---
 dbfilter.go |  4 ++--
 dbpool.go   | 31 +++++++++++++++++++++++++++++--
 sysutils.go | 29 +++++++++++++++++++++++++++++
 types.go    |  1 -
 4 files changed, 60 insertions(+), 5 deletions(-)

diff --git a/dbfilter.go b/dbfilter.go
index e7a0c40..a41aa85 100644
--- a/dbfilter.go
+++ b/dbfilter.go
@@ -103,10 +103,10 @@ func (this *dbFilter) GE(fieldName string, val string) DbQueryFilter {
 	return f
 }
 
-func (this *dbFilter) Custom(fieldName string, val string) DbQueryFilter {
+func (this *dbFilter) Custom(val string) DbQueryFilter {
 	f := &PageQueryOne{
 		cond: QC_CUSTOM,
-		fieldName: fieldName,
+		fieldName: "",
 		val: val}
 	return f
 }
diff --git a/dbpool.go b/dbpool.go
index 0206815..c2facdf 100644
--- a/dbpool.go
+++ b/dbpool.go
@@ -215,6 +215,32 @@ func (this *dbPool) Upsert(
 		})
 }
 
+func (this *dbPool) UpsertEx(
+	dataSource string,
+	tblName string,
+	whereKv [][]string,
+	updateKv [][]string,
+	insertKv [][]string,
+	cb ExecResultCb,
+	updateCb func(*DataSet) bool) {
+	this.OrmSelectOne(dataSource, tblName, whereKv,
+		func(err error, ds *DataSet) {
+			if err != nil {
+				cb(err, 0, 0)
+				return
+			}
+			if ds.Next() {
+				if len(updateKv) > 0 {
+					if updateCb(ds) {
+						this.Update(dataSource, tblName, whereKv, updateKv, cb)
+					}
+				}
+			} else {
+				this.Insert(dataSource, tblName, insertKv, cb)
+			}
+		})
+}
+
 func (this *dbPool) PageQuery(
 	dataSource string,
 	perPage int32,
@@ -287,10 +313,11 @@ func (this *dbPool) StreamPageQuery(
 	if filter != nil {
 		finalySql += filter.GenSql()
 	}
+	finalySql += fmt.Sprintf(" LIMIT %d ", pageSize + 1)
 	if orderBy != "" {
 		finalySql += " " + orderBy + " "
 	}
-	//GetSysLog().Info("finalySql:%s", finalySql)
+	GetSysLog().Info("finalySql:%s", finalySql)
 	this.queryOne(
 		dataSource,
 		finalySql,
@@ -302,7 +329,7 @@ func (this *dbPool) StreamPageQuery(
 			}
 			pagination.PreviousCursor = cursor
 			for rows.Next() {
-				if (rows.NumOfReaded() < int64(pageSize)) {
+				if (rows.NumOfReaded() <= int64(pageSize)) {
 					fillCb(rows)
 				} else if (rows.NumOfReaded() == int64(pageSize)) {
 					pagination.NextCursor = q5.ToInt64(rows.GetByName("idx"))
diff --git a/sysutils.go b/sysutils.go
index e070c3b..9e4f5fe 100644
--- a/sysutils.go
+++ b/sysutils.go
@@ -3,6 +3,8 @@ package f5
 import (
 	"q5"
 	"os"
+	"time"
+	"sync"
 	"errors"
 	"gorm.io/gorm"
 )
@@ -14,6 +16,8 @@ const (
 )
 
 var serverEnv int32
+var globalLock sync.Mutex
+var globalLockHash map[string]*sync.Mutex
 
 func IsTestEnv() bool {
 	return serverEnv == TEST_ENV
@@ -45,6 +49,30 @@ func IsOrmErrRecordNotFound(err error) bool {
 	return errors.Is(err, gorm.ErrRecordNotFound)
 }
 
+func AllocLock(key string) *sync.Mutex {
+	var l *sync.Mutex
+	globalLock.Lock()
+	if p, ok := globalLockHash[key]; ok {
+		l = p
+	} else {
+		l = new(sync.Mutex)
+		globalLockHash[key] = l
+	}
+	globalLock.Unlock()
+	go func () {
+		time.Sleep(time.Second * 30)
+		globalLock.Lock()
+		delete(globalLockHash, key)
+		globalLock.Unlock()
+	}()
+	l.Lock()
+	return l
+}
+
+func ReleaseLock(l *sync.Mutex) {
+	l.Unlock()
+}
+
 func init() {
 	switch os.Getenv("SERVER_ENV") {
 	case "TEST":
@@ -54,4 +82,5 @@ func init() {
 	default:
 		serverEnv = ONLINE_ENV
 	}
+	globalLockHash = map[string]*sync.Mutex{}
 }
diff --git a/types.go b/types.go
index 59a22d7..cd8d196 100644
--- a/types.go
+++ b/types.go
@@ -2,7 +2,6 @@ package f5
 
 import (
 	"q5"
-
 	"github.com/gin-gonic/gin"
 )