Generate EndSessions operation

GODRIVER-1051

Change-Id: I861e747c15ae88d6bda24e3b0df38b1c33edab5b

Author: connieyu

mongo/client.go

@@ -9,6 +9,7 @@ package mongo
 import (
 	"context"
 	"crypto/tls"
+	"strconv"
 	"strings"
 	"time"
 
@@ -19,6 +20,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/readconcern"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
 	"go.mongodb.org/mongo-driver/mongo/writeconcern"
+	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/auth"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/connstring"
@@ -27,11 +29,10 @@ import (
 	"go.mongodb.org/mongo-driver/x/mongo/driver/session"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/topology"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/uuid"
-	"go.mongodb.org/mongo-driver/x/mongo/driverlegacy"
-	"go.mongodb.org/mongo-driver/x/network/command"
 )
 
 const defaultLocalThreshold = 15 * time.Millisecond
+const batchSize = 10000
 
 // Client performs operations on a given topology.
 type Client struct {
@@ -190,12 +191,31 @@ func (c *Client) endSessions(ctx context.Context) {
 	if c.topology.SessionPool == nil {
 		return
 	}
-	cmd := command.EndSessions{
-		Clock:      c.clock,
-		SessionIDs: c.topology.SessionPool.IDSlice(),
+
+	ids := c.topology.SessionPool.IDSlice()
+	idx, idArray := bsoncore.AppendArrayStart(nil)
+	for i, id := range ids {
+		idDoc, _ := id.MarshalBSON()
+		idArray = bsoncore.AppendDocumentElement(idArray, strconv.Itoa(i), idDoc)
+	}
+	idArray, _ = bsoncore.AppendArrayEnd(idArray, idx)
+
+	op := operation.NewEndSessions(idArray).ClusterClock(c.clock).Deployment(c.topology).
+		ServerSelector(description.ReadPrefSelector(readpref.PrimaryPreferred())).CommandMonitor(c.monitor).Database("admin")
+
+	idx, idArray = bsoncore.AppendArrayStart(nil)
+	totalNumIDs := len(ids)
+	for i := 0; i < totalNumIDs; i++ {
+		idDoc, _ := ids[i].MarshalBSON()
+		idArray = bsoncore.AppendDocumentElement(idArray, strconv.Itoa(i), idDoc)
+		if ((i+1)%batchSize) == 0 || i == totalNumIDs-1 {
+			idArray, _ = bsoncore.AppendArrayEnd(idArray, idx)
+			_ = op.SessionIDs(idArray).Execute(ctx)
+			idArray = idArray[:0]
+			idx = 0
+		}
 	}
 
-	_, _ = driverlegacy.EndSessions(ctx, cmd, c.topology, description.ReadPrefSelector(readpref.PrimaryPreferred()))
 }
 
 func (c *Client) configure(opts *options.ClientOptions) error {

mongo/client_internal_test.go

@@ -20,6 +20,7 @@ import (
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/bsoncodec"
 	"go.mongodb.org/mongo-driver/bson/bsonrw"
+	"go.mongodb.org/mongo-driver/event"
 	"go.mongodb.org/mongo-driver/internal/testutil"
 	"go.mongodb.org/mongo-driver/mongo/options"
 	"go.mongodb.org/mongo-driver/mongo/readpref"
@@ -546,3 +547,29 @@ func TestClient_Watch_Disconnected(t *testing.T) {
 	require.Nil(t, change)
 	require.Equal(t, err, ErrClientDisconnected)
 }
+
+func TestEndSessions(t *testing.T) {
+	skipIfBelow36(t)
+	cs := testutil.ConnString(t)
+	client, err := NewClient(options.Client().ApplyURI(cs.String()).SetMonitor(monitor))
+	require.NoError(t, err)
+	err = client.Connect(nil)
+	require.NoError(t, err)
+
+	_, err = client.ListDatabases(ctx, bsonx.Doc{})
+	require.NoError(t, err)
+
+	drainChannels()
+
+	err = client.Disconnect(ctx)
+	require.NoError(t, err)
+
+	var started *event.CommandStartedEvent
+	select {
+	case started = <-startedChan:
+	default:
+		t.Fatalf("expected a CommandStartedEvent but none found")
+	}
+
+	require.Equal(t, "endSessions", started.CommandName)
+}

x/mongo/driver/drivergen/drivergen.go

@@ -154,9 +154,11 @@ func (op Operation) CommandMethod() (string, error) {
 			return "", fmt.Errorf("unknown request field type %s", field.Type)
 		}
 		var rf struct {
-			ShortName     string
-			Name          string
-			ParameterName string
+			ShortName              string
+			Name                   string
+			ParameterName          string
+			MinWireVersion         int
+			MinWireVersionRequired int
 		}
 		rf.ShortName = op.ShortName()
 		rf.Name = op.Command.Parameter
@@ -173,7 +175,7 @@ func (op Operation) CommandMethod() (string, error) {
 	sort.Strings(names)
 	for _, name := range names {
 		field := op.Request[name]
-		if name == op.Properties.Batches {
+		if name == op.Properties.Batches || field.Skip {
 			continue
 		}
 		var tmpl *template.Template
@@ -538,6 +540,7 @@ type RequestField struct {
 	Slice                  bool
 	Constructor            bool
 	Variadic               bool
+	Skip                   bool
 	Documentation          string
 	MinWireVersion         int
 	MinWireVersionRequired int

x/mongo/driver/operation/end_sessions.go

@@ -0,0 +1,16 @@
+version = 0
+name = "EndSessions"
+documentation = "EndSessions performs an endSessions operation."
+
+[properties]
+disabled = ["collection"]
+
+[command]
+name = "endSessions"
+parameter = "sessionIDs"
+
+[request.sessionIDs]
+type = "array"
+documentation = "sessionIDs specify the sessions to be expired."
+skip = true
+constructor = true

x/mongo/driver/operation/end_sessions.toml

@@ -12,3 +12,4 @@ package operation
 //go:generate operationgen commit_transaction.toml operation commit_transaction.go
 //go:generate operationgen abort_transaction.toml operation abort_transaction.go
 //go:generate operationgen count.toml operation count.go
+//go:generate operationgen end_sessions.toml operation end_sessions.go