mongodb.go 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792
  1. package mongodb
  2. import (
  3. "context"
  4. "encoding/json"
  5. "fmt"
  6. "log"
  7. "math/big"
  8. "reflect"
  9. "runtime"
  10. "strconv"
  11. "strings"
  12. "time"
  13. "go.mongodb.org/mongo-driver/bson"
  14. "go.mongodb.org/mongo-driver/bson/primitive"
  15. "go.mongodb.org/mongo-driver/mongo"
  16. "go.mongodb.org/mongo-driver/mongo/options"
  17. )
  18. func NewMgo(addr, db string, size int) *MongodbSim {
  19. mgo := &MongodbSim{
  20. MongodbAddr: addr,
  21. Size: size,
  22. DbName: db,
  23. }
  24. mgo.InitPool()
  25. return mgo
  26. }
  27. type Bluk struct {
  28. ms *MgoSess
  29. writes []mongo.WriteModel
  30. }
  31. func (b *Bluk) Insert(doc interface{}) {
  32. write := mongo.NewInsertOneModel()
  33. write.SetDocument(doc)
  34. b.writes = append(b.writes, write)
  35. }
  36. func (b *Bluk) Update(doc ...interface{}) {
  37. write := mongo.NewUpdateOneModel()
  38. write.SetFilter(doc[0])
  39. ue := ObjToM(doc[1])
  40. autoUpdateTime(b.ms.db, b.ms.coll, ue)
  41. write.SetUpdate(ue)
  42. write.SetUpsert(false)
  43. b.writes = append(b.writes, write)
  44. }
  45. func (b *Bluk) UpdateAll(doc ...interface{}) {
  46. write := mongo.NewUpdateManyModel()
  47. write.SetFilter(doc[0])
  48. ue := ObjToM(doc[1])
  49. autoUpdateTime(b.ms.db, b.ms.coll, ue)
  50. write.SetUpdate(ue)
  51. write.SetUpsert(false)
  52. b.writes = append(b.writes, write)
  53. }
  54. func (b *Bluk) Upsert(doc ...interface{}) {
  55. write := mongo.NewUpdateOneModel()
  56. write.SetFilter(doc[0])
  57. ue := ObjToM(doc[1])
  58. autoUpdateTime(b.ms.db, b.ms.coll, ue)
  59. write.SetUpdate(ue)
  60. write.SetUpsert(true)
  61. b.writes = append(b.writes, write)
  62. }
  63. func (b *Bluk) Remove(doc interface{}) {
  64. write := mongo.NewDeleteOneModel()
  65. write.SetFilter(doc)
  66. b.writes = append(b.writes, write)
  67. }
  68. func (b *Bluk) RemoveAll(doc interface{}) {
  69. write := mongo.NewDeleteManyModel()
  70. write.SetFilter(doc)
  71. b.writes = append(b.writes, write)
  72. }
  73. func (b *Bluk) Run() (*mongo.BulkWriteResult, error) {
  74. return b.ms.M.C.Database(b.ms.db).Collection(b.ms.coll).BulkWrite(b.ms.M.Ctx, b.writes)
  75. }
  76. //
  77. type MgoIter struct {
  78. Cursor *mongo.Cursor
  79. Ctx context.Context
  80. }
  81. func (mt *MgoIter) Next(result interface{}) bool {
  82. if mt.Cursor != nil {
  83. if mt.Cursor.Next(mt.Ctx) {
  84. rType := reflect.TypeOf(result)
  85. rVal := reflect.ValueOf(result)
  86. if rType.Kind() == reflect.Ptr {
  87. rType = rType.Elem()
  88. rVal = rVal.Elem()
  89. }
  90. var err error
  91. if rType.Kind() == reflect.Map {
  92. r := make(map[string]interface{})
  93. err = mt.Cursor.Decode(&r)
  94. if rVal.CanSet() {
  95. rVal.Set(reflect.ValueOf(r))
  96. } else {
  97. for it := rVal.MapRange(); it.Next(); {
  98. rVal.SetMapIndex(it.Key(), reflect.Value{})
  99. }
  100. for it := reflect.ValueOf(r).MapRange(); it.Next(); {
  101. rVal.SetMapIndex(it.Key(), it.Value())
  102. }
  103. }
  104. } else {
  105. err = mt.Cursor.Decode(&result)
  106. }
  107. if err != nil {
  108. log.Println("mgo cur err", err.Error())
  109. mt.Cursor.Close(mt.Ctx)
  110. return false
  111. }
  112. return true
  113. } else {
  114. mt.Cursor.Close(mt.Ctx)
  115. return false
  116. }
  117. } else {
  118. return false
  119. }
  120. }
  121. //
  122. type MgoSess struct {
  123. db string
  124. coll string
  125. query interface{}
  126. sorts []string
  127. fields interface{}
  128. limit int64
  129. skip int64
  130. pipe []map[string]interface{}
  131. all interface{}
  132. M *MongodbSim
  133. }
  134. func (ms *MgoSess) DB(name string) *MgoSess {
  135. ms.db = name
  136. return ms
  137. }
  138. func (ms *MgoSess) C(name string) *MgoSess {
  139. ms.coll = name
  140. return ms
  141. }
  142. func (ms *MgoSess) Bulk() *Bluk {
  143. return &Bluk{ms: ms}
  144. }
  145. func (ms *MgoSess) Find(q interface{}) *MgoSess {
  146. if q == nil {
  147. q = map[string]interface{}{}
  148. }
  149. ms.query = q
  150. return ms
  151. }
  152. func (ms *MgoSess) FindId(_id interface{}) *MgoSess {
  153. ms.query = map[string]interface{}{"_id": _id}
  154. return ms
  155. }
  156. func (ms *MgoSess) Select(fields interface{}) *MgoSess {
  157. ms.fields = fields
  158. return ms
  159. }
  160. func (ms *MgoSess) Limit(limit int64) *MgoSess {
  161. ms.limit = limit
  162. return ms
  163. }
  164. func (ms *MgoSess) Skip(skip int64) *MgoSess {
  165. ms.skip = skip
  166. return ms
  167. }
  168. func (ms *MgoSess) Sort(sorts ...string) *MgoSess {
  169. ms.sorts = sorts
  170. return ms
  171. }
  172. func (ms *MgoSess) Pipe(p []map[string]interface{}) *MgoSess {
  173. ms.pipe = p
  174. return ms
  175. }
  176. func (ms *MgoSess) Insert(doc interface{}) error {
  177. _, err := ms.M.C.Database(ms.db).Collection(ms.coll).InsertOne(ms.M.Ctx, doc)
  178. return err
  179. }
  180. func (ms *MgoSess) Remove(filter interface{}) error {
  181. _, err := ms.M.C.Database(ms.db).Collection(ms.coll).DeleteOne(ms.M.Ctx, filter)
  182. return err
  183. }
  184. func (ms *MgoSess) RemoveId(_id interface{}) error {
  185. _, err := ms.M.C.Database(ms.db).Collection(ms.coll).DeleteOne(ms.M.Ctx, map[string]interface{}{"_id": _id})
  186. return err
  187. }
  188. func (ms *MgoSess) RemoveAll(filter interface{}) (*mongo.DeleteResult, error) {
  189. return ms.M.C.Database(ms.db).Collection(ms.coll).DeleteMany(ms.M.Ctx, filter)
  190. }
  191. func (ms *MgoSess) Upsert(filter, update interface{}) (*mongo.UpdateResult, error) {
  192. ct := options.Update()
  193. ct.SetUpsert(true)
  194. ue := ObjToM(update)
  195. autoUpdateTime(ms.db, ms.coll, ue)
  196. return ms.M.C.Database(ms.db).Collection(ms.coll).UpdateOne(ms.M.Ctx, filter, ue, ct)
  197. }
  198. func (ms *MgoSess) UpsertId(filter, update interface{}) (*mongo.UpdateResult, error) {
  199. ct := options.Update()
  200. ct.SetUpsert(true)
  201. ue := ObjToM(update)
  202. autoUpdateTime(ms.db, ms.coll, ue)
  203. return ms.M.C.Database(ms.db).Collection(ms.coll).UpdateOne(ms.M.Ctx, map[string]interface{}{"_id": filter}, ue, ct)
  204. }
  205. func (ms *MgoSess) UpdateId(filter, update interface{}) error {
  206. ue := ObjToM(update)
  207. autoUpdateTime(ms.db, ms.coll, ue)
  208. _, err := ms.M.C.Database(ms.db).Collection(ms.coll).UpdateOne(ms.M.Ctx, map[string]interface{}{"_id": filter}, ue)
  209. return err
  210. }
  211. func (ms *MgoSess) Update(filter, update interface{}) error {
  212. ue := ObjToM(update)
  213. autoUpdateTime(ms.db, ms.coll, ue)
  214. _, err := ms.M.C.Database(ms.db).Collection(ms.coll).UpdateOne(ms.M.Ctx, filter, ue)
  215. return err
  216. }
  217. func (ms *MgoSess) Count() (int64, error) {
  218. return ms.M.C.Database(ms.db).Collection(ms.coll).CountDocuments(ms.M.Ctx, ms.query)
  219. }
  220. func (ms *MgoSess) One(v *map[string]interface{}) {
  221. of := options.FindOne()
  222. of.SetProjection(ms.fields)
  223. sr := ms.M.C.Database(ms.db).Collection(ms.coll).FindOne(ms.M.Ctx, ms.query, of)
  224. if sr.Err() == nil {
  225. sr.Decode(&v)
  226. }
  227. }
  228. func (ms *MgoSess) All(v *[]map[string]interface{}) {
  229. cur, err := ms.M.C.Database(ms.db).Collection(ms.coll).Aggregate(ms.M.Ctx, ms.pipe)
  230. if err == nil && cur.Err() == nil {
  231. cur.All(ms.M.Ctx, v)
  232. }
  233. }
  234. func (ms *MgoSess) Iter() *MgoIter {
  235. it := &MgoIter{}
  236. coll := ms.M.C.Database(ms.db).Collection(ms.coll)
  237. var cur *mongo.Cursor
  238. var err error
  239. if ms.query != nil {
  240. find := options.Find()
  241. if ms.skip > 0 {
  242. find.SetSkip(ms.skip)
  243. }
  244. if ms.limit > 0 {
  245. find.SetLimit(ms.limit)
  246. }
  247. find.SetBatchSize(100)
  248. if len(ms.sorts) > 0 {
  249. sort := bson.D{}
  250. for _, k := range ms.sorts {
  251. switch k[:1] {
  252. case "-":
  253. sort = append(sort, bson.E{k[1:], -1})
  254. case "+":
  255. sort = append(sort, bson.E{k[1:], 1})
  256. default:
  257. sort = append(sort, bson.E{k, 1})
  258. }
  259. }
  260. find.SetSort(sort)
  261. }
  262. if ms.fields != nil {
  263. find.SetProjection(ms.fields)
  264. }
  265. cur, err = coll.Find(ms.M.Ctx, ms.query, find)
  266. if err != nil {
  267. log.Println("mgo find err", err.Error())
  268. }
  269. } else if ms.pipe != nil {
  270. aggregate := options.Aggregate()
  271. aggregate.SetBatchSize(100)
  272. cur, err = coll.Aggregate(ms.M.Ctx, ms.pipe, aggregate)
  273. if err != nil {
  274. log.Println("mgo aggregate err", err.Error())
  275. }
  276. }
  277. if err == nil {
  278. it.Cursor = cur
  279. it.Ctx = ms.M.Ctx
  280. }
  281. return it
  282. }
  283. type MongodbSim struct {
  284. MongodbAddr string
  285. Size int
  286. // MinSize int
  287. DbName string
  288. C *mongo.Client
  289. Ctx context.Context
  290. ShortCtx context.Context
  291. pool chan bool
  292. UserName string
  293. Password string
  294. ReplSet string
  295. }
  296. func (m *MongodbSim) GetMgoConn() *MgoSess {
  297. //m.Open()
  298. ms := &MgoSess{}
  299. ms.M = m
  300. return ms
  301. }
  302. func (m *MongodbSim) DestoryMongoConn(ms *MgoSess) {
  303. //m.Close()
  304. ms.M = nil
  305. ms = nil
  306. }
  307. func (m *MongodbSim) Destroy() {
  308. //m.Close()
  309. m.C.Disconnect(nil)
  310. m.C = nil
  311. }
  312. func (m *MongodbSim) InitPool() {
  313. opts := options.Client()
  314. registry := bson.NewRegistryBuilder().RegisterTypeMapEntry(bson.TypeArray, reflect.TypeOf([]interface{}{})).Build()
  315. opts.SetRegistry(registry)
  316. opts.SetConnectTimeout(3 * time.Second)
  317. opts.SetHosts(strings.Split(m.MongodbAddr, ","))
  318. //opts.ApplyURI("mongodb://" + m.MongodbAddr)
  319. opts.SetMaxPoolSize(uint64(m.Size))
  320. if m.UserName != "" && m.Password != "" {
  321. cre := options.Credential{
  322. Username: m.UserName,
  323. Password: m.Password,
  324. }
  325. opts.SetAuth(cre)
  326. }
  327. /*ms := strings.Split(m.MongodbAddr, ",")
  328. if m.ReplSet == "" && len(ms) > 1 {
  329. m.ReplSet = "qfws"
  330. }*/
  331. if m.ReplSet != "" {
  332. opts.SetReplicaSet(m.ReplSet)
  333. opts.SetDirect(false)
  334. }
  335. m.pool = make(chan bool, m.Size)
  336. opts.SetMaxConnIdleTime(2 * time.Hour)
  337. m.Ctx, _ = context.WithTimeout(context.Background(), 99999*time.Hour)
  338. m.ShortCtx, _ = context.WithTimeout(context.Background(), 1*time.Minute)
  339. client, err := mongo.Connect(m.ShortCtx, opts)
  340. if err != nil {
  341. log.Println("mgo init error:", err.Error())
  342. } else {
  343. m.C = client
  344. }
  345. }
  346. func (m *MongodbSim) Open() {
  347. m.pool <- true
  348. }
  349. func (m *MongodbSim) Close() {
  350. <-m.pool
  351. }
  352. func (m *MongodbSim) Save(c string, doc interface{}) string {
  353. defer catch()
  354. m.Open()
  355. defer m.Close()
  356. coll := m.C.Database(m.DbName).Collection(c)
  357. obj := ObjToM(doc)
  358. id := primitive.NewObjectID()
  359. (*obj)["_id"] = id
  360. _, err := coll.InsertOne(m.Ctx, obj)
  361. if nil != err {
  362. log.Println("SaveError", err)
  363. return ""
  364. }
  365. return id.Hex()
  366. }
  367. //原_id不变
  368. func (m *MongodbSim) SaveByOriID(c string, doc interface{}) bool {
  369. defer catch()
  370. m.Open()
  371. defer m.Close()
  372. coll := m.C.Database(m.DbName).Collection(c)
  373. _, err := coll.InsertOne(m.Ctx, ObjToM(doc))
  374. if nil != err {
  375. log.Println("SaveByOriIDError", err)
  376. return false
  377. }
  378. return true
  379. }
  380. //批量插入
  381. func (m *MongodbSim) SaveBulk(c string, doc ...map[string]interface{}) bool {
  382. defer catch()
  383. m.Open()
  384. defer m.Close()
  385. coll := m.C.Database(m.DbName).Collection(c)
  386. var writes []mongo.WriteModel
  387. for _, d := range doc {
  388. write := mongo.NewInsertOneModel()
  389. write.SetDocument(d)
  390. writes = append(writes, write)
  391. }
  392. br, e := coll.BulkWrite(m.Ctx, writes)
  393. if e != nil {
  394. b := strings.Index(e.Error(), "duplicate") > -1
  395. log.Println("mgo savebulk error:", e.Error())
  396. if br != nil {
  397. log.Println("mgo savebulk size:", br.InsertedCount)
  398. }
  399. return b
  400. }
  401. return true
  402. }
  403. //批量插入
  404. func (m *MongodbSim) SaveBulkInterface(c string, doc ...interface{}) bool {
  405. defer catch()
  406. m.Open()
  407. defer m.Close()
  408. coll := m.C.Database(m.DbName).Collection(c)
  409. var writes []mongo.WriteModel
  410. for _, d := range doc {
  411. write := mongo.NewInsertOneModel()
  412. write.SetDocument(d)
  413. writes = append(writes, write)
  414. }
  415. br, e := coll.BulkWrite(m.Ctx, writes)
  416. if e != nil {
  417. b := strings.Index(e.Error(), "duplicate") > -1
  418. log.Println("mgo SaveBulkInterface error:", e.Error())
  419. if br != nil {
  420. log.Println("mgo SaveBulkInterface size:", br.InsertedCount)
  421. }
  422. return b
  423. }
  424. return true
  425. }
  426. //按条件统计
  427. func (m *MongodbSim) Count(c string, q interface{}) int {
  428. r, _ := m.CountByErr(c, q)
  429. return r
  430. }
  431. //统计
  432. func (m *MongodbSim) CountByErr(c string, q interface{}) (int, error) {
  433. defer catch()
  434. m.Open()
  435. defer m.Close()
  436. var res int64
  437. var err error
  438. if filter := ObjToM(q); filter != nil && len(*filter) > 0 {
  439. res, err = m.C.Database(m.DbName).Collection(c).CountDocuments(m.Ctx, filter)
  440. } else {
  441. res, err = m.C.Database(m.DbName).Collection(c).EstimatedDocumentCount(m.Ctx)
  442. }
  443. if err != nil {
  444. log.Println("统计错误", err.Error())
  445. return 0, err
  446. } else {
  447. return int(res), nil
  448. }
  449. }
  450. //按条件删除
  451. func (m *MongodbSim) Delete(c string, q interface{}) int64 {
  452. defer catch()
  453. m.Open()
  454. defer m.Close()
  455. res, err := m.C.Database(m.DbName).Collection(c).DeleteMany(m.Ctx, ObjToM(q))
  456. if err != nil && res == nil {
  457. log.Println("删除错误", err.Error())
  458. }
  459. return res.DeletedCount
  460. }
  461. //删除对象
  462. func (m *MongodbSim) Del(c string, q interface{}) bool {
  463. defer catch()
  464. m.Open()
  465. defer m.Close()
  466. _, err := m.C.Database(m.DbName).Collection(c).DeleteMany(m.Ctx, ObjToM(q))
  467. if err != nil {
  468. log.Println("删除错误", err.Error())
  469. return false
  470. }
  471. return true
  472. }
  473. //按条件更新
  474. func (m *MongodbSim) Update(c string, q, u interface{}, upsert bool, multi bool) bool {
  475. defer catch()
  476. m.Open()
  477. defer m.Close()
  478. ct := options.Update()
  479. if upsert {
  480. ct.SetUpsert(true)
  481. }
  482. coll := m.C.Database(m.DbName).Collection(c)
  483. ue := ObjToM(u)
  484. autoUpdateTime(m.DbName, c, ue)
  485. var err error
  486. if multi {
  487. _, err = coll.UpdateMany(m.Ctx, ObjToM(q), ue, ct)
  488. } else {
  489. _, err = coll.UpdateOne(m.Ctx, ObjToM(q), ue, ct)
  490. }
  491. if err != nil {
  492. log.Println("更新错误", err.Error())
  493. return false
  494. }
  495. return true
  496. }
  497. func (m *MongodbSim) UpdateById(c string, id interface{}, set interface{}) bool {
  498. defer catch()
  499. m.Open()
  500. defer m.Close()
  501. q := make(map[string]interface{})
  502. if sid, ok := id.(string); ok {
  503. q["_id"], _ = primitive.ObjectIDFromHex(sid)
  504. } else {
  505. q["_id"] = id
  506. }
  507. ue := ObjToM(set)
  508. autoUpdateTime(m.DbName, c, ue)
  509. _, err := m.C.Database(m.DbName).Collection(c).UpdateOne(m.Ctx, q, ue)
  510. if nil != err {
  511. log.Println("UpdateByIdError", err)
  512. return false
  513. }
  514. return true
  515. }
  516. //批量更新
  517. func (m *MongodbSim) UpdateBulkAll(db, c string, doc ...[]map[string]interface{}) bool {
  518. return m.NewUpdateBulk(db, c, false, false, doc...)
  519. }
  520. func (m *MongodbSim) UpdateBulk(c string, doc ...[]map[string]interface{}) bool {
  521. return m.UpdateBulkAll(m.DbName, c, doc...)
  522. }
  523. //批量插入
  524. func (m *MongodbSim) UpSertBulk(c string, doc ...[]map[string]interface{}) bool {
  525. return m.NewUpdateBulk(m.DbName, c, true, false, doc...)
  526. }
  527. //批量插入
  528. func (m *MongodbSim) UpSertMultiBulk(c string, upsert, multi bool, doc ...[]map[string]interface{}) bool {
  529. return m.NewUpdateBulk(m.DbName, c, upsert, multi, doc...)
  530. }
  531. //批量插入
  532. func (m *MongodbSim) NewUpdateBulk(db, c string, upsert, multi bool, doc ...[]map[string]interface{}) bool {
  533. defer catch()
  534. m.Open()
  535. defer m.Close()
  536. coll := m.C.Database(db).Collection(c)
  537. var writes []mongo.WriteModel
  538. for _, d := range doc {
  539. if multi {
  540. write := mongo.NewUpdateManyModel()
  541. write.SetFilter(d[0])
  542. ue := ObjToM(d[1])
  543. autoUpdateTime(m.DbName, c, ue)
  544. write.SetUpdate(ue)
  545. write.SetUpsert(upsert)
  546. writes = append(writes, write)
  547. } else {
  548. write := mongo.NewUpdateOneModel()
  549. write.SetFilter(d[0])
  550. ue := ObjToM(d[1])
  551. autoUpdateTime(m.DbName, c, ue)
  552. write.SetUpdate(ue)
  553. write.SetUpsert(upsert)
  554. writes = append(writes, write)
  555. }
  556. }
  557. br, e := coll.BulkWrite(m.Ctx, writes)
  558. if e != nil {
  559. log.Println("mgo upsert error:", e.Error())
  560. return br == nil || br.UpsertedCount == 0
  561. }
  562. // else {
  563. // if r.UpsertedCount != int64(len(doc)) {
  564. // log.Println("mgo upsert uncomplete:uc/dc", r.UpsertedCount, len(doc))
  565. // }
  566. // return true
  567. // }
  568. return true
  569. }
  570. //查询单条对象
  571. func (m *MongodbSim) FindOne(c string, query interface{}) (*map[string]interface{}, bool) {
  572. return m.FindOneByField(c, query, nil)
  573. }
  574. //查询单条对象
  575. func (m *MongodbSim) FindOneByField(c string, query interface{}, fields interface{}) (*map[string]interface{}, bool) {
  576. defer catch()
  577. res, ok := m.Find(c, query, nil, fields, true, -1, -1)
  578. if nil != res && len(*res) > 0 {
  579. return &((*res)[0]), ok
  580. }
  581. return nil, ok
  582. }
  583. //查询单条对象
  584. func (m *MongodbSim) FindById(c string, query string, fields interface{}) (*map[string]interface{}, bool) {
  585. defer catch()
  586. m.Open()
  587. defer m.Close()
  588. of := options.FindOne()
  589. of.SetProjection(ObjToOth(fields))
  590. res := make(map[string]interface{})
  591. _id, err := primitive.ObjectIDFromHex(query)
  592. if err != nil {
  593. log.Println("_id error", err)
  594. return &res, true
  595. }
  596. sr := m.C.Database(m.DbName).Collection(c).FindOne(m.Ctx, map[string]interface{}{"_id": _id}, of)
  597. if sr.Err() == nil {
  598. sr.Decode(&res)
  599. }
  600. return &res, true
  601. }
  602. //底层查询方法
  603. func (m *MongodbSim) Find(c string, query interface{}, order interface{}, fields interface{}, single bool, start int, limit int) (*[]map[string]interface{}, bool) {
  604. defer catch()
  605. m.Open()
  606. defer m.Close()
  607. var res []map[string]interface{}
  608. coll := m.C.Database(m.DbName).Collection(c)
  609. if single {
  610. res = make([]map[string]interface{}, 1)
  611. of := options.FindOne()
  612. of.SetProjection(ObjToOth(fields))
  613. of.SetSort(ObjToM(order))
  614. if sr := coll.FindOne(m.Ctx, ObjToM(query), of); sr.Err() == nil {
  615. sr.Decode(&res[0])
  616. }
  617. } else {
  618. res = []map[string]interface{}{}
  619. of := options.Find()
  620. of.SetProjection(ObjToOth(fields))
  621. of.SetSort(ObjToM(order))
  622. if start > -1 {
  623. of.SetSkip(int64(start))
  624. of.SetLimit(int64(limit))
  625. }
  626. cur, err := coll.Find(m.Ctx, ObjToM(query), of)
  627. if err == nil && cur.Err() == nil {
  628. cur.All(m.Ctx, &res)
  629. }
  630. }
  631. return &res, true
  632. }
  633. func ObjToOth(query interface{}) *bson.M {
  634. return ObjToMQ(query, false)
  635. }
  636. func ObjToM(query interface{}) *bson.M {
  637. return ObjToMQ(query, true)
  638. }
  639. //obj(string,M)转M,查询用到
  640. func ObjToMQ(query interface{}, isQuery bool) *bson.M {
  641. data := make(bson.M)
  642. defer catch()
  643. if s2, ok2 := query.(*map[string]interface{}); ok2 {
  644. data = bson.M(*s2)
  645. } else if s3, ok3 := query.(*bson.M); ok3 {
  646. return s3
  647. } else if s3, ok3 := query.(*primitive.M); ok3 {
  648. return s3
  649. } else if s, ok := query.(string); ok {
  650. json.Unmarshal([]byte(strings.Replace(s, "'", "\"", -1)), &data)
  651. if ss, oks := data["_id"]; oks && isQuery {
  652. switch ss.(type) {
  653. case string:
  654. data["_id"], _ = primitive.ObjectIDFromHex(ss.(string))
  655. case map[string]interface{}:
  656. tmp := ss.(map[string]interface{})
  657. for k, v := range tmp {
  658. tmp[k], _ = primitive.ObjectIDFromHex(v.(string))
  659. }
  660. data["_id"] = tmp
  661. }
  662. }
  663. } else if s1, ok1 := query.(map[string]interface{}); ok1 {
  664. data = s1
  665. } else if s4, ok4 := query.(bson.M); ok4 {
  666. data = s4
  667. } else if s4, ok4 := query.(primitive.M); ok4 {
  668. data = s4
  669. } else {
  670. data = nil
  671. }
  672. return &data
  673. }
  674. func intAllDef(num interface{}, defaultNum int) int {
  675. if i, ok := num.(int); ok {
  676. return int(i)
  677. } else if i0, ok0 := num.(int32); ok0 {
  678. return int(i0)
  679. } else if i1, ok1 := num.(float64); ok1 {
  680. return int(i1)
  681. } else if i2, ok2 := num.(int64); ok2 {
  682. return int(i2)
  683. } else if i3, ok3 := num.(float32); ok3 {
  684. return int(i3)
  685. } else if i4, ok4 := num.(string); ok4 {
  686. in, _ := strconv.Atoi(i4)
  687. return int(in)
  688. } else if i5, ok5 := num.(int16); ok5 {
  689. return int(i5)
  690. } else if i6, ok6 := num.(int8); ok6 {
  691. return int(i6)
  692. } else if i7, ok7 := num.(*big.Int); ok7 {
  693. in, _ := strconv.Atoi(fmt.Sprint(i7))
  694. return int(in)
  695. } else if i8, ok8 := num.(*big.Float); ok8 {
  696. in, _ := strconv.Atoi(fmt.Sprint(i8))
  697. return int(in)
  698. } else {
  699. return defaultNum
  700. }
  701. }
  702. //出错拦截
  703. func catch() {
  704. if r := recover(); r != nil {
  705. log.Println(r)
  706. for skip := 0; ; skip++ {
  707. _, file, line, ok := runtime.Caller(skip)
  708. if !ok {
  709. break
  710. }
  711. go log.Printf("%v,%v\n", file, line)
  712. }
  713. }
  714. }
  715. //根据bsonID转string
  716. func BsonIdToSId(uid interface{}) string {
  717. if uid == nil {
  718. return ""
  719. } else if u, ok := uid.(string); ok {
  720. return u
  721. } else if u, ok := uid.(primitive.ObjectID); ok {
  722. return u.Hex()
  723. } else {
  724. return ""
  725. }
  726. }
  727. func StringTOBsonId(id string) (bid primitive.ObjectID) {
  728. defer catch()
  729. if id != "" {
  730. bid, _ = primitive.ObjectIDFromHex(id)
  731. }
  732. return
  733. }
  734. func ToObjectIds(ids []string) []primitive.ObjectID {
  735. _ids := []primitive.ObjectID{}
  736. for _, v := range ids {
  737. _id, _ := primitive.ObjectIDFromHex(v)
  738. _ids = append(_ids, _id)
  739. }
  740. return _ids
  741. }
  742. //自动添加更新时间
  743. func autoUpdateTime(db, coll string, ue *bson.M) {
  744. if db == "qfw" && coll == "user" {
  745. set := ObjToM((*ue)["$set"])
  746. if *set == nil {
  747. set = &bson.M{}
  748. }
  749. (*set)["auto_updatetime"] = time.Now().Unix()
  750. (*ue)["$set"] = set
  751. }
  752. }