cachedmodel.go 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242
  1. package mongoc
  2. import (
  3. "log"
  4. "github.com/globalsign/mgo"
  5. "github.com/tal-tech/go-zero/core/stores/cache"
  6. "github.com/tal-tech/go-zero/core/stores/internal"
  7. "github.com/tal-tech/go-zero/core/stores/mongo"
  8. "github.com/tal-tech/go-zero/core/stores/redis"
  9. )
  10. type Model struct {
  11. *mongo.Model
  12. cache internal.Cache
  13. generateCollection func(*mgo.Session) *cachedCollection
  14. }
  15. func MustNewNodeModel(url, database, collection string, rds *redis.Redis, opts ...cache.Option) *Model {
  16. model, err := NewNodeModel(url, database, collection, rds, opts...)
  17. if err != nil {
  18. log.Fatal(err)
  19. }
  20. return model
  21. }
  22. func MustNewModel(url, database, collection string, c cache.CacheConf, opts ...cache.Option) *Model {
  23. model, err := NewModel(url, database, collection, c, opts...)
  24. if err != nil {
  25. log.Fatal(err)
  26. }
  27. return model
  28. }
  29. func NewNodeModel(url, database, collection string, rds *redis.Redis, opts ...cache.Option) (*Model, error) {
  30. c := internal.NewCacheNode(rds, sharedCalls, stats, mgo.ErrNotFound, opts...)
  31. return createModel(url, database, collection, c, func(collection mongo.Collection) *cachedCollection {
  32. return newCollection(collection, c)
  33. })
  34. }
  35. func NewModel(url, database, collection string, conf cache.CacheConf, opts ...cache.Option) (*Model, error) {
  36. c := internal.NewCache(conf, sharedCalls, stats, mgo.ErrNotFound, opts...)
  37. return createModel(url, database, collection, c, func(collection mongo.Collection) *cachedCollection {
  38. return newCollection(collection, c)
  39. })
  40. }
  41. func (mm *Model) Count(query interface{}) (int, error) {
  42. return mm.executeInt(func(c *cachedCollection) (int, error) {
  43. return c.Count(query)
  44. })
  45. }
  46. func (mm *Model) DelCache(keys ...string) error {
  47. return mm.cache.DelCache(keys...)
  48. }
  49. func (mm *Model) GetCache(key string, v interface{}) error {
  50. return mm.cache.GetCache(key, v)
  51. }
  52. func (mm *Model) GetCollection(session *mgo.Session) *cachedCollection {
  53. return mm.generateCollection(session)
  54. }
  55. func (mm *Model) FindAllNoCache(v interface{}, query interface{}, opts ...QueryOption) error {
  56. return mm.execute(func(c *cachedCollection) error {
  57. return c.FindAllNoCache(v, query, opts...)
  58. })
  59. }
  60. func (mm *Model) FindOne(v interface{}, key string, query interface{}) error {
  61. return mm.execute(func(c *cachedCollection) error {
  62. return c.FindOne(v, key, query)
  63. })
  64. }
  65. func (mm *Model) FindOneNoCache(v interface{}, query interface{}) error {
  66. return mm.execute(func(c *cachedCollection) error {
  67. return c.FindOneNoCache(v, query)
  68. })
  69. }
  70. func (mm *Model) FindOneId(v interface{}, key string, id interface{}) error {
  71. return mm.execute(func(c *cachedCollection) error {
  72. return c.FindOneId(v, key, id)
  73. })
  74. }
  75. func (mm *Model) FindOneIdNoCache(v interface{}, id interface{}) error {
  76. return mm.execute(func(c *cachedCollection) error {
  77. return c.FindOneIdNoCache(v, id)
  78. })
  79. }
  80. func (mm *Model) Insert(docs ...interface{}) error {
  81. return mm.execute(func(c *cachedCollection) error {
  82. return c.Insert(docs...)
  83. })
  84. }
  85. func (mm *Model) Pipe(pipeline interface{}) (mongo.Pipe, error) {
  86. return mm.pipe(func(c *cachedCollection) mongo.Pipe {
  87. return c.Pipe(pipeline)
  88. })
  89. }
  90. func (mm *Model) Remove(selector interface{}, keys ...string) error {
  91. return mm.execute(func(c *cachedCollection) error {
  92. return c.Remove(selector, keys...)
  93. })
  94. }
  95. func (mm *Model) RemoveNoCache(selector interface{}) error {
  96. return mm.execute(func(c *cachedCollection) error {
  97. return c.RemoveNoCache(selector)
  98. })
  99. }
  100. func (mm *Model) RemoveAll(selector interface{}, keys ...string) (*mgo.ChangeInfo, error) {
  101. return mm.change(func(c *cachedCollection) (*mgo.ChangeInfo, error) {
  102. return c.RemoveAll(selector, keys...)
  103. })
  104. }
  105. func (mm *Model) RemoveAllNoCache(selector interface{}) (*mgo.ChangeInfo, error) {
  106. return mm.change(func(c *cachedCollection) (*mgo.ChangeInfo, error) {
  107. return c.RemoveAllNoCache(selector)
  108. })
  109. }
  110. func (mm *Model) RemoveId(id interface{}, keys ...string) error {
  111. return mm.execute(func(c *cachedCollection) error {
  112. return c.RemoveId(id, keys...)
  113. })
  114. }
  115. func (mm *Model) RemoveIdNoCache(id interface{}) error {
  116. return mm.execute(func(c *cachedCollection) error {
  117. return c.RemoveIdNoCache(id)
  118. })
  119. }
  120. func (mm *Model) SetCache(key string, v interface{}) error {
  121. return mm.cache.SetCache(key, v)
  122. }
  123. func (mm *Model) Update(selector, update interface{}, keys ...string) error {
  124. return mm.execute(func(c *cachedCollection) error {
  125. return c.Update(selector, update, keys...)
  126. })
  127. }
  128. func (mm *Model) UpdateNoCache(selector, update interface{}) error {
  129. return mm.execute(func(c *cachedCollection) error {
  130. return c.UpdateNoCache(selector, update)
  131. })
  132. }
  133. func (mm *Model) UpdateId(id, update interface{}, keys ...string) error {
  134. return mm.execute(func(c *cachedCollection) error {
  135. return c.UpdateId(id, update, keys...)
  136. })
  137. }
  138. func (mm *Model) UpdateIdNoCache(id, update interface{}) error {
  139. return mm.execute(func(c *cachedCollection) error {
  140. return c.UpdateIdNoCache(id, update)
  141. })
  142. }
  143. func (mm *Model) Upsert(selector, update interface{}, keys ...string) (*mgo.ChangeInfo, error) {
  144. return mm.change(func(c *cachedCollection) (*mgo.ChangeInfo, error) {
  145. return c.Upsert(selector, update, keys...)
  146. })
  147. }
  148. func (mm *Model) UpsertNoCache(selector, update interface{}) (*mgo.ChangeInfo, error) {
  149. return mm.change(func(c *cachedCollection) (*mgo.ChangeInfo, error) {
  150. return c.UpsertNoCache(selector, update)
  151. })
  152. }
  153. func (mm *Model) change(fn func(c *cachedCollection) (*mgo.ChangeInfo, error)) (*mgo.ChangeInfo, error) {
  154. session, err := mm.TakeSession()
  155. if err != nil {
  156. return nil, err
  157. }
  158. defer mm.PutSession(session)
  159. return fn(mm.GetCollection(session))
  160. }
  161. func (mm *Model) execute(fn func(c *cachedCollection) error) error {
  162. session, err := mm.TakeSession()
  163. if err != nil {
  164. return err
  165. }
  166. defer mm.PutSession(session)
  167. return fn(mm.GetCollection(session))
  168. }
  169. func (mm *Model) executeInt(fn func(c *cachedCollection) (int, error)) (int, error) {
  170. session, err := mm.TakeSession()
  171. if err != nil {
  172. return 0, err
  173. }
  174. defer mm.PutSession(session)
  175. return fn(mm.GetCollection(session))
  176. }
  177. func (mm *Model) pipe(fn func(c *cachedCollection) mongo.Pipe) (mongo.Pipe, error) {
  178. session, err := mm.TakeSession()
  179. if err != nil {
  180. return nil, err
  181. }
  182. defer mm.PutSession(session)
  183. return fn(mm.GetCollection(session)), nil
  184. }
  185. func createModel(url, database, collection string, c internal.Cache,
  186. create func(mongo.Collection) *cachedCollection) (*Model, error) {
  187. model, err := mongo.NewModel(url, database, collection)
  188. if err != nil {
  189. return nil, err
  190. }
  191. return &Model{
  192. Model: model,
  193. cache: c,
  194. generateCollection: func(session *mgo.Session) *cachedCollection {
  195. collection := model.GetCollection(session)
  196. return create(collection)
  197. },
  198. }, nil
  199. }