796 lines
22 KiB
Go
796 lines
22 KiB
Go
package db
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"go_dreamfactory/comm"
|
|
"go_dreamfactory/lego/core"
|
|
"go_dreamfactory/lego/sys/log"
|
|
"go_dreamfactory/lego/sys/mgo"
|
|
lgredis "go_dreamfactory/lego/sys/redis"
|
|
"go_dreamfactory/lego/sys/redis/pipe"
|
|
"go_dreamfactory/lego/utils/codec"
|
|
"go_dreamfactory/lego/utils/codec/codecore"
|
|
"go_dreamfactory/lego/utils/codec/json"
|
|
"reflect"
|
|
"time"
|
|
"unsafe"
|
|
|
|
"github.com/go-redis/redis/v8"
|
|
|
|
"github.com/modern-go/reflect2"
|
|
"go.mongodb.org/mongo-driver/bson"
|
|
"go.mongodb.org/mongo-driver/bson/primitive"
|
|
"go.mongodb.org/mongo-driver/mongo"
|
|
)
|
|
|
|
var defconf = &codecore.Config{
|
|
SortMapKeys: true,
|
|
IndentionStep: 1,
|
|
OnlyTaggedField: false,
|
|
DisallowUnknownFields: false,
|
|
CaseSensitive: false,
|
|
TagKey: "json",
|
|
}
|
|
|
|
const (
|
|
DB_ModelTable core.SqlTable = "model_log"
|
|
)
|
|
|
|
func newDBConn(conf DBConfig) (conn *DBConn, err error) {
|
|
conn = &DBConn{}
|
|
if conf.RedisIsCluster {
|
|
conn.Redis, err = lgredis.NewSys(
|
|
lgredis.SetRedisType(lgredis.Redis_Cluster),
|
|
lgredis.SetRedis_Cluster_Addr(conf.RedisAddr),
|
|
lgredis.SetRedis_Cluster_Password(conf.RedisPassword))
|
|
} else {
|
|
conn.Redis, err = lgredis.NewSys(
|
|
lgredis.SetRedisType(lgredis.Redis_Single),
|
|
lgredis.SetRedis_Single_Addr(conf.RedisAddr[0]),
|
|
lgredis.SetRedis_Single_Password(conf.RedisPassword),
|
|
lgredis.SetRedis_Single_DB(conf.RedisDB),
|
|
)
|
|
}
|
|
if err != nil {
|
|
log.Error(err.Error(), log.Field{"config", conf})
|
|
return
|
|
}
|
|
if conn.Mgo, err = mgo.NewSys(
|
|
mgo.SetMongodbUrl(conf.MongodbUrl),
|
|
mgo.SetMongodbDatabase(conf.MongodbDatabase),
|
|
); err != nil {
|
|
log.Error(err.Error(), log.Field{"config", conf})
|
|
return
|
|
}
|
|
return
|
|
}
|
|
|
|
type DBConn struct {
|
|
Redis lgredis.ISys
|
|
Mgo mgo.ISys
|
|
}
|
|
|
|
func NewDBModel(tableName string, expired time.Duration, conn *DBConn) *DBModel {
|
|
return &DBModel{
|
|
TableName: tableName,
|
|
Expired: expired,
|
|
Redis: conn.Redis,
|
|
DB: conn.Mgo,
|
|
}
|
|
}
|
|
|
|
//DB模型
|
|
type DBModel struct {
|
|
TableName string
|
|
Expired time.Duration //过期时间
|
|
Redis lgredis.ISys
|
|
DB mgo.ISys
|
|
}
|
|
|
|
func (this *DBModel) ukey(uid string) string {
|
|
return fmt.Sprintf("%s:%s{%s}", this.TableName, uid, this.TableName)
|
|
}
|
|
func (this *DBModel) ukeylist(uid string, id string) string {
|
|
return fmt.Sprintf("%s:%s-%s{%s}", this.TableName, uid, id, this.TableName)
|
|
}
|
|
func (this *DBModel) InsertModelLogs(table string, uID string, target interface{}) (err error) {
|
|
|
|
data := &comm.Autogenerated{
|
|
ID: primitive.NewObjectID().Hex(),
|
|
UID: uID,
|
|
Act: string(comm.LogHandleType_Insert),
|
|
}
|
|
data.D = append(data.D, table) // D[0]
|
|
data.D = append(data.D, target) // D[1]
|
|
|
|
_, err = this.DB.InsertOne(DB_ModelTable, data)
|
|
if err != nil {
|
|
log.Errorf("insert model db err %v", err)
|
|
}
|
|
|
|
return err
|
|
}
|
|
func (this *DBModel) DeleteModelLogs(table string, uID string, where interface{}) (err error) {
|
|
|
|
data := &comm.Autogenerated{
|
|
ID: primitive.NewObjectID().Hex(),
|
|
UID: uID,
|
|
Act: string(comm.LogHandleType_Delete),
|
|
}
|
|
|
|
data.D = append(data.D, table) // D[0]
|
|
data.D = append(data.D, where) // D[1]
|
|
|
|
_, err = this.DB.InsertOne(DB_ModelTable, data)
|
|
if err != nil {
|
|
log.Errorf("insert model db err %v", err)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
func (this *DBModel) UpdateModelLogs(table string, uID string, where bson.M, target interface{}) (err error) {
|
|
|
|
data := &comm.Autogenerated{
|
|
ID: primitive.NewObjectID().Hex(),
|
|
UID: uID,
|
|
Act: string(comm.LogHandleType_Update),
|
|
}
|
|
data.D = append(data.D, table) // D[0]
|
|
data.D = append(data.D, where) // D[1]
|
|
data.D = append(data.D, target) // D[2]
|
|
|
|
_, err = this.DB.InsertOne(DB_ModelTable, data)
|
|
if err != nil {
|
|
log.Errorf("insert model db err %v", err)
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
//添加新的数据
|
|
func (this *DBModel) Add(uid string, data interface{}, opt ...DBOption) (err error) {
|
|
if err = this.Redis.HMSet(this.ukey(uid), data); err != nil {
|
|
return
|
|
}
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.InsertModelLogs(this.TableName, uid, []interface{}{data})
|
|
}
|
|
if this.Expired > 0 {
|
|
// err = this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//添加新的数据到列表
|
|
func (this *DBModel) AddList(uid string, id string, data interface{}, opt ...DBOption) (err error) {
|
|
key := this.ukeylist(uid, id)
|
|
if err = this.Redis.HMSet(key, data); err != nil {
|
|
return
|
|
}
|
|
if err = this.Redis.HSet(this.ukey(uid), id, key); err != nil {
|
|
return
|
|
}
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.InsertModelLogs(this.TableName, uid, []interface{}{data})
|
|
}
|
|
if this.Expired > 0 {
|
|
// err = this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//添加新的多个数据到列表 data map[string]type
|
|
func (this *DBModel) AddLists(uid string, data interface{}, opt ...DBOption) (err error) {
|
|
vof := reflect.ValueOf(data)
|
|
if !vof.IsValid() {
|
|
return fmt.Errorf("Model_Comp: AddLists(nil)")
|
|
}
|
|
if vof.Kind() != reflect.Map {
|
|
return fmt.Errorf("Model_Comp: AddLists(non-pointer %T)", data)
|
|
}
|
|
listskeys := make(map[string]string)
|
|
keys := vof.MapKeys()
|
|
lists := make([]interface{}, 0, len(keys))
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
for _, k := range keys {
|
|
value := vof.MapIndex(k)
|
|
keydata := k.Interface().(string)
|
|
valuedata := value.Interface()
|
|
key := this.ukeylist(uid, keydata)
|
|
pipe.HMSet(key, valuedata)
|
|
listskeys[keydata] = key
|
|
lists = append(lists, valuedata)
|
|
}
|
|
pipe.HMSetForMap(this.ukey(uid), listskeys)
|
|
if _, err = pipe.Exec(); err != nil {
|
|
return
|
|
}
|
|
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.InsertModelLogs(this.TableName, uid, lists)
|
|
}
|
|
if this.Expired > 0 {
|
|
childs := make(map[string]struct{}, len(listskeys))
|
|
for _, v := range listskeys {
|
|
childs[v] = struct{}{}
|
|
}
|
|
UpDateModelExpired(this.ukey(uid), childs, this.Expired)
|
|
// this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
}
|
|
return
|
|
}
|
|
|
|
//添加队列
|
|
func (this *DBModel) AddQueues(key string, uplimit int64, data interface{}) (outkey []string, err error) {
|
|
vof := reflect.ValueOf(data)
|
|
if !vof.IsValid() {
|
|
err = fmt.Errorf("Model_Comp: AddLists(nil)")
|
|
return
|
|
}
|
|
if vof.Kind() != reflect.Map {
|
|
err = fmt.Errorf("Model_Comp: AddLists(non-pointer %T)", data)
|
|
return
|
|
}
|
|
keys := make([]string, 0)
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
for _, k := range vof.MapKeys() {
|
|
value := vof.MapIndex(k)
|
|
tkey := k.Interface().(string)
|
|
valuedata := value.Interface()
|
|
pipe.HMSet(tkey, valuedata)
|
|
keys = append(keys, tkey)
|
|
}
|
|
pipe.RPushForStringSlice(key, keys...)
|
|
lcmd := pipe.Llen(key)
|
|
|
|
if _, err = pipe.Exec(); err == nil {
|
|
if lcmd.Val() > uplimit*3 { //操作3倍上限移除多余数据
|
|
off := uplimit - lcmd.Val()
|
|
if outkey, err = this.Redis.LRangeToStringSlice(key, 0, int(off-1)).Result(); err != nil {
|
|
return
|
|
}
|
|
pipe.Ltrim(key, int(off), -1)
|
|
for _, v := range outkey {
|
|
pipe.Delete(v)
|
|
}
|
|
_, err = pipe.Exec()
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
//修改数据多个字段 uid 作为主键
|
|
func (this *DBModel) Change(uid string, data map[string]interface{}, opt ...DBOption) (err error) {
|
|
if err = this.Redis.HMSet(this.ukey(uid), data); err != nil {
|
|
return
|
|
}
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.UpdateModelLogs(this.TableName, uid, bson.M{"uid": uid}, data)
|
|
}
|
|
if this.Expired > 0 {
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
// this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
//修改数据多个字段 uid 作为主键
|
|
func (this *DBModel) ChangeList(uid string, _id string, data map[string]interface{}, opt ...DBOption) (err error) {
|
|
if err = this.Redis.HMSet(this.ukeylist(uid, _id), data); err != nil {
|
|
return
|
|
}
|
|
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.UpdateModelLogs(this.TableName, uid, bson.M{"_id": _id, "uid": uid}, data)
|
|
}
|
|
if this.Expired > 0 {
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
// this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
//读取全部数据
|
|
func (this *DBModel) Get(uid string, data interface{}, opt ...DBOption) (err error) {
|
|
if err = this.Redis.HGetAll(this.ukey(uid), data); err != nil && err != lgredis.RedisNil {
|
|
return
|
|
}
|
|
if err == lgredis.RedisNil {
|
|
if err = this.DB.FindOne(core.SqlTable(this.TableName), bson.M{"uid": uid}).Decode(data); err != nil {
|
|
return
|
|
}
|
|
err = this.Redis.HMSet(this.ukey(uid), data)
|
|
}
|
|
// option := newDBOption(opt...)
|
|
if this.Expired > 0 {
|
|
// this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//获取列表数据 注意 data 必须是 切片的指针 *[]type
|
|
func (this *DBModel) GetList(uid string, data interface{}) (err error) {
|
|
var (
|
|
dtype reflect2.Type
|
|
dkind reflect.Kind
|
|
sType reflect2.Type
|
|
sliceType *reflect2.UnsafeSliceType
|
|
sliceelemType reflect2.Type
|
|
decoder codecore.IDecoderMapJson
|
|
encoder codecore.IEncoderMapJson
|
|
dptr unsafe.Pointer
|
|
elemPtr unsafe.Pointer
|
|
n int
|
|
ok bool
|
|
keys map[string]string
|
|
tempdata map[string]string
|
|
result []*redis.StringStringMapCmd
|
|
c *mongo.Cursor
|
|
)
|
|
keys = make(map[string]string)
|
|
dptr = reflect2.PtrOf(data)
|
|
dtype = reflect2.TypeOf(data)
|
|
dkind = dtype.Kind()
|
|
if dkind != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(non-pointer %T)", data)
|
|
return
|
|
}
|
|
sType = dtype.(*reflect2.UnsafePtrType).Elem()
|
|
if sType.Kind() != reflect.Slice {
|
|
err = fmt.Errorf("MCompModel: GetList(data no slice %T)", data)
|
|
return
|
|
}
|
|
sliceType = sType.(*reflect2.UnsafeSliceType)
|
|
sliceelemType = sliceType.Elem()
|
|
if sliceelemType.Kind() != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(sliceelemType non-pointer %T)", data)
|
|
return
|
|
}
|
|
if decoder, ok = codec.DecoderOf(sliceelemType, defconf).(codecore.IDecoderMapJson); !ok {
|
|
err = fmt.Errorf("MCompModel: GetList(data not support MarshalMapJson %T)", data)
|
|
return
|
|
}
|
|
sliceelemType = sliceelemType.(*reflect2.UnsafePtrType).Elem()
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
if keys, err = this.Redis.HGetAllToMapString(this.ukey(uid)); err == nil {
|
|
result = make([]*redis.StringStringMapCmd, 0)
|
|
for _, v := range keys {
|
|
cmd := pipe.HGetAllToMapString(v)
|
|
result = append(result, cmd)
|
|
}
|
|
pipe.Exec()
|
|
for _, v := range result {
|
|
tempdata, err = v.Result()
|
|
sliceType.UnsafeGrow(dptr, n+1)
|
|
elemPtr = sliceType.UnsafeGetIndex(dptr, n)
|
|
if *((*unsafe.Pointer)(elemPtr)) == nil {
|
|
newPtr := sliceelemType.UnsafeNew()
|
|
if err = decoder.DecodeForMapJson(newPtr, json.GetReader([]byte{}), tempdata); err != nil {
|
|
log.Errorf("err:%v", err)
|
|
return
|
|
}
|
|
*((*unsafe.Pointer)(elemPtr)) = newPtr
|
|
} else {
|
|
decoder.DecodeForMapJson(*((*unsafe.Pointer)(elemPtr)), json.GetReader([]byte{}), tempdata)
|
|
}
|
|
n++
|
|
}
|
|
}
|
|
if err == lgredis.RedisNil {
|
|
//query from mgo
|
|
if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"uid": uid}); err != nil {
|
|
return err
|
|
} else {
|
|
if encoder, ok = codec.EncoderOf(sliceelemType, defconf).(codecore.IEncoderMapJson); !ok {
|
|
err = fmt.Errorf("MCompModel: GetList(data not support UnMarshalMapJson %T)", data)
|
|
return
|
|
}
|
|
n = 0
|
|
for c.Next(context.Background()) {
|
|
_id := c.Current.Lookup("_id").StringValue()
|
|
sliceType.UnsafeGrow(dptr, n+1)
|
|
elemPtr = sliceType.UnsafeGetIndex(dptr, n)
|
|
if *((*unsafe.Pointer)(elemPtr)) == nil {
|
|
newPtr := sliceelemType.UnsafeNew()
|
|
*((*unsafe.Pointer)(elemPtr)) = newPtr
|
|
}
|
|
elem := sliceType.GetIndex(data, n)
|
|
if err = c.Decode(elem); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = encoder.EncodeToMapJson(*((*unsafe.Pointer)(elemPtr)), json.GetWriter()); err != nil {
|
|
return
|
|
}
|
|
key := this.ukeylist(uid, _id)
|
|
pipe.HMSetForMap(key, tempdata)
|
|
keys[_id] = key
|
|
n++
|
|
}
|
|
if len(keys) > 0 {
|
|
pipe.HMSetForMap(this.ukey(uid), keys)
|
|
_, err = pipe.Exec()
|
|
}
|
|
}
|
|
}
|
|
if this.Expired > 0 {
|
|
childs := make(map[string]struct{}, len(keys))
|
|
for _, v := range keys {
|
|
childs[v] = struct{}{}
|
|
}
|
|
UpDateModelExpired(this.ukey(uid), childs, this.Expired)
|
|
// this.Redis.Expire(this.ukey(uid), option.Expire)
|
|
}
|
|
return err
|
|
}
|
|
|
|
//获取队列数据
|
|
func (this *DBModel) GetQueues(key string, count int, data interface{}) (err error) {
|
|
var (
|
|
dtype reflect2.Type
|
|
dkind reflect.Kind
|
|
sType reflect2.Type
|
|
sliceType *reflect2.UnsafeSliceType
|
|
sliceelemType reflect2.Type
|
|
dptr unsafe.Pointer
|
|
elemPtr unsafe.Pointer
|
|
decoder codecore.IDecoderMapJson
|
|
ok bool
|
|
n int
|
|
keys = make([]string, 0)
|
|
result = make([]*redis.StringStringMapCmd, 0)
|
|
tempdata map[string]string
|
|
)
|
|
|
|
dptr = reflect2.PtrOf(data)
|
|
dtype = reflect2.TypeOf(data)
|
|
dkind = dtype.Kind()
|
|
if dkind != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(non-pointer %T)", data)
|
|
return
|
|
}
|
|
sType = dtype.(*reflect2.UnsafePtrType).Elem()
|
|
if sType.Kind() != reflect.Slice {
|
|
err = fmt.Errorf("MCompModel: GetList(data no slice %T)", data)
|
|
return
|
|
}
|
|
sliceType = sType.(*reflect2.UnsafeSliceType)
|
|
sliceelemType = sliceType.Elem()
|
|
if sliceelemType.Kind() != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(sliceelemType non-pointer %T)", data)
|
|
return
|
|
}
|
|
if decoder, ok = codec.DecoderOf(sliceelemType, defconf).(codecore.IDecoderMapJson); !ok {
|
|
err = fmt.Errorf("MCompModel: GetList(data not support MarshalMapJson %T)", data)
|
|
return
|
|
}
|
|
sliceelemType = sliceelemType.(*reflect2.UnsafePtrType).Elem()
|
|
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
if keys, err = this.Redis.LRangeToStringSlice(key, -1*count, -1).Result(); err == nil {
|
|
result = make([]*redis.StringStringMapCmd, 0)
|
|
for _, v := range keys {
|
|
cmd := pipe.HGetAllToMapString(v)
|
|
result = append(result, cmd)
|
|
}
|
|
pipe.Exec()
|
|
for _, v := range result {
|
|
tempdata, err = v.Result()
|
|
sliceType.UnsafeGrow(dptr, n+1)
|
|
elemPtr = sliceType.UnsafeGetIndex(dptr, n)
|
|
if *((*unsafe.Pointer)(elemPtr)) == nil {
|
|
newPtr := sliceelemType.UnsafeNew()
|
|
if err = decoder.DecodeForMapJson(newPtr, json.GetReader([]byte{}), tempdata); err != nil {
|
|
log.Errorf("err:%v", err)
|
|
return
|
|
}
|
|
*((*unsafe.Pointer)(elemPtr)) = newPtr
|
|
} else {
|
|
decoder.DecodeForMapJson(*((*unsafe.Pointer)(elemPtr)), json.GetReader([]byte{}), tempdata)
|
|
}
|
|
n++
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
//读取单个数据中 多个字段数据
|
|
func (this *DBModel) GetFields(uid string, data interface{}, fields ...string) (err error) {
|
|
if err = this.Redis.HMGet(this.ukey(uid), data, fields...); err != nil && err != lgredis.RedisNil {
|
|
return
|
|
}
|
|
if err == lgredis.RedisNil {
|
|
if err = this.DB.FindOne(core.SqlTable(this.TableName), bson.M{"uid": uid}).Decode(data); err != nil {
|
|
return err
|
|
}
|
|
if err = this.Redis.HMSet(this.ukey(uid), data); err != nil {
|
|
return
|
|
}
|
|
}
|
|
if this.Expired > 0 {
|
|
UpDateModelExpired(this.ukey(uid), nil, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//读取List列表中单个数据中 多个字段数据
|
|
func (this *DBModel) GetListFields(uid string, id string, data interface{}, fields ...string) (err error) {
|
|
var (
|
|
c *mongo.Cursor
|
|
keys map[string]string
|
|
tempdata map[string]string
|
|
)
|
|
if err = this.Redis.HMGet(this.ukeylist(uid, id), data, fields...); err != nil && err != lgredis.RedisNil {
|
|
return
|
|
}
|
|
if err == lgredis.RedisNil {
|
|
if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"uid": uid}); err != nil {
|
|
return err
|
|
} else {
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
dtype := reflect2.TypeOf(data).(*reflect2.UnsafeStructType)
|
|
for c.Next(context.Background()) {
|
|
_id := c.Current.Lookup("_id").StringValue()
|
|
if _id != id {
|
|
temp := dtype.New()
|
|
if err = c.Decode(temp); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = json.MarshalMap(temp); err != nil {
|
|
return
|
|
}
|
|
} else {
|
|
if err = c.Decode(data); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = json.MarshalMap(data); err != nil {
|
|
return
|
|
}
|
|
}
|
|
key := this.ukeylist(uid, _id)
|
|
pipe.HMSetForMap(key, tempdata)
|
|
keys[_id] = key
|
|
}
|
|
if len(keys) > 0 {
|
|
pipe.HMSetForMap(this.ukey(uid), keys)
|
|
_, err = pipe.Exec()
|
|
}
|
|
}
|
|
}
|
|
if this.Expired > 0 {
|
|
childs := make(map[string]struct{}, len(keys))
|
|
for _, v := range keys {
|
|
childs[v] = struct{}{}
|
|
}
|
|
UpDateModelExpired(this.ukey(uid), childs, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//读取列表数据中单个数据
|
|
func (this *DBModel) GetListObj(uid string, id string, data interface{}) (err error) {
|
|
var (
|
|
c *mongo.Cursor
|
|
keys map[string]string
|
|
tempdata map[string]string
|
|
isfound bool
|
|
)
|
|
keys = make(map[string]string)
|
|
if err = this.Redis.HGetAll(this.ukeylist(uid, id), data); err != nil && err != lgredis.RedisNil {
|
|
return
|
|
}
|
|
isfound = true
|
|
if err == lgredis.RedisNil {
|
|
isfound = false
|
|
if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"uid": uid}); err != nil {
|
|
return err
|
|
} else {
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
// dtype := reflect2.TypeOf(data).(*reflect2.UnsafeStructType)
|
|
dtype := reflect2.TypeOf(data).(*reflect2.UnsafePtrType).Elem().(*reflect2.UnsafeStructType)
|
|
for c.Next(context.Background()) {
|
|
_id := c.Current.Lookup("_id").StringValue()
|
|
if _id != id {
|
|
temp := dtype.New()
|
|
if err = c.Decode(temp); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = json.MarshalMap(temp); err != nil {
|
|
return
|
|
}
|
|
} else {
|
|
if err = c.Decode(data); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = json.MarshalMap(data); err != nil {
|
|
return
|
|
}
|
|
isfound = true
|
|
}
|
|
key := this.ukeylist(uid, _id)
|
|
pipe.HMSetForMap(key, tempdata)
|
|
keys[_id] = key
|
|
}
|
|
if len(keys) > 0 {
|
|
pipe.HMSetForMap(this.ukey(uid), keys)
|
|
_, err = pipe.Exec()
|
|
}
|
|
}
|
|
}
|
|
if !isfound {
|
|
err = fmt.Errorf("no found data table:%s uid:%s id:%s", this.TableName, uid, id)
|
|
return
|
|
}
|
|
if this.Expired > 0 {
|
|
childs := make(map[string]struct{}, len(keys))
|
|
for _, v := range keys {
|
|
childs[v] = struct{}{}
|
|
}
|
|
UpDateModelExpired(this.ukey(uid), childs, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//读取列表数据中单个数据
|
|
func (this *DBModel) GetListObjs(uid string, ids []string, data interface{}) (err error) {
|
|
var (
|
|
dtype reflect2.Type
|
|
dkind reflect.Kind
|
|
sType reflect2.Type
|
|
sliceType *reflect2.UnsafeSliceType
|
|
sliceelemType reflect2.Type
|
|
decoder codecore.IDecoderMapJson
|
|
encoder codecore.IEncoderMapJson
|
|
dptr unsafe.Pointer
|
|
elemPtr unsafe.Pointer
|
|
n int
|
|
ok bool
|
|
keys map[string]string = make(map[string]string)
|
|
tempdata map[string]string
|
|
onfound []string = make([]string, 0, len(ids))
|
|
pipe *pipe.RedisPipe = this.Redis.RedisPipe(context.TODO())
|
|
result []*redis.StringStringMapCmd = make([]*redis.StringStringMapCmd, len(ids))
|
|
c *mongo.Cursor
|
|
)
|
|
dptr = reflect2.PtrOf(data)
|
|
dtype = reflect2.TypeOf(data)
|
|
dkind = dtype.Kind()
|
|
if dkind != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(non-pointer %T)", data)
|
|
return
|
|
}
|
|
sType = dtype.(*reflect2.UnsafePtrType).Elem()
|
|
if sType.Kind() != reflect.Slice {
|
|
err = fmt.Errorf("MCompModel: GetList(data no slice %T)", data)
|
|
return
|
|
}
|
|
sliceType = sType.(*reflect2.UnsafeSliceType)
|
|
sliceelemType = sliceType.Elem()
|
|
if sliceelemType.Kind() != reflect.Ptr {
|
|
err = fmt.Errorf("MCompModel: GetList(sliceelemType non-pointer %T)", data)
|
|
return
|
|
}
|
|
if decoder, ok = codec.DecoderOf(sliceelemType, defconf).(codecore.IDecoderMapJson); !ok {
|
|
err = fmt.Errorf("MCompModel: GetList(data not support MarshalMapJson %T)", data)
|
|
return
|
|
}
|
|
sliceelemType = sliceelemType.(*reflect2.UnsafePtrType).Elem()
|
|
for i, v := range ids {
|
|
result[i] = pipe.HGetAllToMapString(this.ukeylist(uid, v))
|
|
}
|
|
if _, err = pipe.Exec(); err == nil {
|
|
for i, v := range result {
|
|
if tempdata, err = v.Result(); err == nil {
|
|
sliceType.UnsafeGrow(dptr, n+1)
|
|
elemPtr = sliceType.UnsafeGetIndex(dptr, n)
|
|
if *((*unsafe.Pointer)(elemPtr)) == nil {
|
|
newPtr := sliceelemType.UnsafeNew()
|
|
if err = decoder.DecodeForMapJson(newPtr, json.GetReader([]byte{}), tempdata); err != nil {
|
|
log.Errorf("err:%v", err)
|
|
return
|
|
}
|
|
*((*unsafe.Pointer)(elemPtr)) = newPtr
|
|
} else {
|
|
decoder.DecodeForMapJson(*((*unsafe.Pointer)(elemPtr)), json.GetReader([]byte{}), tempdata)
|
|
}
|
|
n++
|
|
} else {
|
|
onfound = append(onfound, ids[i])
|
|
}
|
|
}
|
|
} else {
|
|
onfound = ids
|
|
}
|
|
|
|
if err == lgredis.RedisNil {
|
|
if c, err = this.DB.Find(core.SqlTable(this.TableName), bson.M{"uid": uid}); err != nil {
|
|
return err
|
|
} else {
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
for c.Next(context.Background()) {
|
|
_id := c.Current.Lookup("_id").StringValue()
|
|
sliceType.UnsafeGrow(dptr, n+1)
|
|
elemPtr = sliceType.UnsafeGetIndex(dptr, n)
|
|
if *((*unsafe.Pointer)(elemPtr)) == nil {
|
|
newPtr := sliceelemType.UnsafeNew()
|
|
*((*unsafe.Pointer)(elemPtr)) = newPtr
|
|
}
|
|
elem := sliceType.GetIndex(data, n)
|
|
if err = c.Decode(elem); err != nil {
|
|
return
|
|
}
|
|
if tempdata, err = encoder.EncodeToMapJson(*((*unsafe.Pointer)(elemPtr)), json.GetWriter()); err != nil {
|
|
return
|
|
}
|
|
key := this.ukeylist(uid, _id)
|
|
pipe.HMSetForMap(key, tempdata)
|
|
keys[_id] = key
|
|
n++
|
|
}
|
|
if len(keys) > 0 {
|
|
pipe.HMSetForMap(this.ukey(uid), keys)
|
|
_, err = pipe.Exec()
|
|
}
|
|
}
|
|
}
|
|
if this.Expired > 0 {
|
|
childs := make(map[string]struct{}, len(keys))
|
|
for _, v := range keys {
|
|
childs[v] = struct{}{}
|
|
}
|
|
UpDateModelExpired(this.ukey(uid), childs, this.Expired)
|
|
}
|
|
return
|
|
}
|
|
|
|
//删除用户数据
|
|
func (this *DBModel) Del(uid string, opt ...DBOption) (err error) {
|
|
err = this.Redis.Delete(this.ukey(uid))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
option := newDBOption(opt...)
|
|
if option.IsMgoLog {
|
|
err = this.DeleteModelLogs(this.TableName, uid, bson.M{"uid": uid})
|
|
}
|
|
return nil
|
|
}
|
|
|
|
//删除多条数据
|
|
func (this *DBModel) DelListlds(uid string, ids ...string) (err error) {
|
|
listkey := this.ukey(uid)
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
for _, v := range ids {
|
|
key := this.ukeylist(uid, v)
|
|
pipe.Delete(key)
|
|
}
|
|
pipe.HDel(listkey, ids...)
|
|
if _, err = pipe.Exec(); err == nil {
|
|
err = this.DeleteModelLogs(this.TableName, uid, bson.M{"_id": bson.M{"$in": ids}})
|
|
}
|
|
return
|
|
}
|
|
|
|
//批量删除数据
|
|
func (this *DBModel) BatchDelLists(uid string) (err error) {
|
|
var keys map[string]string
|
|
pipe := this.Redis.RedisPipe(context.TODO())
|
|
if keys, err = this.Redis.HGetAllToMapString(this.ukey(uid)); err == nil {
|
|
for _, v := range keys {
|
|
pipe.Delete(v)
|
|
}
|
|
pipe.Delete(this.ukey(uid))
|
|
pipe.Exec()
|
|
}
|
|
return
|
|
}
|