Merge pull request #23 from jiangz222/aggregate

support aggregate

Author: jiangz222

README.md

@@ -86,7 +86,7 @@ var oneUserInfo = UserInfo{
 Create index
 
 ```go
-cli.EnsureIndexes(ctx, []string{"name"}, []string{"age", "name,weight"})
+cli.EnsureIndexes(ctx, []string{}, []string{"age", "name,weight"})
 ```
 
 - Insert a document
@@ -113,36 +113,41 @@ err = cli.Remove(ctx, BsonT{"age": 7})
 - Insert multiple data
 
 ```go
-// batch insert
+// multiple insert
 var batchUserInfoI = []interface{}{
-    UserInfo{Name: "wxy", Age: 6, Weight: 20},
-    UserInfo{Name: "jZ", Age: 6, Weight: 25},
-    UserInfo{Name: "zp", Age: 6, Weight: 30},
-    UserInfo{Name: "yxw", Age: 6, Weight: 35},
+	UserInfo{Name: "a1", Age: 6, Weight: 20},
+	UserInfo{Name: "b2", Age: 6, Weight: 25},
+	UserInfo{Name: "c3", Age: 6, Weight: 30},
+	UserInfo{Name: "d4", Age: 6, Weight: 35},
+	UserInfo{Name: "a1", Age: 7, Weight: 40},
+	UserInfo{Name: "a1", Age: 8, Weight: 45},
 }
 result, err = cli.Collection.InsertMany(ctx, batchUserInfoI)
 ```
 
 - Search all, sort and limit
-
 ```go
 // find all, sort and limit
 batch := []UserInfo{}
 cli.Find(ctx, BsonT{"age": 6}).Sort("weight").Limit(7).All(&batch)
 ```
-
+- Count
+````go
+count, err := cli.Find(ctx, BsonT{"age": 6}).Count()
+````
+- Aggregate
+```go
+matchStage := qmgo.D{{"$match", []qmgo.E{{"weight", qmgo.D{{"$gt", 30}}}}}}
+groupStage := qmgo.D{{"$group", qmgo.D{{"_id", "$name"}, {"total", qmgo.D{{"$sum", "$age"}}}}}}
+var showsWithInfo []qmgo.M
+err = cli.Aggregate(context.Background(), qmgo.Pipeline{matchStage, groupStage}).All(&showsWithInfo)
+```
 ## Feature
-
-- Supported
-  - CRUD to documents
-  - Create indexes
-  - Sort、limit、count
-- TODO
-  - Transaction
-  - Aggregate
-  - Options for every operation
-
-
+- CRUD to documents
+- Create indexes
+- Sort、limit、count、select
+- Cursor
+- Aggregate
 
 ## `qmgo` vs `mgo` vs `go.mongodb.org/mongo-driver`
 
@@ -155,8 +160,8 @@ How do we do in`go.mongodb.org/mongo-driver`:
 // find all, sort and limit
 findOptions := options.Find()
 findOptions.SetLimit(7) // set limit
-var sorts bson.D
-sorts = append(sorts, bson.E{Key: "weight", Value: 1})
+var sorts D
+sorts = append(sorts, E{Key: "weight", Value: 1})
 findOptions.SetSort(sorts) // set sort
 
 batch := []UserInfo{}

README_ZH.md

@@ -78,7 +78,7 @@ var oneUserInfo = UserInfo{
 创建索引
 
 ```go
-cli.EnsureIndexes(ctx, []string{"name"}, []string{"age", "name,weight"})
+cli.EnsureIndexes(ctx, []string{}, []string{"age", "name,weight"})
 ```
 
 - 插入一个文档
@@ -105,12 +105,14 @@ err = cli.Remove(ctx, BsonT{"age": 7})
 - 插入多条数据
 
 ```go
-// batch insert
+// multiple insert
 var batchUserInfoI = []interface{}{
-    UserInfo{Name: "wxy", Age: 6, Weight: 20},
-    UserInfo{Name: "jZ", Age: 6, Weight: 25},
-    UserInfo{Name: "zp", Age: 6, Weight: 30},
-    UserInfo{Name: "yxw", Age: 6, Weight: 35},
+	UserInfo{Name: "a1", Age: 6, Weight: 20},
+	UserInfo{Name: "b2", Age: 6, Weight: 25},
+	UserInfo{Name: "c3", Age: 6, Weight: 30},
+	UserInfo{Name: "d4", Age: 6, Weight: 35},
+	UserInfo{Name: "a1", Age: 7, Weight: 40},
+	UserInfo{Name: "a1", Age: 8, Weight: 45},
 }
 result, err = cli.Collection.InsertMany(ctx, batchUserInfoI)
 ```
@@ -123,17 +125,26 @@ batch := []UserInfo{}
 cli.Find(ctx, BsonT{"age": 6}).Sort("weight").Limit(7).All(&batch)
 ```
 
-## 功能
+- Count
+````go
+count, err := cli.Find(ctx, BsonT{"age": 6}).Count()
+````
 
-- 已经支持
-  - 文档的增删改查
-  - 索引配置
-  - 查询`Sort`、`Limit`、`Count`
+- Aggregate
+```go
+matchStage := qmgo.D{{"$match", []qmgo.E{{"weight", qmgo.D{{"$gt", 30}}}}}}
+groupStage := qmgo.D{{"$group", qmgo.D{{"_id", "$name"}, {"total", qmgo.D{{"$sum", "$age"}}}}}}
+var showsWithInfo []qmgo.M
+err = cli.Aggregate(context.Background(), qmgo.Pipeline{matchStage, groupStage}).All(&showsWithInfo)
+```
+
+## 功能
+- 文档的增删改查
+- 索引配置
+- `Sort`、`Limit`、`Count`、`Select`
+- `Cursor`
+- 聚合`Aggregate`
 
-- TODO:
-  - 事务
-  - 聚合`Aggregate`
-  - 操作支持`Options` 
 
 ## `qmgo` vs `mgo` vs `go.mongodb.org/mongo-driver`
 
@@ -146,8 +157,8 @@ cli.Find(ctx, BsonT{"age": 6}).Sort("weight").Limit(7).All(&batch)
 // find all 、sort and limit
 findOptions := options.Find()
 findOptions.SetLimit(7)  // set limit
-var sorts bson.D
-sorts = append(sorts, bson.E{Key: "weight", Value: 1})
+var sorts D
+sorts = append(sorts, E{Key: "weight", Value: 1})
 findOptions.SetSort(sorts) // set sort
 
 batch := []UserInfo{}

aggregate.go

@@ -0,0 +1,53 @@
+package qmgo
+
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+)
+
+type Pipeline []bson.D
+
+type Aggregate struct {
+	ctx        context.Context
+	pipeline   interface{}
+	collection *mongo.Collection
+}
+
+// All iterates the cursor from aggregate and decodes each document into results.
+func (a *Aggregate) All(results interface{}) error {
+	c, err := a.collection.Aggregate(a.ctx, a.pipeline)
+	if err != nil {
+		return err
+	}
+	return c.All(a.ctx, results)
+}
+
+// One iterates the cursor from aggregate and decodes current document into result.
+func (a *Aggregate) One(result interface{}) error {
+	c, err := a.collection.Aggregate(a.ctx, a.pipeline)
+	if err != nil {
+		return err
+	}
+	cr := Cursor{
+		ctx:    a.ctx,
+		cursor: c,
+		err:    err,
+	}
+	defer cr.Close()
+	if !cr.Next(result) {
+		return ERR_NO_SUCH_RECORD
+	}
+	return err
+}
+
+// Iter return the cursor after aggregate
+func (a *Aggregate) Iter() CursorI {
+	c, err := a.collection.Aggregate(a.ctx, a.pipeline)
+	return &Cursor{
+		ctx:    a.ctx,
+		cursor: c,
+		err:    err,
+	}
+}

aggregate_test.go

@@ -0,0 +1,102 @@
+package qmgo
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+func TestAggregate(t *testing.T) {
+	ast := require.New(t)
+
+	var cli *QmgoClient
+
+	cli = initClient("test")
+	cli.DropCollection(context.Background())
+	cli.EnsureIndexes(context.Background(), nil, []string{"name"})
+
+	id1 := primitive.NewObjectID()
+	id2 := primitive.NewObjectID()
+	id3 := primitive.NewObjectID()
+	id4 := primitive.NewObjectID()
+	id5 := primitive.NewObjectID()
+	docs := []interface{}{
+		QueryTestItem{Id: id1, Name: "Alice", Age: 10},
+		QueryTestItem{Id: id2, Name: "Alice", Age: 12},
+		QueryTestItem{Id: id3, Name: "Lucas", Age: 33},
+		QueryTestItem{Id: id4, Name: "Lucas", Age: 22},
+		QueryTestItem{Id: id5, Name: "Lucas", Age: 44},
+	}
+	cli.InsertMany(context.Background(), docs)
+	matchStage := bson.D{{"$match", []bson.E{{"age", bson.D{{"$gt", 11}}}}}}
+	groupStage := bson.D{{"$group", bson.D{{"_id", "$name"}, {"total", bson.D{{"$sum", "$age"}}}}}}
+	var showsWithInfo []bson.M
+	// aggregate ALL()
+	err := cli.Aggregate(context.Background(), Pipeline{matchStage, groupStage}).All(&showsWithInfo)
+	ast.NoError(err)
+	ast.Equal(2, len(showsWithInfo))
+	for _, v := range showsWithInfo {
+		if "Alice" == v["_id"] {
+			ast.Equal(int32(12), v["total"])
+			continue
+		}
+		if "Lucas" == v["_id"] {
+			ast.Equal(int32(99), v["total"])
+			continue
+		}
+		ast.Error(errors.New("error"), "impossible")
+	}
+	// Iter()
+	iter := cli.Aggregate(context.Background(), Pipeline{matchStage, groupStage})
+	ast.NotNil(iter)
+	err = iter.All(&showsWithInfo)
+	ast.NoError(err)
+	for _, v := range showsWithInfo {
+		if "Alice" == v["_id"] {
+			ast.Equal(int32(12), v["total"])
+			continue
+		}
+		if "Lucas" == v["_id"] {
+			ast.Equal(int32(99), v["total"])
+			continue
+		}
+		ast.Error(errors.New("error"), "impossible")
+	}
+	// One()
+	var oneInfo bson.M
+
+	iter = cli.Aggregate(context.Background(), Pipeline{matchStage, groupStage})
+	ast.NotNil(iter)
+
+	err = iter.One(&oneInfo)
+	ast.NoError(err)
+	ast.Equal(true, oneInfo["_id"] == "Alice" || oneInfo["_id"] == "Lucas")
+
+	// iter
+	iter = cli.Aggregate(context.Background(), Pipeline{matchStage, groupStage})
+	ast.NotNil(iter)
+
+	i := iter.Iter()
+
+	ct := i.Next(&oneInfo)
+	ast.Equal(true, oneInfo["_id"] == "Alice" || oneInfo["_id"] == "Lucas")
+	ast.Equal(true, ct)
+	ct = i.Next(&oneInfo)
+	ast.Equal(true, oneInfo["_id"] == "Alice" || oneInfo["_id"] == "Lucas")
+	ast.Equal(true, ct)
+	ct = i.Next(&oneInfo)
+	ast.Equal(false, ct)
+
+	// err
+	ast.Error(cli.Aggregate(context.Background(), 1).All(&showsWithInfo))
+	ast.Error(cli.Aggregate(context.Background(), 1).One(&showsWithInfo))
+	ast.Error(cli.Aggregate(context.Background(), 1).Iter().Err())
+	matchStage = bson.D{{"$match", []bson.E{{"age", bson.D{{"$gt", 100}}}}}}
+	groupStage = bson.D{{"$group", bson.D{{"_id", "$name"}, {"total", bson.D{{"$sum", "$age"}}}}}}
+	ast.Error(cli.Aggregate(context.Background(), Pipeline{matchStage, groupStage}).One(&showsWithInfo))
+
+}

base.go

@@ -2,7 +2,6 @@ package qmgo
 
 import (
 	"strings"
-	"time"
 )
 
 // QmgoConfig for initial mongodb instance
@@ -31,10 +30,10 @@ func IsDup(err error) bool {
 	return strings.Contains(err.Error(), "E11000")
 }
 
-// Now return Millisecond current time
-func Now() time.Time {
-	return time.Unix(0, time.Now().UnixNano()/1e6*1e6)
-}
+//// Now return Millisecond current time
+//func Now() time.Time {
+//	return time.Unix(0, time.Now().UnixNano()/1e6*1e6)
+//}
 
 // SplitSortField handle sort symbol: "+"/"-" in front of field
 // if "+"， return sort as 1

client.go

@@ -27,11 +27,6 @@ func Open(ctx context.Context, conf *Config) (cli *QmgoClient, err error) {
 	}
 
 	db := client.Database(conf.Database)
-	if err != nil {
-		fmt.Println("new database fail", err)
-		return
-	}
-
 	coll := db.Collection(conf.Coll)
 
 	cli = &QmgoClient{