mongodb.go 20 KB

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