/
session.go
218 lines (175 loc) · 5.83 KB
/
session.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
package daos
import (
"reflect"
"github.com/globalsign/mgo"
"github.com/globalsign/mgo/bson"
"github.com/tomochain/dex-server/app"
"github.com/tomochain/dex-server/utils"
)
// Database struct contains the pointer to mgo.session
// It is a wrapper over mgo to help utilize mgo connection pool
type Database struct {
Session *mgo.Session
}
// Global instance of Database struct for singleton use
var db *Database
var logger = utils.Logger
// InitSession initializes a new session with mongodb
func InitSession(session *mgo.Session) (*mgo.Session, error) {
if db == nil {
if session == nil {
db1, err := mgo.Dial(app.Config.MongoURL)
if err != nil {
logger.Error(err)
return nil, err
}
session = db1
}
db = &Database{session}
}
return db.Session, nil
}
func (d *Database) InitDatabase(session *mgo.Session) {
d.Session = session
}
// Create is a wrapper for mgo.Insert function.
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) Create(dbName, collection string, data ...interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).Insert(data...)
return
}
// GetByID is a wrapper for mgo.FindId function.
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) GetByID(dbName, collection string, id bson.ObjectId, response interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).FindId(id).One(response)
return
}
// Get is a wrapper for mgo.Find function.
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) Get(dbName, collection string, query interface{}, offset, limit int, response interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).Find(query).Skip(offset).Limit(limit).All(response)
return
}
// GetOne return one document
func (d *Database) GetOne(dbName, collection string, query interface{}, response interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).Find(query).One(response)
// logger.Debug(err, query)
return
}
func (d *Database) Query(dbName, collection string, query interface{}, selector interface{}, offset, limit int, response interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).Find(query).Skip(offset).Limit(limit).Select(selector).All(response)
return
}
// GetAndSort is a wrapper for mgo.Find function with SORT function in pipeline.
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) GetAndSort(dbName, collection string, query interface{}, sort []string, offset, limit int, response interface{}) (err error) {
sc := d.Session.Copy()
defer sc.Close()
err = sc.DB(dbName).C(collection).Find(query).Sort(sort...).Skip(offset).Limit(limit).All(response)
return
}
// Update is a wrapper for mgo.Update function.
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) Update(dbName, collection string, query interface{}, update interface{}) error {
sc := d.Session.Copy()
defer sc.Close()
err := sc.DB(dbName).C(collection).Update(query, update)
if err != nil {
logger.Error(err)
return err
}
return nil
}
func (d *Database) Upsert(dbName, collection string, query interface{}, update interface{}) (interface{}, error) {
sc := d.Session.Copy()
defer sc.Close()
changed, err := sc.DB(dbName).C(collection).Upsert(query, update)
if err != nil {
logger.Error(err)
return nil, err
}
return changed.UpsertedId, nil
}
func (d *Database) UpdateAll(dbName, collection string, query interface{}, update interface{}) error {
sc := d.Session.Copy()
defer sc.Close()
_, err := sc.DB(dbName).C(collection).UpdateAll(query, update)
if err != nil {
logger.Error(err)
return err
}
return nil
}
func (d *Database) FindAndModify(dbName, collection string, query interface{}, change mgo.Change, response interface{}) error {
sc := d.Session.Copy()
defer sc.Close()
_, err := sc.DB(dbName).C(collection).Find(query).Apply(change, response)
if err != nil {
logger.Error(err)
return err
}
return nil
}
// Aggregate is a wrapper for mgo.Pipe function.
// It is used to make mongo aggregate pipeline queries
// It creates a copy of session initialized, sends query over this session
// and returns the session to connection pool
func (d *Database) Aggregate(dbName, collection string, query []bson.M, response interface{}) error {
sc := d.Session.Copy()
defer sc.Close()
result := reflect.ValueOf(response).Interface()
err := sc.DB(dbName).C(collection).Pipe(query).All(result)
if err != nil {
logger.Error(err)
return err
}
return nil
}
// Remove removes one document matching a certain query
func (d *Database) Remove(dbName, collection string, query []bson.M) error {
sc := d.Session.Copy()
defer sc.Close()
err := sc.DB(dbName).C(collection).Remove(query)
if err != nil {
logger.Error(err)
return err
}
return nil
}
// RemoveAll removes all the documents from a collection matching a certain query
func (d *Database) RemoveAll(dbName, collection string, query interface{}) error {
sc := d.Session.Copy()
defer sc.Close()
_, err := sc.DB(dbName).C(collection).RemoveAll(query)
if err != nil {
logger.Error(err)
return err
}
return nil
}
// DropCollection drops all the documents in a collection
func (d *Database) DropCollection(dbName, collection string) error {
sc := d.Session.Copy()
defer sc.Close()
err := sc.DB(dbName).C(collection).DropCollection()
if err != nil {
logger.Error(err)
return err
}
return nil
}