GODRIVER-3891 Remove the operation.Command type.

internal/integtest/integtest.go

@@ -25,10 +25,23 @@ import (
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/connstring"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/topology"
 )
 
+// dropDatabase drops the given database on the provided deployment.
+func dropDatabase(db string, deployment driver.Deployment) error {
+	cmd := bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "dropDatabase", 1))
+	return driver.Operation{
+		CommandFn: func(dst []byte, _ description.SelectedServer) ([]byte, error) {
+			return append(dst, cmd[4:len(cmd)-1]...), nil
+		},
+		Database:   db,
+		Selector:   &serverselector.Write{},
+		Deployment: deployment,
+	}.Execute(context.Background())
+}
+
 var (
 	connectionString     *connstring.ConnString
 	connectionStringOnce sync.Once
@@ -106,8 +119,7 @@ func MonitoredTopology(t *testing.T, dbName string, monitor *event.CommandMonito
 	} else {
 		_ = monitoredTopology.Connect()
 
-		err = operation.NewCommand(bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "dropDatabase", 1))).
-			Database(dbName).ServerSelector(&serverselector.Write{}).Deployment(monitoredTopology).Execute(context.Background())
+		err = dropDatabase(dbName, monitoredTopology)
 
 		require.NoError(t, err)
 	}
@@ -133,9 +145,7 @@ func Topology(t *testing.T) *topology.Topology {
 		} else {
 			_ = liveTopology.Connect()
 
-			err = operation.NewCommand(bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "dropDatabase", 1))).
-				Database(DBName(t)).ServerSelector(&serverselector.Write{}).
-				Deployment(liveTopology).Execute(context.Background())
+			err = dropDatabase(DBName(t), liveTopology)
 			require.NoError(t, err)
 		}
 	})

mongo/database.go

@@ -152,9 +152,8 @@ func (db *Database) Aggregate(
 func (db *Database) processRunCommand(
 	ctx context.Context,
 	cmd any,
-	cursorCommand bool,
 	opts ...options.Lister[options.RunCmdOptions],
-) (*operation.Command, *session.Client, error) {
+) (*driver.Operation, *session.Client, error) {
 	args, err := mongoutil.NewOptions[options.RunCmdOptions](append(defaultRunCmdOpts, opts...)...)
 	if err != nil {
 		return nil, nil, fmt.Errorf("failed to construct options from builder: %w", err)
@@ -195,26 +194,29 @@ func (db *Database) processRunCommand(
 		readSelect = makePinnedSelector(sess, readSelect)
 	}
 
-	var op *operation.Command
-	switch retryOverload := db.client.retryReads && db.client.retryWrites; cursorCommand {
-	case true:
-		cursorOpts := db.client.createBaseCursorOptions(retryOverload)
+	retryOverload := db.client.retryReads && db.client.retryWrites
 
-		cursorOpts.MarshalValueEncoderFn = newEncoderFn(db.bsonOpts, db.registry)
-
-		op = operation.NewCursorCommand(runCmdDoc, cursorOpts)
-	default:
-		op = operation.NewCommand(runCmdDoc)
-		maxAdaptiveRetries := db.client.effectiveAdaptiveRetries(retryOverload)
-		op = op.MaxAdaptiveRetries(maxAdaptiveRetries).
-			EnableOverloadRetargeting(db.client.enableOverloadRetargeting)
-	}
-
-	return op.Session(sess).CommandMonitor(db.client.monitor).
-		ServerSelector(readSelect).ClusterClock(db.client.clock).
-		Database(db.name).Deployment(db.client.deployment).
-		Crypt(db.client.cryptFLE).ReadPreference(args.ReadPreference).ServerAPI(db.client.serverAPI).
-		Timeout(db.client.timeout).Logger(db.client.logger).Authenticator(db.client.authenticator), sess, nil
+	op := driver.Operation{
+		CommandFn: func(dst []byte, _ description.SelectedServer) ([]byte, error) {
+			return append(dst, runCmdDoc[4:len(runCmdDoc)-1]...), nil
+		},
+		Client:                    sess,
+		Clock:                     db.client.clock,
+		CommandMonitor:            db.client.monitor,
+		Database:                  db.name,
+		Deployment:                db.client.deployment,
+		ReadPreference:            args.ReadPreference,
+		Selector:                  readSelect,
+		MaxAdaptiveRetries:        db.client.effectiveAdaptiveRetries(retryOverload),
+		EnableOverloadRetargeting: db.client.enableOverloadRetargeting,
+		Crypt:                     db.client.cryptFLE,
+		ServerAPI:                 db.client.serverAPI,
+		Timeout:                   db.client.timeout,
+		Logger:                    db.client.logger,
+		Authenticator:             db.client.authenticator,
+	}
+
+	return &op, sess, nil
 }
 
 // RunCommand executes the given command against the database.
@@ -244,19 +246,25 @@ func (db *Database) RunCommand(
 		ctx = context.Background()
 	}
 
-	op, sess, err := db.processRunCommand(ctx, runCommand, false, opts...)
+	op, sess, err := db.processRunCommand(ctx, runCommand, opts...)
 	defer closeImplicitSession(sess)
 	if err != nil {
 		return &SingleResult{err: err}
 	}
 
+	var response bsoncore.Document
+	op.ProcessResponseFn = func(_ context.Context, resp bsoncore.Document, _ driver.ResponseInfo) error {
+		response = resp
+		return nil
+	}
+
 	err = op.Execute(ctx)
 	// RunCommand can be used to run a write, thus execute may return a write error
 	rr, convErr := processWriteError(err)
 	return &SingleResult{
 		ctx:          ctx,
 		err:          convErr,
-		rdr:          bson.Raw(op.Result()),
+		rdr:          bson.Raw(response),
 		bsonOpts:     db.bsonOpts,
 		reg:          db.registry,
 		Acknowledged: rr.isAcknowledged(),
@@ -286,12 +294,30 @@ func (db *Database) RunCommandCursor(
 		ctx = context.Background()
 	}
 
-	op, sess, err := db.processRunCommand(ctx, runCommand, true, opts...)
+	op, sess, err := db.processRunCommand(ctx, runCommand, opts...)
 	if err != nil {
 		closeImplicitSession(sess)
 		return nil, wrapErrors(err)
 	}
 
+	retryOverload := db.client.retryReads && db.client.retryWrites
+	cursorOpts := db.client.createBaseCursorOptions(retryOverload)
+	cursorOpts.MarshalValueEncoderFn = newEncoderFn(db.bsonOpts, db.registry)
+
+	var bc *driver.BatchCursor
+	op.ProcessResponseFn = func(_ context.Context, resp bsoncore.Document, info driver.ResponseInfo) error {
+		curDoc, err := driver.ExtractCursorDocument(resp)
+		if err != nil {
+			return err
+		}
+		cursorRes, err := driver.NewCursorResponse(curDoc, info)
+		if err != nil {
+			return err
+		}
+		bc, err = driver.NewBatchCursor(cursorRes, sess, db.client.clock, cursorOpts)
+		return err
+	}
+
 	if err = op.Execute(ctx); err != nil {
 		closeImplicitSession(sess)
 		if errors.Is(err, driver.ErrNoCursor) {
@@ -301,11 +327,6 @@ func (db *Database) RunCommandCursor(
 		return nil, wrapErrors(err)
 	}
 
-	bc, err := op.ResultCursor()
-	if err != nil {
-		closeImplicitSession(sess)
-		return nil, wrapErrors(err)
-	}
 	cursor, err := newCursorWithSession(bc, db.bsonOpts, db.registry, sess,
 		withCursorOptionClientTimeout(db.client.timeout))
 	return cursor, wrapErrors(err)

x/mongo/driver/auth/sasl.go

@@ -13,9 +13,29 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
 )
 
+// runCommand runs a generic command against the connection in cfg and returns
+// the raw server response.
+func runCommand(ctx context.Context, cfg *driver.AuthConfig, db string, cmd bsoncore.Document) (bsoncore.Document, error) {
+	var res bsoncore.Document
+	err := driver.Operation{
+		CommandFn: func(dst []byte, _ description.SelectedServer) ([]byte, error) {
+			return append(dst, cmd[4:len(cmd)-1]...), nil
+		},
+		ProcessResponseFn: func(_ context.Context, resp bsoncore.Document, _ driver.ResponseInfo) error {
+			res = resp
+			return nil
+		},
+		Database:   db,
+		Deployment: driver.SingleConnectionDeployment{C: cfg.Connection},
+		Clock:      cfg.ClusterClock,
+		ServerAPI:  cfg.ServerAPI,
+	}.Execute(ctx)
+	return res, err
+}
+
 // SaslClient is the client piece of a sasl conversation.
 type SaslClient interface {
 	Start() (string, []byte, error)
@@ -132,17 +152,10 @@ func (sc *saslConversation) Finish(ctx context.Context, cfg *driver.AuthConfig,
 			bsoncore.AppendInt32Element(nil, "conversationId", int32(cid)),
 			bsoncore.AppendBinaryElement(nil, "payload", 0x00, payload),
 		)
-		saslContinueCmd := operation.NewCommand(doc).
-			Database(sc.source).
-			Deployment(driver.SingleConnectionDeployment{cfg.Connection}).
-			ClusterClock(cfg.ClusterClock).
-			ServerAPI(cfg.ServerAPI)
-
-		err = saslContinueCmd.Execute(ctx)
+		rdr, err = runCommand(ctx, cfg, sc.source, doc)
 		if err != nil {
 			return newError(err, sc.mechanism)
 		}
-		rdr = saslContinueCmd.Result()
 
 		err = bson.Unmarshal(rdr, &saslResp)
 		if err != nil {
@@ -160,14 +173,10 @@ func ConductSaslConversation(ctx context.Context, cfg *driver.AuthConfig, authSo
 	if err != nil {
 		return newError(err, conversation.mechanism)
 	}
-	saslStartCmd := operation.NewCommand(saslStartDoc).
-		Database(authSource).
-		Deployment(driver.SingleConnectionDeployment{cfg.Connection}).
-		ClusterClock(cfg.ClusterClock).
-		ServerAPI(cfg.ServerAPI)
-	if err := saslStartCmd.Execute(ctx); err != nil {
+	saslStartRes, err := runCommand(ctx, cfg, authSource, saslStartDoc)
+	if err != nil {
 		return newError(err, conversation.mechanism)
 	}
 
-	return conversation.Finish(ctx, cfg, saslStartCmd.Result())
+	return conversation.Finish(ctx, cfg, saslStartRes)
 }

x/mongo/driver/auth/x509.go

@@ -12,7 +12,6 @@ import (
 
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
 )
 
 // MongoDBX509 is the mechanism name for MongoDBX509.
@@ -67,13 +66,7 @@ func (a *MongoDBX509Authenticator) CreateSpeculativeConversation() (SpeculativeC
 // Auth authenticates the provided connection by conducting an X509 authentication conversation.
 func (a *MongoDBX509Authenticator) Auth(ctx context.Context, cfg *driver.AuthConfig) error {
 	requestDoc := createFirstX509Message()
-	authCmd := operation.
-		NewCommand(requestDoc).
-		Database(sourceExternal).
-		Deployment(driver.SingleConnectionDeployment{cfg.Connection}).
-		ClusterClock(cfg.ClusterClock).
-		ServerAPI(cfg.ServerAPI)
-	err := authCmd.Execute(ctx)
+	_, err := runCommand(ctx, cfg, sourceExternal, requestDoc)
 	if err != nil {
 		return newAuthError("round trip error", err)
 	}

x/mongo/driver/integration/compressor_test.go

@@ -7,15 +7,13 @@
 package integration
 
 import (
-	"context"
 	"os"
 	"testing"
 
 	"go.mongodb.org/mongo-driver/v2/internal/integtest"
 	"go.mongodb.org/mongo-driver/v2/internal/require"
 	"go.mongodb.org/mongo-driver/v2/mongo/writeconcern"
 	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
 )
 
 func TestCompression(t *testing.T) {
@@ -32,14 +30,9 @@ func TestCompression(t *testing.T) {
 		bsoncore.BuildDocument(nil, bsoncore.AppendStringElement(nil, "name", "compression_test")),
 	)
 
-	cmd := operation.NewCommand(bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "serverStatus", 1))).
-		Deployment(integtest.Topology(t)).
-		Database(integtest.DBName(t))
-
-	ctx := context.Background()
-	err := cmd.Execute(ctx)
+	serverStatusCmd := bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "serverStatus", 1))
+	result, err := executeCommand(integtest.Topology(t), nil, integtest.DBName(t), serverStatusCmd)
 	noerr(t, err)
-	result := cmd.Result()
 
 	serverVersion, err := result.LookupErr("version")
 	noerr(t, err)

x/mongo/driver/integration/main_test.go

@@ -24,7 +24,7 @@ import (
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/auth"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/connstring"
-	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/topology"
 )
 
@@ -123,20 +123,41 @@ func addCompressorToURI(uri string) string {
 	return uri + "compressors=" + comp
 }
 
+// executeCommand runs an arbitrary command against the given deployment and
+// returns the raw server response.
+func executeCommand(
+	deployment driver.Deployment,
+	selector description.ServerSelector,
+	db string,
+	cmd bsoncore.Document,
+) (bsoncore.Document, error) {
+	var res bsoncore.Document
+	err := driver.Operation{
+		CommandFn: func(dst []byte, _ description.SelectedServer) ([]byte, error) {
+			return append(dst, cmd[4:len(cmd)-1]...), nil
+		},
+		ProcessResponseFn: func(_ context.Context, resp bsoncore.Document, _ driver.ResponseInfo) error {
+			res = resp
+			return nil
+		},
+		Database:   db,
+		Deployment: deployment,
+		Selector:   selector,
+	}.Execute(context.Background())
+	return res, err
+}
+
 // runCommand runs an arbitrary command on a given database of the target
 // server.
 func runCommand(s driver.Server, db string, cmd bsoncore.Document) error {
-	op := operation.NewCommand(cmd).
-		Database(db).
-		Deployment(driver.SingleServerDeployment{Server: s})
-	return op.Execute(context.Background())
+	_, err := executeCommand(driver.SingleServerDeployment{Server: s}, nil, db, cmd)
+	return err
 }
 
 // dropCollection drops the collection in the test cluster.
 func dropCollection(t *testing.T, dbname, colname string) {
-	err := operation.NewCommand(bsoncore.BuildDocument(nil, bsoncore.AppendStringElement(nil, "drop", colname))).
-		Database(dbname).ServerSelector(&serverselector.Write{}).Deployment(integtest.Topology(t)).
-		Execute(context.Background())
+	dropCmd := bsoncore.BuildDocument(nil, bsoncore.AppendStringElement(nil, "drop", colname))
+	_, err := executeCommand(integtest.Topology(t), &serverselector.Write{}, dbname, dropCmd)
 	if de, ok := err.(driver.Error); err != nil && (!ok || !de.NamespaceNotFound()) {
 		require.NoError(t, err)
 	}