added readpref and tests.

Author: craiggwilson

core/cluster_monitor.go

@@ -32,10 +32,10 @@ func StartClusterMonitor(opts ClusterOptions) (*ClusterMonitor, error) {
 
 	if opts.ReplicaSetName != "" {
 		m.fsm.setName = opts.ReplicaSetName
-		m.fsm.ClusterType = desc.ReplicaSetNoPrimary
+		m.fsm.Type = desc.ReplicaSetNoPrimary
 	}
 	if opts.ConnectionMode == SingleMode {
-		m.fsm.ClusterType = desc.Single
+		m.fsm.Type = desc.Single
 	}
 
 	for _, ep := range opts.Servers {
@@ -210,15 +210,15 @@ func (fsm *clusterMonitorFSM) apply(d *desc.Server) {
 	copy(newServers, fsm.Servers)
 
 	fsm.Cluster = desc.Cluster{
-		ClusterType: fsm.ClusterType,
-		Servers:     newServers,
+		Type:    fsm.Type,
+		Servers: newServers,
 	}
 
 	if _, ok := fsm.findServer(d.Endpoint); !ok {
 		return
 	}
 
-	switch fsm.ClusterType {
+	switch fsm.Type {
 	case desc.UnknownClusterType:
 		fsm.applyToUnknownClusterType(d)
 	case desc.Sharded:
@@ -233,7 +233,7 @@ func (fsm *clusterMonitorFSM) apply(d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) applyToReplicaSetNoPrimary(d *desc.Server) {
-	switch d.ServerType {
+	switch d.Type {
 	case desc.Standalone, desc.Mongos:
 		fsm.removeServerByEndpoint(d.Endpoint)
 	case desc.RSPrimary:
@@ -246,7 +246,7 @@ func (fsm *clusterMonitorFSM) applyToReplicaSetNoPrimary(d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) applyToReplicaSetWithPrimary(d *desc.Server) {
-	switch d.ServerType {
+	switch d.Type {
 	case desc.Standalone, desc.Mongos:
 		fsm.removeServerByEndpoint(d.Endpoint)
 		fsm.checkIfHasPrimary()
@@ -261,7 +261,7 @@ func (fsm *clusterMonitorFSM) applyToReplicaSetWithPrimary(d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) applyToShardedClusterType(d *desc.Server) {
-	switch d.ServerType {
+	switch d.Type {
 	case desc.Mongos, desc.UnknownServerType:
 		fsm.replaceServer(d)
 	case desc.Standalone, desc.RSPrimary, desc.RSSecondary, desc.RSArbiter, desc.RSMember, desc.RSGhost:
@@ -270,7 +270,7 @@ func (fsm *clusterMonitorFSM) applyToShardedClusterType(d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) applyToSingle(d *desc.Server) {
-	switch d.ServerType {
+	switch d.Type {
 	case desc.UnknownServerType:
 		fsm.replaceServer(d)
 	case desc.Standalone, desc.Mongos:
@@ -291,7 +291,7 @@ func (fsm *clusterMonitorFSM) applyToSingle(d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) applyToUnknownClusterType(d *desc.Server) {
-	switch d.ServerType {
+	switch d.Type {
 	case desc.Mongos:
 		fsm.setType(desc.Sharded)
 		fsm.replaceServer(d)
@@ -433,7 +433,7 @@ func (fsm *clusterMonitorFSM) addServer(endpoint desc.Endpoint) {
 
 func (fsm *clusterMonitorFSM) findPrimary() (int, bool) {
 	for i, s := range fsm.Servers {
-		if s.ServerType == desc.RSPrimary {
+		if s.Type == desc.RSPrimary {
 			return i, true
 		}
 	}
@@ -474,5 +474,5 @@ func (fsm *clusterMonitorFSM) setServer(i int, d *desc.Server) {
 }
 
 func (fsm *clusterMonitorFSM) setType(clusterType desc.ClusterType) {
-	fsm.ClusterType = clusterType
+	fsm.Type = clusterType
 }

core/desc/cluster.go

@@ -2,8 +2,8 @@ package desc
 
 // Cluster is a description of a cluster.
 type Cluster struct {
-	ClusterType ClusterType
-	Servers     []*Server
+	Servers []*Server
+	Type    ClusterType
 }
 
 // Server returns the ServerDesc with the specified endpoint.

core/desc/server.go

@@ -13,20 +13,22 @@ const UnsetRTT = -1 * time.Millisecond
 type Server struct {
 	Endpoint Endpoint
 
-	CanonicalEndpoint  Endpoint
-	ElectionID         bson.ObjectId
-	LastError          error
-	LastWriteTimestamp time.Time
-	MaxBatchCount      uint16
-	MaxDocumentSize    uint32
-	MaxMessageSize     uint32
-	Members            []Endpoint
-	ServerType         ServerType
-	SetName            string
-	SetVersion         uint32
-	Tags               []TagSet
-	WireVersion        Range
-	Version            Version
+	CanonicalEndpoint Endpoint
+	ElectionID        bson.ObjectId
+	HeartbeatInterval time.Duration
+	LastError         error
+	LastUpdateTime    time.Time
+	LastWriteTime     time.Time
+	MaxBatchCount     uint16
+	MaxDocumentSize   uint32
+	MaxMessageSize    uint32
+	Members           []Endpoint
+	SetName           string
+	SetVersion        uint32
+	Tags              TagSet
+	Type              ServerType
+	WireVersion       Range
+	Version           Version
 
 	averageRTT    time.Duration
 	averageRTTSet bool

core/desc/tag.go

@@ -6,5 +6,42 @@ type Tag struct {
 	Value string
 }
 
-// TagSet is an ordered list of Tags.
+// NewTagSet creates a new tag set by taking the entries in pairs.
+func NewTagSet(tags ...string) TagSet {
+	if len(tags)%2 != 0 {
+		panic("desc.NewTagSet: argument count is odd")
+	}
+
+	var set TagSet
+	for i := 0; i < len(tags); i += 2 {
+		set = append(set, Tag{Name: tags[i], Value: tags[i+1]})
+	}
+	return set
+}
+
+// TagSet is an ordered list of name/value pairs.
 type TagSet []Tag
+
+// Contains indicates whether the name/value pair
+// exists in the tag set.
+func (ts TagSet) Contains(name, value string) bool {
+	for _, t := range ts {
+		if t.Name == name && t.Value == value {
+			return true
+		}
+	}
+
+	return false
+}
+
+// ContainsAll indicates whether all the name/value pairs
+// exist in the tag set.
+func (ts TagSet) ContainsAll(other []Tag) bool {
+	for _, ot := range other {
+		if !ts.Contains(ot.Name, ot.Value) {
+			return false
+		}
+	}
+
+	return true
+}

core/feature/supported.go

@@ -0,0 +1,17 @@
+package feature
+
+import (
+	"fmt"
+
+	"github.com/10gen/mongo-go-driver/core/desc"
+)
+
+// MaxStaleness returns an error if the given server
+// does not support max staleness.
+func MaxStaleness(version desc.Version) error {
+	if !version.AtLeast(3, 4, 0) {
+		return fmt.Errorf("max staleness is only supported for servers 3.4 or newer")
+	}
+
+	return nil
+}

core/feature/supported_test.go

@@ -0,0 +1,32 @@
+package feature_test
+
+import (
+	"testing"
+
+	"github.com/10gen/mongo-go-driver/core/desc"
+	"github.com/10gen/mongo-go-driver/core/feature"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMaxStaleness(t *testing.T) {
+
+	tests := []struct {
+		version  desc.Version
+		expected bool
+	}{
+		{desc.NewVersion(2, 4, 0), false},
+		{desc.NewVersion(3, 3, 99), false},
+		{desc.NewVersion(3, 4, 0), true},
+		{desc.NewVersion(3, 4, 1), true},
+	}
+	for _, test := range tests {
+		t.Run(test.version.String(), func(t *testing.T) {
+			err := feature.MaxStaleness(test.version)
+			if test.expected {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}

core/readpref/mode.go

@@ -0,0 +1,26 @@
+package readpref
+
+// Mode indicates the user's preference on reads.
+type Mode uint8
+
+// Mode constants
+const (
+	// PrimaryMode indicates that only a primary is
+	// considered for reading. This is the default
+	// mode.
+	PrimaryMode Mode = iota
+	// PrimaryPreferredMode indicates that if a primary
+	// is available, use it; otherwise, eligible
+	// secondaries will be considered.
+	PrimaryPreferredMode
+	// SecondaryMode indicates that only secondaries
+	// should be considered.
+	SecondaryMode
+	// SecondaryPreferredMode indicates that only secondaries
+	// should be considered when one is available. If none
+	// are available, then a primary will be considered.
+	SecondaryPreferredMode
+	// NearestMode indicates that all primaries and secondaries
+	// will be considered.
+	NearestMode
+)

core/readpref/readpref.go

@@ -0,0 +1,65 @@
+package readpref
+
+import (
+	"time"
+
+	"github.com/10gen/mongo-go-driver/core/desc"
+)
+
+// New constructs a read preference from the mode and optionally
+// some name value pairs.
+func New(mode Mode, tagSets ...desc.TagSet) *ReadPref {
+	// TODO: think about having an error for invalid construction
+
+	return &ReadPref{
+		maxStaleness: time.Duration(-1) * time.Millisecond,
+		mode:         mode,
+		tagSets:      tagSets,
+	}
+}
+
+// NewWithMaxStaleness constructs a read preference from a mode, a
+// maximum staleness, and optionally some name value pairs.
+func NewWithMaxStaleness(mode Mode, maxStaleness time.Duration, tagSets ...desc.TagSet) *ReadPref {
+	// TODO: think about having an error for invalid construction
+
+	return &ReadPref{
+		maxStaleness:    maxStaleness,
+		maxStalenessSet: true,
+		mode:            mode,
+		tagSets:         tagSets,
+	}
+}
+
+var defaultReadPref = &ReadPref{}
+
+// Default returns the default read preference.
+func Default() *ReadPref {
+	return defaultReadPref
+}
+
+// ReadPref determines which servers are considered suitable for read operations.
+type ReadPref struct {
+	maxStaleness    time.Duration
+	maxStalenessSet bool
+	mode            Mode
+	tagSets         []desc.TagSet
+}
+
+// MaxStaleness is the maximum amount of time to allow
+// a server to be considered eligible for selection. The
+// second return value indicates if this value has been set.
+func (r *ReadPref) MaxStaleness() (time.Duration, bool) {
+	return r.maxStaleness, r.maxStalenessSet
+}
+
+// Mode indicates the mode of the read preference.
+func (r *ReadPref) Mode() Mode {
+	return r.mode
+}
+
+// TagSets are multiple tag sets indicating
+// which servers should be considered.
+func (r *ReadPref) TagSets() []desc.TagSet {
+	return r.tagSets
+}