GODRIVER-1572 add TopologyVersion (#403)

Author: iwysiu

x/mongo/driver/description/server.go

@@ -50,6 +50,7 @@ type Server struct {
 	SetVersion              uint32
 	SpeculativeAuthenticate bsoncore.Document
 	Tags                    tag.Set
+	TopologyVersion         *TopologyVersion
 	Kind                    ServerKind
 	WireVersion             *VersionRange
 
@@ -239,13 +240,32 @@ func NewServer(addr address.Address, response bsoncore.Document) Server {
 				return desc
 			}
 			desc.SetVersion = uint32(i64)
+		case "speculativeAuthenticate":
+			desc.SpeculativeAuthenticate, ok = element.Value().DocumentOK()
+			if !ok {
+				desc.LastError = fmt.Errorf("expected 'speculativeAuthenticate' to be a document but it's a BSON %s",
+					element.Value().Type)
+				return desc
+			}
 		case "tags":
 			m, err := decodeStringMap(element, "tags")
 			if err != nil {
 				desc.LastError = err
 				return desc
 			}
 			desc.Tags = tag.NewTagSetFromMap(m)
+		case "topologyVersion":
+			doc, ok := element.Value().DocumentOK()
+			if !ok {
+				desc.LastError = fmt.Errorf("expected 'topologyVersion' to be a document but it's a BSON %s", element.Value().Type)
+				return desc
+			}
+
+			desc.TopologyVersion, err = NewTopologyVersion(doc)
+			if err != nil {
+				desc.LastError = err
+				return desc
+			}
 		}
 	}
 

x/mongo/driver/description/topology_version.go

@@ -0,0 +1,58 @@
+// Copyright (C) MongoDB, Inc. 2017-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+package description
+
+import (
+	"fmt"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
+)
+
+// TopologyVersion represents a software version.
+type TopologyVersion struct {
+	ProcessID primitive.ObjectID
+	Counter   int64
+}
+
+// NewTopologyVersion creates a TopologyVersion based on doc
+func NewTopologyVersion(doc bsoncore.Document) (*TopologyVersion, error) {
+	elements, err := doc.Elements()
+	if err != nil {
+		return nil, err
+	}
+	var tv TopologyVersion
+	var ok bool
+	for _, element := range elements {
+		switch element.Key() {
+		case "processId":
+			tv.ProcessID, ok = element.Value().ObjectIDOK()
+			if !ok {
+				return nil, fmt.Errorf("expected 'processId' to be a objectID but it's a BSON %s", element.Value().Type)
+			}
+		case "counter":
+			tv.Counter, ok = element.Value().Int64OK()
+			if !ok {
+				return nil, fmt.Errorf("expected 'counter' to be an int64 but it's a BSON %s", element.Value().Type)
+			}
+		}
+	}
+	return &tv, nil
+}
+
+// MoreRecentThan returns if this TopologyVersion is more recent than the one passed in.
+// This should be called on the original TopologyVersion with the new one passed in.
+func (tv *TopologyVersion) MoreRecentThan(other *TopologyVersion) bool {
+	if tv == nil || other == nil {
+		return false
+	}
+	if tv.ProcessID != other.ProcessID {
+		return false
+	}
+
+	return tv.Counter > other.Counter
+}

x/mongo/driver/errors.go

@@ -116,11 +116,12 @@ func (wce WriteCommandError) Retryable(wireVersion *description.VersionRange) bo
 // WriteConcernError is a write concern failure that occurred as a result of a
 // write operation.
 type WriteConcernError struct {
-	Name    string
-	Code    int64
-	Message string
-	Details bsoncore.Document
-	Labels  []string
+	Name            string
+	Code            int64
+	Message         string
+	Details         bsoncore.Document
+	Labels          []string
+	TopologyVersion *description.TopologyVersion
 }
 
 func (wce WriteConcernError) Error() string {
@@ -200,11 +201,12 @@ func (we WriteErrors) Error() string {
 
 // Error is a command execution error from the database.
 type Error struct {
-	Code    int32
-	Message string
-	Labels  []string
-	Name    string
-	Wrapped error
+	Code            int32
+	Message         string
+	Labels          []string
+	Name            string
+	Wrapped         error
+	TopologyVersion *description.TopologyVersion
 }
 
 // UnsupportedStorageEngine returns whether e came as a result of an unsupported storage engine
@@ -324,6 +326,7 @@ func extractError(rdr bsoncore.Document) error {
 	var code int32
 	var labels []string
 	var ok bool
+	var tv *description.TopologyVersion
 	var wcError WriteCommandError
 	elems, err := rdr.Elements()
 	if err != nil {
@@ -428,6 +431,15 @@ func extractError(rdr bsoncore.Document) error {
 					}
 				}
 			}
+		case "topologyVersion":
+			doc, ok := elem.Value().DocumentOK()
+			if !ok {
+				break
+			}
+			version, err := description.NewTopologyVersion(doc)
+			if err == nil {
+				tv = version
+			}
 		}
 	}
 
@@ -437,15 +449,19 @@ func extractError(rdr bsoncore.Document) error {
 		}
 
 		return Error{
-			Code:    code,
-			Message: errmsg,
-			Name:    codeName,
-			Labels:  labels,
+			Code:            code,
+			Message:         errmsg,
+			Name:            codeName,
+			Labels:          labels,
+			TopologyVersion: tv,
 		}
 	}
 
 	if len(wcError.WriteErrors) > 0 || wcError.WriteConcernError != nil {
 		wcError.Labels = labels
+		if wcError.WriteConcernError != nil {
+			wcError.WriteConcernError.TopologyVersion = tv
+		}
 		return wcError
 	}