|
|
@@ -0,0 +1,378 @@
|
|
|
+package mongodbutil
|
|
|
+
|
|
|
+import (
|
|
|
+ "encoding/json"
|
|
|
+ "errors"
|
|
|
+ "fmt"
|
|
|
+ "log"
|
|
|
+ "qfw/util"
|
|
|
+ "runtime"
|
|
|
+ "strings"
|
|
|
+
|
|
|
+ . "gopkg.in/mgo.v2/bson"
|
|
|
+)
|
|
|
+
|
|
|
+//统计
|
|
|
+func (m *Pool) Count(c string, query interface{}) int {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ //log.Println("count:", m.Size, m.MongodbAddr, m.DB, sess, m.GetMgoConn(), m)
|
|
|
+ var n int = 0
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var err error
|
|
|
+ n, err = coll.Find(ObjToM(query)).Count()
|
|
|
+ if nil != err {
|
|
|
+ log.Println("CountError", err)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return n
|
|
|
+}
|
|
|
+
|
|
|
+//统计
|
|
|
+func (m *Pool) CountByErr(c string, query interface{}) (int, error) {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ //log.Println("count:", m.Size, m.MongodbAddr, m.DB, sess, m.GetMgoConn(), m)
|
|
|
+ var n int = 0
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var err error
|
|
|
+ n, err = coll.Find(ObjToM(query)).Count()
|
|
|
+ if nil != err {
|
|
|
+ return 0, err
|
|
|
+ } else {
|
|
|
+ return n, nil
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+ return n, errors.New("no sess")
|
|
|
+}
|
|
|
+
|
|
|
+func (m *Pool) Update(c string, query interface{}, set interface{}, upsert bool, multi bool) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := true
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var err error
|
|
|
+ if upsert {
|
|
|
+ _, err = coll.Upsert(ObjToM(query), ObjToM(set))
|
|
|
+ } else {
|
|
|
+ if multi {
|
|
|
+ _, err = coll.UpdateAll(ObjToM(query), ObjToM(set))
|
|
|
+ } else {
|
|
|
+ err = coll.Update(ObjToM(query), ObjToM(set))
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if nil != err {
|
|
|
+ log.Println("UpdateError", err)
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+func (m *Pool) UpdateById(c string, id interface{}, set interface{}) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := false
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var q interface{}
|
|
|
+ if sid, ok := id.(string); ok {
|
|
|
+ q = M{"_id": util.StringTOBsonId(sid)}
|
|
|
+ } else {
|
|
|
+ q = M{"_id": id}
|
|
|
+ }
|
|
|
+ err := coll.Update(q, ObjToM(set))
|
|
|
+ if nil != err {
|
|
|
+ log.Println("UpdateByIdError", err)
|
|
|
+ b = false
|
|
|
+ } else {
|
|
|
+ b = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+//批量更新
|
|
|
+
|
|
|
+func (m *Pool) UpdateBulkAll(db, c string, doc ...[]map[string]interface{}) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := true
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(db).C(c)
|
|
|
+ bulk := coll.Bulk()
|
|
|
+ for _, v := range doc {
|
|
|
+ if len(v) == 2 {
|
|
|
+ bulk.Update(v[0], v[1])
|
|
|
+ }
|
|
|
+ }
|
|
|
+ _, err := bulk.Run()
|
|
|
+ if nil != err {
|
|
|
+ log.Println("BulkError", err)
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+func (m *Pool) UpdateBulk(c string, doc ...[]map[string]interface{}) bool {
|
|
|
+ return m.UpdateBulkAll(m.DB, c, doc...)
|
|
|
+}
|
|
|
+
|
|
|
+//批量更新
|
|
|
+func (m *Pool) UpSertBulk(c string, doc ...[]map[string]interface{}) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := true
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ bulk := coll.Bulk()
|
|
|
+ for _, v := range doc {
|
|
|
+ if len(v) == 2 {
|
|
|
+ bulk.Upsert(v[0], v[1])
|
|
|
+ }
|
|
|
+ }
|
|
|
+ _, err := bulk.Run()
|
|
|
+ if nil != err {
|
|
|
+ log.Println("BulkUpsertError", err)
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+//批量插入
|
|
|
+func (m *Pool) SaveBulk(c string, doc ...map[string]interface{}) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := true
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ bulk := coll.Bulk()
|
|
|
+ for _, v := range doc {
|
|
|
+ bulk.Insert(v)
|
|
|
+ }
|
|
|
+ _, err := bulk.Run()
|
|
|
+ if nil != err {
|
|
|
+ log.Println("BulkError", err)
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+func (m *Pool) Save(c string, doc interface{}) string {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ obj := ObjToM(doc)
|
|
|
+ id := NewObjectId()
|
|
|
+ (*obj)["_id"] = id
|
|
|
+ err := coll.Insert(obj)
|
|
|
+ if nil != err {
|
|
|
+ log.Println("SaveError", err)
|
|
|
+ return ""
|
|
|
+ }
|
|
|
+ return (strings.Split(fmt.Sprintf("%s", id), `"`)[1])
|
|
|
+ }
|
|
|
+ return ""
|
|
|
+}
|
|
|
+
|
|
|
+//查询单条对象
|
|
|
+func (m *Pool) FindOne(c string, query interface{}) (*map[string]interface{}, bool) {
|
|
|
+ return m.FindOneByField(c, query, nil)
|
|
|
+}
|
|
|
+
|
|
|
+//查询单条对象
|
|
|
+func (m *Pool) FindOneByField(c string, query interface{}, fields interface{}) (*map[string]interface{}, bool) {
|
|
|
+ defer util.Catch()
|
|
|
+ res, ok := m.Find(c, query, nil, fields, true, -1, -1)
|
|
|
+ if nil != res && len(*res) > 0 {
|
|
|
+ return &((*res)[0]), ok
|
|
|
+ }
|
|
|
+ return nil, ok
|
|
|
+}
|
|
|
+
|
|
|
+//查询单条对象
|
|
|
+func (m *Pool) FindById(c string, query string, fields interface{}) (*map[string]interface{}, bool) {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ var res map[string]interface{}
|
|
|
+ b := false
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ res = make(map[string]interface{})
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var err error
|
|
|
+ err = coll.FindId(ObjectIdHex(query)).Select(ObjToOth(fields)).One(&res)
|
|
|
+ if nil != err {
|
|
|
+ log.Println("FindByIdError", err,query)
|
|
|
+ }
|
|
|
+ b = true
|
|
|
+ }
|
|
|
+ return &res, b
|
|
|
+}
|
|
|
+
|
|
|
+//底层查询方法
|
|
|
+func (m *Pool) Find(c string, query interface{}, order interface{}, fields interface{}, single bool, start int, limit int) (*[]map[string]interface{}, bool) {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ var res []map[string]interface{}
|
|
|
+ b := false
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ res = make([]map[string]interface{}, 1)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ var err error
|
|
|
+ if single {
|
|
|
+ err = coll.Find(ObjToM(query)).Select(ObjToOth(fields)).Sort(ObjToArr(order)...).One(&res[0])
|
|
|
+ } else if start > -1 {
|
|
|
+ err = coll.Find(ObjToM(query)).Select(ObjToOth(fields)).Sort(ObjToArr(order)...).Skip(start).Limit(limit).All(&res)
|
|
|
+ } else {
|
|
|
+ err = coll.Find(ObjToM(query)).Select(ObjToOth(fields)).Sort(ObjToArr(order)...).All(&res)
|
|
|
+ }
|
|
|
+ if nil != err {
|
|
|
+ //log.Println("FindError", err)
|
|
|
+ }
|
|
|
+ b = true
|
|
|
+ }
|
|
|
+ return &res, b
|
|
|
+}
|
|
|
+
|
|
|
+//删除对象
|
|
|
+func (m *Pool) Del(c string, query interface{}) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := false
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ _, err := coll.RemoveAll(ObjToM(query))
|
|
|
+ if nil != err {
|
|
|
+ log.Println("DelError", err)
|
|
|
+ b = false
|
|
|
+ } else {
|
|
|
+ b = true
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
|
|
+
|
|
|
+func (m *Pool) GetObjectId(str string) ObjectId {
|
|
|
+ return ObjectIdHex(str)
|
|
|
+}
|
|
|
+
|
|
|
+func ObjToOth(query interface{}) *M {
|
|
|
+ return ObjToMQ(query, false)
|
|
|
+}
|
|
|
+func ObjToM(query interface{}) *M {
|
|
|
+ return ObjToMQ(query, true)
|
|
|
+}
|
|
|
+
|
|
|
+//obj(string,M)转M,查询用到
|
|
|
+func ObjToMQ(query interface{}, isQuery bool) *M {
|
|
|
+ data := make(M)
|
|
|
+ defer func() {
|
|
|
+ if r := recover(); r != nil {
|
|
|
+ log.Println("[E]", r)
|
|
|
+ for skip := 1; ; skip++ {
|
|
|
+ _, file, line, ok := runtime.Caller(skip)
|
|
|
+ if !ok {
|
|
|
+ break
|
|
|
+ }
|
|
|
+ go log.Printf("%v,%v\n", file, line)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }()
|
|
|
+ if s2, ok2 := query.(*map[string]interface{}); ok2 {
|
|
|
+ data = M(*s2)
|
|
|
+ } else if s3, ok3 := query.(*M); ok3 {
|
|
|
+ return s3
|
|
|
+ } else if s, ok := query.(string); ok {
|
|
|
+ json.Unmarshal([]byte(strings.Replace(s, "'", "\"", -1)), &data)
|
|
|
+ if ss, oks := data["_id"]; oks && isQuery {
|
|
|
+ switch ss.(type) {
|
|
|
+ case string:
|
|
|
+ data["_id"] = ObjectIdHex(ss.(string))
|
|
|
+ case map[string]interface{}:
|
|
|
+ tmp := ss.(map[string]interface{})
|
|
|
+ for k, v := range tmp {
|
|
|
+ tmp[k] = ObjectIdHex(v.(string))
|
|
|
+ }
|
|
|
+ data["_id"] = tmp
|
|
|
+ }
|
|
|
+
|
|
|
+ }
|
|
|
+ } else if s1, ok1 := query.(map[string]interface{}); ok1 {
|
|
|
+ data = s1
|
|
|
+ } else if s4, ok4 := query.(M); ok4 {
|
|
|
+ data = s4
|
|
|
+ } else {
|
|
|
+ data = nil
|
|
|
+ }
|
|
|
+ return &data
|
|
|
+}
|
|
|
+
|
|
|
+//对象转数组
|
|
|
+func ObjToArr(obj interface{}) []string {
|
|
|
+ if s, ok := obj.(string); ok {
|
|
|
+ if strings.ContainsAny(s, "{") {
|
|
|
+ //暂时简单支持此种写法
|
|
|
+ var temp = make(M)
|
|
|
+ var str = []string{}
|
|
|
+ json.Unmarshal([]byte(s), &temp)
|
|
|
+ for k, v := range temp {
|
|
|
+ m := util.IntAll(v)
|
|
|
+ if m > 0 {
|
|
|
+ str = append(str, k)
|
|
|
+ } else {
|
|
|
+ str = append(str, "-"+k)
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return str
|
|
|
+ } else {
|
|
|
+ return strings.Split(s, ",")
|
|
|
+ }
|
|
|
+ } else if s1, ok1 := obj.([]string); ok1 {
|
|
|
+ return s1
|
|
|
+ } else {
|
|
|
+ return []string{}
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+//删除表
|
|
|
+func (m *Pool) DelColl(c string) bool {
|
|
|
+ defer util.Catch()
|
|
|
+ sess := m.Get()
|
|
|
+ b := true
|
|
|
+ if sess != nil {
|
|
|
+ defer m.Close(sess)
|
|
|
+ coll := sess.DB(m.DB).C(c)
|
|
|
+ err := coll.DropCollection()
|
|
|
+ if err != nil {
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ } else {
|
|
|
+ b = false
|
|
|
+ }
|
|
|
+ return b
|
|
|
+}
|
