Skip to content

Commit 4c3b748

Browse files
xiamxiam
authored
migrate mongodb adapter to the official MongoDB Go driver
1 parent fb5164f commit 4c3b748

File tree

21 files changed

+480
-454
lines changed

21 files changed

+480
-454
lines changed

‎adapter/cockroachdb/helper_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -99,7 +99,7 @@ func (h *Helper) TearDown() error {
9999
return h.sess.Close()
100100
}
101101

102-
func (h *Helper) TearUp() error {
102+
func (h *Helper) SetUp() error {
103103
var err error
104104

105105
h.sess, err = Open(settings)

‎adapter/mongo/Makefile‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
SHELL ?= bash
22

3-
MONGO_VERSION ?= 4
4-
MONGO_SUPPORTED ?= $(MONGO_VERSION) 3
3+
MONGO_VERSION ?= 8
4+
MONGO_SUPPORTED ?= $(MONGO_VERSION) 7
55
PROJECT ?= $(subst .,_,"upper_mongo_$(MONGO_VERSION)")
66

77
DB_HOST ?= 127.0.0.1

‎adapter/mongo/collection.go‎

Lines changed: 36 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -22,22 +22,22 @@
2222
package mongo
2323

2424
import (
25+
"context"
2526
"fmt"
27+
"reflect"
2628
"strings"
2729
"sync"
2830

29-
"reflect"
30-
3131
db "github.com/upper/db/v4"
3232
"github.com/upper/db/v4/internal/adapter"
33-
mgo "gopkg.in/mgo.v2"
34-
"gopkg.in/mgo.v2/bson"
33+
"go.mongodb.org/mongo-driver/bson"
34+
"go.mongodb.org/mongo-driver/mongo"
3535
)
3636

3737
// Collection represents a mongodb collection.
3838
type Collection struct {
3939
parent *Source
40-
collection *mgo.Collection
40+
collection *mongo.Collection
4141
}
4242

4343
var (
@@ -108,9 +108,15 @@ func compare(field string, cmp *adapter.Comparison) (string, interface{}) {
108108
}
109109
return field, bson.M{"$ne": value}
110110
case adapter.ComparisonOperatorRegExp, adapter.ComparisonOperatorLike:
111-
return field, bson.RegEx{Pattern: value.(string), Options: ""}
111+
return field, bson.M{
112+
"$regex": value.(string),
113+
}
112114
case adapter.ComparisonOperatorNotRegExp, adapter.ComparisonOperatorNotLike:
113-
return field, bson.M{"$not": bson.RegEx{Pattern: value.(string), Options: ""}}
115+
return field, bson.M{
116+
"$not": bson.M{
117+
"$regex": value.(string),
118+
},
119+
}
114120
}
115121

116122
if cmpOp, ok := comparisonOperators[op]; ok {
@@ -122,8 +128,8 @@ func compare(field string, cmp *adapter.Comparison) (string, interface{}) {
122128
panic(fmt.Sprintf("Unsupported operator %v", op))
123129
}
124130

125-
// compileStatement transforms conditions into something *mgo.Session can
126-
// understand.
131+
// compileStatement transforms upper-db conditions into something that the
132+
// adapter can understand.
127133
func compileStatement(cond db.Cond) bson.M {
128134
conds := bson.M{}
129135

@@ -170,10 +176,7 @@ func compileStatement(cond db.Cond) bson.M {
170176
return conds
171177
}
172178

173-
// compileConditions compiles terms into something *mgo.Session can
174-
// understand.
175179
func (col *Collection) compileConditions(term interface{}) interface{} {
176-
177180
switch t := term.(type) {
178181
case []interface{}:
179182
values := []interface{}{}
@@ -208,8 +211,6 @@ func (col *Collection) compileConditions(term interface{}) interface{} {
208211
return nil
209212
}
210213

211-
// compileQuery compiles terms into something that *mgo.Session can
212-
// understand.
213214
func (col *Collection) compileQuery(terms ...interface{}) interface{} {
214215
compiled := col.compileConditions(terms)
215216
if compiled == nil {
@@ -236,13 +237,12 @@ func (col *Collection) compileQuery(terms ...interface{}) interface{} {
236237

237238
// Name returns the name of the table or tables that form the collection.
238239
func (col *Collection) Name() string {
239-
return col.collection.Name
240+
return col.collection.Name()
240241
}
241242

242243
// Truncate deletes all rows from the table.
243244
func (col *Collection) Truncate() error {
244-
err := col.collection.DropCollection()
245-
245+
err := col.collection.Drop(context.Background())
246246
if err != nil {
247247
return err
248248
}
@@ -268,100 +268,33 @@ func (col *Collection) UpdateReturning(item interface{}) error {
268268

269269
// Insert inserts a record (map or struct) into the collection.
270270
func (col *Collection) Insert(item interface{}) (db.InsertResult, error) {
271-
var err error
271+
ctx := context.Background()
272272

273-
id := getID(item)
274-
275-
if col.parent.versionAtLeast(2, 6, 0, 0) {
276-
// this breaks MongoDb older than 2.6
277-
if _, err = col.collection.Upsert(bson.M{"_id": id}, item); err != nil {
278-
return nil, err
279-
}
280-
} else {
281-
// Allocating a new ID.
282-
if err = col.collection.Insert(bson.M{"_id": id}); err != nil {
283-
return nil, err
284-
}
285-
286-
// Now append data the user wants to append.
287-
if err = col.collection.Update(bson.M{"_id": id}, item); err != nil {
288-
// Cleanup allocated ID
289-
if err := col.collection.Remove(bson.M{"_id": id}); err != nil {
290-
return nil, err
291-
}
292-
return nil, err
293-
}
273+
res, err := col.collection.InsertOne(ctx, item)
274+
if err != nil {
275+
return nil, err
294276
}
295277

296-
return db.NewInsertResult(id), nil
278+
return db.NewInsertResult(res.InsertedID), nil
297279
}
298280

299281
// Exists returns true if the collection exists.
300282
func (col *Collection) Exists() (bool, error) {
301-
query := col.parent.database.C(`system.namespaces`).Find(map[string]string{`name`: fmt.Sprintf(`%s.%s`, col.parent.database.Name, col.collection.Name)})
302-
count, err := query.Count()
303-
return count > 0, err
304-
}
283+
ctx := context.Background()
284+
mcol := col.parent.database.Collection("system.namespaces")
305285

306-
// Fetches object _id or generates a new one if object doesn't have one or the one it has is invalid
307-
func getID(item interface{}) interface{} {
308-
v := reflect.ValueOf(item) // convert interface to Value
309-
v = reflect.Indirect(v) // convert pointers
310-
311-
switch v.Kind() {
312-
case reflect.Map:
313-
if inItem, ok := item.(map[string]interface{}); ok {
314-
if id, ok := inItem["_id"]; ok {
315-
bsonID, ok := id.(bson.ObjectId)
316-
if ok {
317-
return bsonID
318-
}
319-
}
320-
}
321-
case reflect.Struct:
322-
t := v.Type()
323-
324-
idCacheMutex.RLock()
325-
fieldName, found := idCache[t]
326-
idCacheMutex.RUnlock()
327-
328-
if !found {
329-
for n := 0; n < t.NumField(); n++ {
330-
field := t.Field(n)
331-
if field.PkgPath != "" {
332-
continue // Private field
333-
}
334-
335-
tag := field.Tag.Get("bson")
336-
if tag == "" {
337-
tag = field.Tag.Get("db")
338-
}
339-
340-
if tag == "" {
341-
continue
342-
}
343-
344-
parts := strings.Split(tag, ",")
345-
346-
if parts[0] == "_id" {
347-
fieldName = field.Name
348-
idCacheMutex.RLock()
349-
idCache[t] = fieldName
350-
idCacheMutex.RUnlock()
351-
break
352-
}
353-
}
354-
}
355-
if fieldName != "" {
356-
if bsonID, ok := v.FieldByName(fieldName).Interface().(bson.ObjectId); ok {
357-
if bsonID.Valid() {
358-
return bsonID
359-
}
360-
} else {
361-
return v.FieldByName(fieldName).Interface()
362-
}
363-
}
286+
mcur, err := mcol.Find(ctx, bson.M{
287+
"name": fmt.Sprintf("%s.%s", col.parent.database.Name(), col.collection.Name()),
288+
})
289+
if err != nil {
290+
return false, err
291+
}
292+
defer mcur.Close(ctx)
293+
294+
hasNext := mcur.Next(ctx)
295+
if err := mcur.Err(); err != nil {
296+
return false, err
364297
}
365298

366-
return bson.NewObjectId()
299+
return hasNext, nil
367300
}

‎adapter/mongo/connection.go‎

Lines changed: 19 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -27,10 +27,14 @@ import (
2727
"strings"
2828
)
2929

30-
const connectionScheme = `mongodb`
30+
const (
31+
defaultScheme = "mongodb"
32+
srvScheme = "mongodb+srv"
33+
)
3134

3235
// ConnectionURL implements a MongoDB connection struct.
3336
type ConnectionURL struct {
37+
Scheme string
3438
User string
3539
Password string
3640
Host string
@@ -39,13 +43,12 @@ type ConnectionURL struct {
3943
}
4044

4145
func (c ConnectionURL) String() (s string) {
46+
vv := url.Values{}
4247

4348
if c.Database == "" {
4449
return ""
4550
}
4651

47-
vv := url.Values{}
48-
4952
// Do we have any options?
5053
if c.Options == nil {
5154
c.Options = map[string]string{}
@@ -67,9 +70,13 @@ func (c ConnectionURL) String() (s string) {
6770
}
6871
}
6972

73+
if c.Scheme == "" {
74+
c.Scheme = defaultScheme
75+
}
76+
7077
// Building URL.
7178
u := url.URL{
72-
Scheme: connectionScheme,
79+
Scheme: c.Scheme,
7380
Path: c.Database,
7481
Host: c.Host,
7582
User: userInfo,
@@ -80,17 +87,20 @@ func (c ConnectionURL) String() (s string) {
8087
}
8188

8289
// ParseURL parses s into a ConnectionURL struct.
90+
// See https://www.mongodb.com/docs/manual/reference/connection-string/
8391
func ParseURL(s string) (conn ConnectionURL, err error) {
8492
var u *url.URL
8593

86-
if !strings.HasPrefix(s, connectionScheme+"://") {
87-
return conn, fmt.Errorf(`Expecting mongodb:// connection scheme.`)
94+
hasPrefix := strings.HasPrefix(s, defaultScheme+"://") || strings.HasPrefix(s, srvScheme+"://")
95+
if !hasPrefix {
96+
return conn, fmt.Errorf("invalid scheme")
8897
}
8998

9099
if u, err = url.Parse(s); err != nil {
91-
return conn, err
100+
return conn, fmt.Errorf("invalid URL: %v", err)
92101
}
93102

103+
conn.Scheme = u.Scheme
94104
conn.Host = u.Host
95105

96106
// Deleting / from start of the string.
@@ -108,12 +118,12 @@ func ParseURL(s string) (conn ConnectionURL, err error) {
108118
var vv url.Values
109119

110120
if vv, err = url.ParseQuery(u.RawQuery); err != nil {
111-
return conn, err
121+
return conn, fmt.Errorf("invalid query: %v", err)
112122
}
113123

114124
for k := range vv {
115125
conn.Options[k] = vv.Get(k)
116126
}
117127

118-
return conn, err
128+
return conn, nil
119129
}

‎adapter/mongo/connection_test.go‎

Lines changed: 23 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -70,30 +70,33 @@ func TestConnectionURL(t *testing.T) {
7070
}
7171

7272
func TestParseConnectionURL(t *testing.T) {
73-
var u ConnectionURL
74-
var s string
75-
var err error
73+
{
74+
const s = "mongodb:///mydatabase"
7675

77-
s = "mongodb:///mydatabase"
76+
u, err := ParseURL(s)
77+
require.NoError(t, err)
7878

79-
u, err = ParseURL(s)
80-
require.NoError(t, err)
81-
82-
assert.Equal(t, "mydatabase", u.Database)
79+
assert.Equal(t, "mydatabase", u.Database)
80+
}
8381

84-
s = "mongodb://user:pass@localhost,1.2.3.4,example.org:1234/another_database?cache=foobar&mode=ro"
82+
{
83+
const s = "mongodb://user:pass@localhost,1.2.3.4,example.org:1234/another_database?cache=foobar&mode=ro"
8584

86-
u, err = ParseURL(s)
87-
require.NoError(t, err)
85+
u, err := ParseURL(s)
86+
require.NoError(t, err)
8887

89-
assert.Equal(t, "another_database", u.Database)
90-
assert.Equal(t, "foobar", u.Options["cache"])
91-
assert.Equal(t, "ro", u.Options["mode"])
92-
assert.Equal(t, "user", u.User)
93-
assert.Equal(t, "pass", u.Password)
94-
assert.Equal(t, "localhost,1.2.3.4,example.org:1234", u.Host)
88+
assert.Equal(t, "another_database", u.Database)
89+
assert.Equal(t, "foobar", u.Options["cache"])
90+
assert.Equal(t, "ro", u.Options["mode"])
91+
assert.Equal(t, "mongodb", u.Scheme)
92+
assert.Equal(t, "user", u.User)
93+
assert.Equal(t, "pass", u.Password)
94+
assert.Equal(t, "localhost,1.2.3.4,example.org:1234", u.Host)
95+
}
9596

96-
s = "http://example.org"
97-
_, err = ParseURL(s)
98-
require.Error(t, err)
97+
{
98+
const s = "http://example.org"
99+
_, err := ParseURL(s)
100+
require.Error(t, err)
101+
}
99102
}

0 commit comments

Comments
 (0)