修复Pipe

src/mongodb/mongodb.go

@@ -242,35 +242,48 @@ func (ms *MgoSess) All(v *[]map[string]interface{}) {
 }
 func (ms *MgoSess) Iter() *MgoIter {
 	it := &MgoIter{}
-	find := options.Find()
-	if ms.skip > 0 {
-		find.SetSkip(ms.skip)
-	}
-	if ms.limit > 0 {
-		find.SetLimit(ms.limit)
-	}
-	find.SetBatchSize(100)
-	if len(ms.sorts) > 0 {
-		sort := bson.D{}
-		for _, k := range ms.sorts {
-			switch k[:1] {
-			case "-":
-				sort = append(sort, bson.E{k[1:], -1})
-			case "+":
-				sort = append(sort, bson.E{k[1:], 1})
-			default:
-				sort = append(sort, bson.E{k, 1})
+	coll := ms.M.C.Database(ms.db).Collection(ms.coll)
+	var cur *mongo.Cursor
+	var err error
+	if ms.query != nil {
+		find := options.Find()
+		if ms.skip > 0 {
+			find.SetSkip(ms.skip)
+		}
+		if ms.limit > 0 {
+			find.SetLimit(ms.limit)
+		}
+		find.SetBatchSize(100)
+		if len(ms.sorts) > 0 {
+			sort := bson.D{}
+			for _, k := range ms.sorts {
+				switch k[:1] {
+				case "-":
+					sort = append(sort, bson.E{k[1:], -1})
+				case "+":
+					sort = append(sort, bson.E{k[1:], 1})
+				default:
+					sort = append(sort, bson.E{k, 1})
+				}
 			}
+			find.SetSort(sort)
+		}
+		if ms.fields != nil {
+			find.SetProjection(ms.fields)
+		}
+		cur, err = coll.Find(ms.M.Ctx, ms.query, find)
+		if err != nil {
+			log.Println("mgo find err", err.Error())
+		}
+	} else if ms.pipe != nil {
+		aggregate := options.Aggregate()
+		aggregate.SetBatchSize(100)
+		cur, err = coll.Aggregate(ms.M.Ctx, ms.pipe, aggregate)
+		if err != nil {
+			log.Println("mgo aggregate err", err.Error())
 		}
-		find.SetSort(sort)
-	}
-	if ms.fields != nil {
-		find.SetProjection(ms.fields)
 	}
-	cur, err := ms.M.C.Database(ms.db).Collection(ms.coll).Find(ms.M.Ctx, ms.query, find)
-	if err != nil {
-		log.Println("mgo find err", err.Error())
-	} else {
+	if err == nil {
 		it.Cursor = cur
 		it.Ctx = ms.M.Ctx
 	}

src/mongodb/mongodb_test.go

@@ -105,23 +105,28 @@ func Test_del(t *testing.T) {
 }
 func Test_itor(t *testing.T) {
 	m := &MongodbSim{
-		MongodbAddr: "192.168.3.128:27080",
+		MongodbAddr: "192.168.3.206:27080",
 		Size:        5,
 		DbName:      "wcj",
 	}
 	m.InitPool()
 	sess := m.GetMgoConn()
-	it := sess.DB("wcj").C("jyopenplatform_user").Find(map[string]interface{}{
-		"i_status": 1,
-		"i_freeze": 1,
-	}).Select(map[string]interface{}{
-		"a_power":          1,
-		"s_m_openid":       1,
-		"o_rulepro":        1,
-		"s_appid":          1,
-		"s_enterprisename": 1,
-	}).Sort("_id").Iter()
+	it := sess.DB("wcj").C("jyopenplatform_user").Pipe([]map[string]interface{}{
+		map[string]interface{}{
+			"$match": map[string]interface{}{
+				"s_province": "河南省",
+			},
+		},
+		map[string]interface{}{
+			"$group": map[string]interface{}{
+				"_id": "$s_city",
+				"count": map[string]interface{}{
+					"$sum": 1,
+				},
+			},
+		},
+	}).Iter()
 	for user := make(map[string]interface{}); it.Next(&user); {
-		log.Println(user["_id"], user["s_appid"])
+		log.Println(user)
 	}
 }