Merge branch 'GODRVIER-396' of https://github.com/igorKryvenko/mongo-go-driver into master

Change-Id: I2b71ac4e4f8e0b5d6b51e57c14b97a79da6c4038

Author: skriptble

core/connstring/connstring.go

@@ -61,6 +61,8 @@ type ConnString struct {
 	ReadConcernLevel                   string
 	ReadPreference                     string
 	ReadPreferenceTagSets              []map[string]string
+	MaxStaleness                       time.Duration
+	MaxStalenessSet                    bool
 	ReplicaSet                         string
 	ServerSelectionTimeout             time.Duration
 	ServerSelectionTimeoutSet          bool
@@ -444,6 +446,13 @@ func (p *parser) addOption(pair string) error {
 			tags[parts[0]] = parts[1]
 		}
 		p.ReadPreferenceTagSets = append(p.ReadPreferenceTagSets, tags)
+	case "maxstaleness":
+		n, err := strconv.Atoi(value)
+		if err != nil || n < 0 {
+			return fmt.Errorf("invalid value for %s: %s", key, value)
+		}
+		p.MaxStaleness = time.Duration(n) * time.Second
+		p.MaxStalenessSet = true
 	case "replicaset":
 		p.ReplicaSet = value
 	case "serverselectiontimeoutms":

core/connstring/connstring_test.go

@@ -359,6 +359,31 @@ func TestReadPreferenceTags(t *testing.T) {
 	}
 }
 
+func TestMaxStaleness(t *testing.T) {
+	tests := []struct {
+		s        string
+		expected time.Duration
+		err      bool
+	}{
+		{s: "maxStaleness=10", expected: time.Duration(10) * time.Second},
+		{s: "maxStaleness=100", expected: time.Duration(100) * time.Second},
+		{s: "maxStaleness=-2", err: true},
+		{s: "maxStaleness=gsdge", err: true},
+	}
+	for _, test := range tests {
+		s := fmt.Sprintf("mongodb://localhost/?%s", test.s)
+		t.Run(s, func(t *testing.T) {
+			cs, err := connstring.Parse(s)
+			if test.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, test.expected, cs.MaxStaleness)
+			}
+		})
+	}
+}
+
 func TestReplicaSet(t *testing.T) {
 	tests := []struct {
 		s        string

mongo/client.go

@@ -17,6 +17,7 @@ import (
 	"github.com/mongodb/mongo-go-driver/core/options"
 	"github.com/mongodb/mongo-go-driver/core/readconcern"
 	"github.com/mongodb/mongo-go-driver/core/readpref"
+	"github.com/mongodb/mongo-go-driver/core/tag"
 	"github.com/mongodb/mongo-go-driver/core/topology"
 	"github.com/mongodb/mongo-go-driver/core/writeconcern"
 )
@@ -97,7 +98,6 @@ func newClient(cs connstring.ConnString, opts *ClientOptions) (*Client, error) {
 	client := &Client{
 		connString:     cs,
 		localThreshold: defaultLocalThreshold,
-		readPreference: readpref.Primary(),
 	}
 
 	if opts != nil {
@@ -122,6 +122,10 @@ func newClient(cs connstring.ConnString, opts *ClientOptions) (*Client, error) {
 	client.topology = topo
 	client.readConcern = readConcernFromConnString(&client.connString)
 	client.writeConcern = writeConcernFromConnString(&client.connString)
+	client.readPreference, err = readPreferenceFromConnString(&client.connString)
+	if err != nil {
+		return nil, err
+	}
 
 	return client, nil
 }
@@ -173,6 +177,31 @@ func writeConcernFromConnString(cs *connstring.ConnString) *writeconcern.WriteCo
 	return wc
 }
 
+func readPreferenceFromConnString(cs *connstring.ConnString) (*readpref.ReadPref, error) {
+	var rp *readpref.ReadPref
+	var err error
+	options := make([]readpref.Option, 0, 1)
+
+	tagSets := tag.NewTagSetsFromMaps(cs.ReadPreferenceTagSets)
+	if len(tagSets) > 0 {
+		options = append(options, readpref.WithTagSets(tagSets...))
+	}
+
+	if cs.MaxStaleness != 0 {
+		options = append(options, readpref.WithMaxStaleness(cs.MaxStaleness))
+	}
+	if len(cs.ReadPreference) > 0 {
+		if rp == nil {
+			mode, _ := readpref.ModeFromString(cs.ReadPreference)
+			rp, err = readpref.New(mode, options...)
+			if err != nil {
+				return nil, err
+			}
+		}
+	}
+	return rp, nil
+}
+
 // Database returns a handle for a given database.
 func (c *Client) Database(name string) *Database {
 	return newDatabase(c, name)

mongo/client_internal_test.go

@@ -16,8 +16,11 @@ import (
 
 	"github.com/mongodb/mongo-go-driver/bson"
 	"github.com/mongodb/mongo-go-driver/core/readpref"
+	"github.com/mongodb/mongo-go-driver/core/tag"
 	"github.com/mongodb/mongo-go-driver/internal/testutil"
 	"github.com/stretchr/testify/require"
+
+	"time"
 )
 
 func createTestClient(t *testing.T) *Client {
@@ -296,3 +299,37 @@ func TestClient_ListDatabaseNames_filter(t *testing.T) {
 	require.Len(t, dbs, 1)
 	require.Equal(t, dbName, dbs[0])
 }
+
+func TestClient_ReadPreference(t *testing.T) {
+	t.Parallel()
+
+	if testing.Short() {
+		t.Skip()
+	}
+	var tags = []tag.Set{
+		{
+			tag.Tag{
+				Name:  "one",
+				Value: "1",
+			},
+		},
+		{
+			tag.Tag{
+				Name:  "two",
+				Value: "2",
+			},
+		},
+	}
+	baseConnString := testutil.ConnString(t)
+	cs := testutil.AddOptionsToURI(baseConnString.String(), "readpreference=secondary&readPreferenceTags=one:1&readPreferenceTags=two:2&maxStaleness=5")
+
+	c, err := NewClient(cs)
+	require.NoError(t, err)
+	require.NotNil(t, c)
+	require.Equal(t, readpref.SecondaryMode, c.readPreference.Mode())
+	require.Equal(t, tags, c.readPreference.TagSets())
+	d, flag := c.readPreference.MaxStaleness()
+	require.True(t, flag)
+	require.Equal(t, time.Duration(5)*time.Second, d)
+
+}

mongo/client_options.go

@@ -247,6 +247,18 @@ func (co *ClientOptions) ReadPreferenceTagSets(m []map[string]string) *ClientOpt
 	return &ClientOptions{next: co, opt: fn}
 }
 
+//MaxStaleness sets the "maxStaleness" field of the read pref to set on the client.
+func (co *ClientOptions) MaxStaleness(d time.Duration) *ClientOptions {
+	var fn option = func(c *Client) error {
+		if !c.connString.MaxStalenessSet {
+			c.connString.MaxStaleness = d
+			c.connString.MaxStalenessSet = true
+		}
+		return nil
+	}
+	return &ClientOptions{next: co, opt: fn}
+}
+
 // ReplicaSet specifies the name of the replica set of the cluster.
 func (co *ClientOptions) ReplicaSet(s string) *ClientOptions {
 	var fn option = func(c *Client) error {

mongo/client_options_test.go

@@ -80,6 +80,7 @@ func TestClientOptions_chainAll(t *testing.T) {
 		ReadPreference("secondary").
 		ReadPreferenceTagSets([]map[string]string{
 			{"nyc": "1"}}).
+		MaxStaleness(2 * time.Second).
 		ReplicaSet("foo").
 		ServerSelectionTimeout(time.Second).
 		Single(false).