Add read concern configuration to Safe{} (globalsign#2)

* Add read concern configuration to Safe{}

* Fix tests

Author: domodwyer

Diff for: session.go

@@ -1889,6 +1889,7 @@ func (s *Session) SetPrefetch(p float64) {
 type Safe struct {
 	W        int    // Min # of servers to ack before success
 	WMode    string // Write mode for MongoDB 2.0+ (e.g. "majority")
+	RMode    string // Read mode for MonogDB 3.2+ ("majority", "local", "linearizable")
 	WTimeout int    // Milliseconds to wait for W before timing out
 	FSync    bool   // Sync via the journal if present, or via data files sync otherwise
 	J        bool   // Sync via the journal if present
@@ -1900,7 +1901,7 @@ func (s *Session) Safe() (safe *Safe) {
 	defer s.m.Unlock()
 	if s.safeOp != nil {
 		cmd := s.safeOp.query.(*getLastError)
-		safe = &Safe{WTimeout: cmd.WTimeout, FSync: cmd.FSync, J: cmd.J}
+		safe = &Safe{WTimeout: cmd.WTimeout, FSync: cmd.FSync, J: cmd.J, RMode: s.queryConfig.op.readConcern}
 		switch w := cmd.W.(type) {
 		case string:
 			safe.WMode = w
@@ -1980,6 +1981,7 @@ func (s *Session) Safe() (safe *Safe) {
 //
 // Relevant documentation:
 //
+//     https://docs.mongodb.com/manual/reference/read-concern/
 //     http://www.mongodb.org/display/DOCS/getLastError+Command
 //     http://www.mongodb.org/display/DOCS/Verifying+Propagation+of+Writes+with+getLastError
 //     http://www.mongodb.org/display/DOCS/Data+Center+Awareness
@@ -1998,6 +2000,7 @@ func (s *Session) SetSafe(safe *Safe) {
 // That is:
 //
 //     - safe.WMode is always used if set.
+//     - safe.RMode is always used if set.
 //     - safe.W is used if larger than the current W and WMode is empty.
 //     - safe.FSync is always used if true.
 //     - safe.J is used if FSync is false.
@@ -2036,6 +2039,13 @@ func (s *Session) ensureSafe(safe *Safe) {
 		w = safe.W
 	}
 
+	// Set the read concern
+	switch safe.RMode {
+	case "majority", "local", "linearizable":
+		s.queryConfig.op.readConcern = safe.RMode
+	default:
+	}
+
 	var cmd getLastError
 	if s.safeOp == nil {
 		cmd = getLastError{1, w, safe.WTimeout, safe.FSync, safe.J}
@@ -3284,7 +3294,9 @@ func prepareFindOp(socket *mongoSocket, op *queryOp, limit int32) bool {
 		Snapshot:    op.options.Snapshot,
 		OplogReplay: op.flags&flagLogReplay != 0,
 		Collation:   op.options.Collation,
+		ReadConcern: readLevel{level: op.readConcern},
 	}
+
 	if op.limit < 0 {
 		find.BatchSize = -op.limit
 		find.SingleBatch = true
@@ -3334,7 +3346,7 @@ type findCmd struct {
 	Comment             string      `bson:"comment,omitempty"`
 	MaxScan             int         `bson:"maxScan,omitempty"`
 	MaxTimeMS           int         `bson:"maxTimeMS,omitempty"`
-	ReadConcern         interface{} `bson:"readConcern,omitempty"`
+	ReadConcern         readLevel   `bson:"readConcern,omitempty"`
 	Max                 interface{} `bson:"max,omitempty"`
 	Min                 interface{} `bson:"min,omitempty"`
 	ReturnKey           bool        `bson:"returnKey,omitempty"`
@@ -3348,6 +3360,12 @@ type findCmd struct {
 	Collation           *Collation  `bson:"collation,omitempty"`
 }
 
+// readLevel provides the nested "level: majority" serialisation needed for the
+// query read concern.
+type readLevel struct {
+	level string `bson:"level,omitempty"`
+}
+
 // getMoreCmd holds the command used for requesting more query results on MongoDB 3.2+.
 //
 // Relevant documentation:

Diff for: session_test.go

@@ -2781,6 +2781,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe := session.Safe()
 	c.Assert(safe.W, Equals, 0)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 0)
 	c.Assert(safe.FSync, Equals, false)
 	c.Assert(safe.J, Equals, false)
@@ -2790,6 +2791,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 1)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 2)
 	c.Assert(safe.FSync, Equals, true)
 	c.Assert(safe.J, Equals, false)
@@ -2799,6 +2801,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 0)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 0)
 	c.Assert(safe.FSync, Equals, false)
 	c.Assert(safe.J, Equals, false)
@@ -2808,6 +2811,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 5)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 6)
 	c.Assert(safe.FSync, Equals, false)
 	c.Assert(safe.J, Equals, true)
@@ -2817,6 +2821,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 5)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 6)
 	c.Assert(safe.FSync, Equals, false)
 	c.Assert(safe.J, Equals, true)
@@ -2826,6 +2831,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 6)
 	c.Assert(safe.WMode, Equals, "")
+	c.Assert(safe.RMode, Equals, "")
 	c.Assert(safe.WTimeout, Equals, 4)
 	c.Assert(safe.FSync, Equals, true)
 	c.Assert(safe.J, Equals, false)
@@ -2835,6 +2841,17 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 0)
 	c.Assert(safe.WMode, Equals, "majority")
+	c.Assert(safe.RMode, Equals, "")
+	c.Assert(safe.WTimeout, Equals, 2)
+	c.Assert(safe.FSync, Equals, true)
+	c.Assert(safe.J, Equals, false)
+
+	// Read concern
+	session.EnsureSafe(&mgo.Safe{RMode: "majority"})
+	safe = session.Safe()
+	c.Assert(safe.W, Equals, 0)
+	c.Assert(safe.WMode, Equals, "majority")
+	c.Assert(safe.RMode, Equals, "majority")
 	c.Assert(safe.WTimeout, Equals, 2)
 	c.Assert(safe.FSync, Equals, true)
 	c.Assert(safe.J, Equals, false)
@@ -2844,6 +2861,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 0)
 	c.Assert(safe.WMode, Equals, "something")
+	c.Assert(safe.RMode, Equals, "majority")
 	c.Assert(safe.WTimeout, Equals, 2)
 	c.Assert(safe.FSync, Equals, true)
 	c.Assert(safe.J, Equals, false)
@@ -2853,6 +2871,7 @@ func (s *S) TestSafeSetting(c *C) {
 	safe = session.Safe()
 	c.Assert(safe.W, Equals, 0)
 	c.Assert(safe.WMode, Equals, "something")
+	c.Assert(safe.RMode, Equals, "majority")
 	c.Assert(safe.WTimeout, Equals, 2)
 	c.Assert(safe.FSync, Equals, true)
 	c.Assert(safe.J, Equals, false)
@@ -2863,6 +2882,7 @@ func (s *S) TestSafeSetting(c *C) {
 	clone.EnsureSafe(&mgo.Safe{WMode: "foo"})
 	safe = session.Safe()
 	c.Assert(safe.WMode, Equals, "something")
+	c.Assert(safe.RMode, Equals, "majority")
 }
 
 func (s *S) TestSafeInsert(c *C) {

Diff for: socket.go

@@ -67,17 +67,18 @@ const (
 )
 
 type queryOp struct {
-	query      interface{}
-	collection string
-	serverTags []bson.D
-	selector   interface{}
-	replyFunc  replyFunc
-	mode       Mode
-	skip       int32
-	limit      int32
-	options    queryWrapper
-	hasOptions bool
-	flags      queryOpFlags
+	query       interface{}
+	collection  string
+	serverTags  []bson.D
+	selector    interface{}
+	replyFunc   replyFunc
+	mode        Mode
+	skip        int32
+	limit       int32
+	options     queryWrapper
+	hasOptions  bool
+	flags       queryOpFlags
+	readConcern string
 }
 
 type queryWrapper struct {