diff --git a/internal/integtest/integtest.go b/internal/integtest/integtest.go index 110409f0cb..5e441355fc 100644 --- a/internal/integtest/integtest.go +++ b/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) } }) diff --git a/mongo/database.go b/mongo/database.go index c1d3474e91..2713698250 100644 --- a/mongo/database.go +++ b/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) diff --git a/x/mongo/driver/auth/sasl.go b/x/mongo/driver/auth/sasl.go index 659fe45e2d..a9148750de 100644 --- a/x/mongo/driver/auth/sasl.go +++ b/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) } diff --git a/x/mongo/driver/auth/x509.go b/x/mongo/driver/auth/x509.go index f839023435..f997b1c8a1 100644 --- a/x/mongo/driver/auth/x509.go +++ b/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) } diff --git a/x/mongo/driver/integration/compressor_test.go b/x/mongo/driver/integration/compressor_test.go index 35fd9fdd99..62ff88e963 100644 --- a/x/mongo/driver/integration/compressor_test.go +++ b/x/mongo/driver/integration/compressor_test.go @@ -7,7 +7,6 @@ package integration import ( - "context" "os" "testing" @@ -15,7 +14,6 @@ import ( "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) diff --git a/x/mongo/driver/integration/main_test.go b/x/mongo/driver/integration/main_test.go index 7ca6090e98..c4de1e0c3e 100644 --- a/x/mongo/driver/integration/main_test.go +++ b/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) } diff --git a/x/mongo/driver/operation/command.go b/x/mongo/driver/operation/command.go deleted file mode 100644 index a4f60abf76..0000000000 --- a/x/mongo/driver/operation/command.go +++ /dev/null @@ -1,266 +0,0 @@ -// Copyright (C) MongoDB, Inc. 2021-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 operation - -import ( - "context" - "errors" - "time" - - "go.mongodb.org/mongo-driver/v2/event" - "go.mongodb.org/mongo-driver/v2/internal/logger" - "go.mongodb.org/mongo-driver/v2/mongo/readpref" - "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/description" - "go.mongodb.org/mongo-driver/v2/x/mongo/driver/session" -) - -// Command is used to run a generic operation. -type Command struct { - authenticator driver.Authenticator - command bsoncore.Document - database string - deployment driver.Deployment - selector description.ServerSelector - readPreference *readpref.ReadPref - clock *session.ClusterClock - session *session.Client - monitor *event.CommandMonitor - resultResponse bsoncore.Document - resultCursor *driver.BatchCursor - maxAdaptiveRetries uint - enableOverloadRetargeting bool - crypt driver.Crypt - serverAPI *driver.ServerAPIOptions - createCursor bool - cursorOpts driver.CursorOptions - timeout *time.Duration - logger *logger.Logger -} - -// NewCommand constructs and returns a new Command. Once the operation is executed, the result may only be accessed via -// the Result() function. -func NewCommand(command bsoncore.Document) *Command { - return &Command{ - command: command, - } -} - -// NewCursorCommand constructs a new Command. Once the operation is executed, the server response will be used to -// construct a cursor, which can be accessed via the ResultCursor() function. -func NewCursorCommand(command bsoncore.Document, cursorOpts driver.CursorOptions) *Command { - return &Command{ - command: command, - cursorOpts: cursorOpts, - createCursor: true, - - maxAdaptiveRetries: cursorOpts.MaxAdaptiveRetries, - enableOverloadRetargeting: cursorOpts.EnableOverloadRetargeting, - } -} - -// Result returns the result of executing this operation. -func (c *Command) Result() bsoncore.Document { return c.resultResponse } - -// ResultCursor returns the BatchCursor that was constructed using the command response. If the operation was not -// configured to create a cursor (i.e. it was created using NewCommand rather than NewCursorCommand), this function -// will return nil and an error. -func (c *Command) ResultCursor() (*driver.BatchCursor, error) { - if !c.createCursor { - return nil, errors.New("command operation was not configured to create a cursor, but a result cursor was requested") - } - return c.resultCursor, nil -} - -// Execute runs this operations and returns an error if the operation did not execute successfully. -func (c *Command) Execute(ctx context.Context) error { - if c.deployment == nil { - return errors.New("the Command operation must have a Deployment set before Execute can be called") - } - - return driver.Operation{ - CommandFn: func(dst []byte, _ description.SelectedServer) ([]byte, error) { - return append(dst, c.command[4:len(c.command)-1]...), nil - }, - ProcessResponseFn: func(_ context.Context, resp bsoncore.Document, info driver.ResponseInfo) error { - c.resultResponse = resp - - if c.createCursor { - curDoc, err := driver.ExtractCursorDocument(resp) - if err != nil { - return err - } - cursorRes, err := driver.NewCursorResponse(curDoc, info) - if err != nil { - return err - } - - c.resultCursor, err = driver.NewBatchCursor(cursorRes, c.session, c.clock, c.cursorOpts) - return err - } - - return nil - }, - Client: c.session, - Clock: c.clock, - CommandMonitor: c.monitor, - Database: c.database, - Deployment: c.deployment, - ReadPreference: c.readPreference, - Selector: c.selector, - MaxAdaptiveRetries: c.maxAdaptiveRetries, - EnableOverloadRetargeting: c.enableOverloadRetargeting, - Crypt: c.crypt, - ServerAPI: c.serverAPI, - Timeout: c.timeout, - Logger: c.logger, - Authenticator: c.authenticator, - }.Execute(ctx) -} - -// Session sets the session for this operation. -func (c *Command) Session(session *session.Client) *Command { - if c == nil { - c = new(Command) - } - - c.session = session - return c -} - -// ClusterClock sets the cluster clock for this operation. -func (c *Command) ClusterClock(clock *session.ClusterClock) *Command { - if c == nil { - c = new(Command) - } - - c.clock = clock - return c -} - -// CommandMonitor sets the monitor to use for APM events. -func (c *Command) CommandMonitor(monitor *event.CommandMonitor) *Command { - if c == nil { - c = new(Command) - } - - c.monitor = monitor - return c -} - -// Database sets the database to run this operation against. -func (c *Command) Database(database string) *Command { - if c == nil { - c = new(Command) - } - - c.database = database - return c -} - -// Deployment sets the deployment to use for this operation. -func (c *Command) Deployment(deployment driver.Deployment) *Command { - if c == nil { - c = new(Command) - } - - c.deployment = deployment - return c -} - -// ReadPreference set the read preference used with this operation. -func (c *Command) ReadPreference(readPreference *readpref.ReadPref) *Command { - if c == nil { - c = new(Command) - } - - c.readPreference = readPreference - return c -} - -// ServerSelector sets the selector used to retrieve a server. -func (c *Command) ServerSelector(selector description.ServerSelector) *Command { - if c == nil { - c = new(Command) - } - - c.selector = selector - return c -} - -// MaxAdaptiveRetries specifies the maximum number of times the driver should retry operations -// that fail with a server side overload error. -func (c *Command) MaxAdaptiveRetries(maxAdaptiveRetries uint) *Command { - if c == nil { - c = new(Command) - } - - c.maxAdaptiveRetries = maxAdaptiveRetries - return c -} - -// EnableOverloadRetargeting specifies whether the driver adds the previously failed server's address -// to the list of deprioritized server addresses -func (c *Command) EnableOverloadRetargeting(enabled bool) *Command { - if c == nil { - c = new(Command) - } - - c.enableOverloadRetargeting = enabled - return c -} - -// Crypt sets the Crypt object to use for automatic encryption and decryption. -func (c *Command) Crypt(crypt driver.Crypt) *Command { - if c == nil { - c = new(Command) - } - - c.crypt = crypt - return c -} - -// ServerAPI sets the server API version for this operation. -func (c *Command) ServerAPI(serverAPI *driver.ServerAPIOptions) *Command { - if c == nil { - c = new(Command) - } - - c.serverAPI = serverAPI - return c -} - -// Timeout sets the timeout for this operation. -func (c *Command) Timeout(timeout *time.Duration) *Command { - if c == nil { - c = new(Command) - } - - c.timeout = timeout - return c -} - -// Logger sets the logger for this operation. -func (c *Command) Logger(logger *logger.Logger) *Command { - if c == nil { - c = new(Command) - } - - c.logger = logger - return c -} - -// Authenticator sets the authenticator to use for this operation. -func (c *Command) Authenticator(authenticator driver.Authenticator) *Command { - if c == nil { - c = new(Command) - } - - c.authenticator = authenticator - return c -}