Contributing:findAndModify support writeConcern (#185)

* socket: only send client metadata once per socket (#105)

Periodic cluster synchronisation calls isMaster() which currently resends the
"client" metadata every call - the spec specifies:

	isMaster commands issued after the initial connection handshake MUST NOT
	contain handshake arguments

	https://github.com/mongodb/specifications/blob/master/source/mongodb-handshake/handshake.rst#connection-handshake

This hotfix prevents subsequent isMaster calls from sending the client metadata
again - fixes #101 and fixes #103.

Thanks to @changwoo-nam @qhenkart @canthefason @jyoon17 for spotting the initial
issue, opening tickets, and having the problem debugged with a PoC fix before I
even woke up.

* Merge Development (#111)

* Brings in a patch on having flusher not suppress errors. (#81)

go-mgo#360

* Fallback to JSON tags when BSON tag isn't present (#91)

* Fallback to JSON tags when BSON tag isn't present


Cleanup.

* Add test to demonstrate tagging fallback.

- Test coverage for tagging test.

* socket: only send client metadata once per socket

Periodic cluster synchronisation calls isMaster() which currently resends the
"client" metadata every call - the spec specifies:

	isMaster commands issued after the initial connection handshake MUST NOT
	contain handshake arguments

	https://github.com/mongodb/specifications/blob/master/source/mongodb-handshake/handshake.rst#connection-handshake

This hotfix prevents subsequent isMaster calls from sending the client metadata
again - fixes #101 and fixes #103.

Thanks to @changwoo-nam @qhenkart @canthefason @jyoon17 for spotting the initial
issue, opening tickets, and having the problem debugged with a PoC fix before I
even woke up.

* Cluster abended test 254 (#100)

* Add a test that mongo Server gets their abended reset as necessary.

See https://github.com/go-mgo/mgo/issues/254 and
https://github.com/go-mgo/mgo/pull/255/files

* Include the patch from Issue 255.

This brings in a test which fails without the patch, and passes with the
patch. Still to be tested, manual tcpkill of a socket.

* changeStream support (#97)

Add $changeStream support

* readme: credit @peterdeka and @steve-gray (#110)

* Hotfix #120  (#136)

* cluster: fix deadlock in cluster synchronisation (#120)

For a impressively thorough breakdown of the problem, see:
	#120 (comment)

Huge thanks to @dvic and @KJTsanaktsidis for the report and fix.

* readme: credit @dvic and @KJTsanaktsidis

* findAndModify support writeConcern

* fix

Author: Mei-Zhao

session.go

@@ -2893,6 +2893,7 @@ 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
@@ -4894,18 +4895,22 @@ type findModifyCmd struct {
 	Collection                  string      `bson:"findAndModify"`
 	Query, Update, Sort, Fields interface{} `bson:",omitempty"`
 	Upsert, Remove, New         bool        `bson:",omitempty"`
+	WriteConcern                interface{} `bson:"writeConcern"`
 }
 
 type valueResult struct {
-	Value     bson.Raw
-	LastError LastError `bson:"lastErrorObject"`
+	Value        bson.Raw
+	LastError    LastError         `bson:"lastErrorObject"`
+	ConcernError writeConcernError `bson:"writeConcernError"`
 }
 
 // Apply runs the findAndModify MongoDB command, which allows updating, upserting
 // or removing a document matching a query and atomically returning either the old
 // version (the default) or the new version of the document (when ReturnNew is true).
 // If no objects are found Apply returns ErrNotFound.
 //
+// If the session is in safe mode, the LastError result will be returned as err.
+//
 // The Sort and Select query methods affect the result of Apply.  In case
 // multiple documents match the query, Sort enables selecting which document to
 // act upon by ordering it first.  Select enables retrieving only a selection
@@ -4942,15 +4947,27 @@ func (q *Query) Apply(change Change, result interface{}) (info *ChangeInfo, err
 	dbname := op.collection[:c]
 	cname := op.collection[c+1:]
 
+	// https://docs.mongodb.com/manual/reference/command/findAndModify/#dbcmd.findAndModify
+	session.m.RLock()
+	safeOp := session.safeOp
+	session.m.RUnlock()
+	var writeConcern interface{}
+	if safeOp == nil {
+		writeConcern = bson.D{{Name: "w", Value: 0}}
+	} else {
+		writeConcern = safeOp.query.(*getLastError)
+	}
+
 	cmd := findModifyCmd{
-		Collection: cname,
-		Update:     change.Update,
-		Upsert:     change.Upsert,
-		Remove:     change.Remove,
-		New:        change.ReturnNew,
-		Query:      op.query,
-		Sort:       op.options.OrderBy,
-		Fields:     op.selector,
+		Collection:   cname,
+		Update:       change.Update,
+		Upsert:       change.Upsert,
+		Remove:       change.Remove,
+		New:          change.ReturnNew,
+		Query:        op.query,
+		Sort:         op.options.OrderBy,
+		Fields:       op.selector,
+		WriteConcern: writeConcern,
 	}
 
 	session = session.Clone()
@@ -4993,6 +5010,14 @@ func (q *Query) Apply(change Change, result interface{}) (info *ChangeInfo, err
 	} else if change.Upsert {
 		info.UpsertedId = lerr.UpsertedId
 	}
+	if doc.ConcernError.Code != 0 {
+		var lerr LastError
+		e := doc.ConcernError
+		lerr.Code = e.Code
+		lerr.Err = e.ErrMsg
+		err = &lerr
+		return info, err
+	}
 	return info, nil
 }
 

session_test.go

@@ -1370,6 +1370,37 @@ func (s *S) TestFindAndModify(c *C) {
 	c.Assert(info, IsNil)
 }
 
+func (s *S) TestFindAndModifyWriteConcern(c *C) {
+	session, err := mgo.Dial("localhost:40011")
+	c.Assert(err, IsNil)
+	defer session.Close()
+
+	coll := session.DB("mydb").C("mycoll")
+	err = coll.Insert(M{"id": 42})
+	c.Assert(err, IsNil)
+
+	// Tweak the safety parameters to something unachievable.
+	session.SetSafe(&mgo.Safe{W: 4, WTimeout: 100})
+
+	var ret struct {
+		Id uint64 `bson:"id"`
+	}
+
+	change := mgo.Change{
+		Update:    M{"$inc": M{"id": 8}},
+		ReturnNew: false,
+	}
+	info, err := coll.Find(M{"id": M{"$exists": true}}).Apply(change, &ret)
+	c.Assert(info.Updated, Equals, 1)
+	c.Assert(info.Matched, Equals, 1)
+	c.Assert(ret.Id, Equals, uint64(42))
+
+	if s.versionAtLeast(3, 2) {
+		// findAndModify support writeConcern after version 3.2.
+		c.Assert(err, ErrorMatches, "timeout|timed out waiting for slaves|Not enough data-bearing nodes|waiting for replication timed out")
+	}
+}
+
 func (s *S) TestFindAndModifyBug997828(c *C) {
 	session, err := mgo.Dial("localhost:40001")
 	c.Assert(err, IsNil)