Implement collation for Collection.Pipe()

session.go

@@ -2510,6 +2510,7 @@ type Pipe struct {
 	allowDisk  bool
 	batchSize  int
 	maxTimeMS  int64
+	collation  *Collation
 }
 
 type pipeCmd struct {
@@ -2519,6 +2520,7 @@ type pipeCmd struct {
 	Explain   bool           `bson:",omitempty"`
 	AllowDisk bool           `bson:"allowDiskUse,omitempty"`
 	MaxTimeMS int64          `bson:"maxTimeMS,omitempty"`
+	Collation *Collation     `bson:"collation,omitempty"`
 }
 
 type pipeCmdCursor struct {
@@ -2539,6 +2541,7 @@ type pipeCmdCursor struct {
 //     http://docs.mongodb.org/manual/applications/aggregation
 //     http://docs.mongodb.org/manual/tutorial/aggregation-examples
 //
+
 func (c *Collection) Pipe(pipeline interface{}) *Pipe {
 	session := c.Database.Session
 	session.m.RLock()
@@ -2572,6 +2575,7 @@ func (p *Pipe) Iter() *Iter {
 		Pipeline:  p.pipeline,
 		AllowDisk: p.allowDisk,
 		Cursor:    &pipeCmdCursor{p.batchSize},
+		Collation: p.collation,
 	}
 	if p.maxTimeMS > 0 {
 		cmd.MaxTimeMS = p.maxTimeMS
@@ -2761,6 +2765,22 @@ func (p *Pipe) SetMaxTime(d time.Duration) *Pipe {
 	return p
 }
 
+// Collation allows to specify language-specific rules for string comparison,
+// such as rules for lettercase and accent marks.
+// When specifying collation, the locale field is mandatory; all other collation
+// fields are optional
+//
+// Relevant documentation:
+//
+//      https://docs.mongodb.com/manual/reference/collation/
+//
+func (p *Pipe) Collation(collation *Collation) *Pipe {
+	if collation != nil {
+		p.collation = collation
+	}
+	return p
+}
+
 // LastError the error status of the preceding write operation on the current connection.
 //
 // Relevant documentation:

session_test.go

@@ -4617,6 +4617,36 @@ func (s *S) TestPipeExplain(c *C) {
 	c.Assert(result.Ok, Equals, 1)
 }
 
+func (s *S) TestPipeCollation(c *C) {
+	if !s.versionAtLeast(2, 1) {
+		c.Skip("Pipe only works on 2.1+")
+	}
+	if !s.versionAtLeast(3, 3, 12) {
+		c.Skip("collations being released with 3.4")
+	}
+
+	session, err := mgo.Dial("localhost:40001")
+	c.Assert(err, IsNil)
+	defer session.Close()
+
+	coll := session.DB("mydb").C("mycoll")
+	beatles := []string{"John", "RINGO", "George", "Paul"}
+	for _, n := range beatles {
+		err := coll.Insert(M{"name": n})
+		c.Assert(err, IsNil)
+	}
+
+	collation := &mgo.Collation{
+		Locale:   "en",
+		Strength: 1, // ignore case/diacritics
+	}
+	var result []struct{ Name string }
+	err = coll.Pipe([]M{{"$match": M{"name": "ringo"}}}).Collation(collation).All(&result)
+	c.Assert(err, IsNil)
+	c.Assert(len(result), Equals, 1)
+	c.Assert(result[0].Name, Equals, "RINGO")
+}
+
 func (s *S) TestBatch1Bug(c *C) {
 	session, err := mgo.Dial("localhost:40001")
 	c.Assert(err, IsNil)