GODRIVER-3889 Move commitTransaction and abortTransaction to the mongo package.

mongo/op_abort_transaction.go

@@ -0,0 +1,78 @@
+// Copyright (C) MongoDB, Inc. 2019-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 mongo
+
+import (
+	"context"
+	"errors"
+
+	"go.mongodb.org/mongo-driver/v2/event"
+	"go.mongodb.org/mongo-driver/v2/internal/driverutil"
+	"go.mongodb.org/mongo-driver/v2/internal/logger"
+	"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"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/session"
+)
+
+// abortTransactionOp performs an abortTransaction operation.
+type abortTransactionOp struct {
+	authenticator             driver.Authenticator
+	recoveryToken             bsoncore.Document
+	session                   *session.Client
+	clock                     *session.ClusterClock
+	monitor                   *event.CommandMonitor
+	database                  string
+	deployment                driver.Deployment
+	selector                  description.ServerSelector
+	writeConcern              *writeconcern.WriteConcern
+	retry                     *driver.RetryMode
+	maxAdaptiveRetries        uint
+	enableOverloadRetargeting bool
+	serverAPI                 *driver.ServerAPIOptions
+	logger                    *logger.Logger
+}
+
+func (at *abortTransactionOp) processResponse(context.Context, bsoncore.Document, driver.ResponseInfo) error {
+	return nil
+}
+
+// Execute runs this operation and returns an error if the operation did not execute successfully.
+func (at *abortTransactionOp) Execute(ctx context.Context) error {
+	if at.deployment == nil {
+		return errors.New("the abortTransaction operation must have a Deployment set before Execute can be called")
+	}
+
+	return driver.Operation{
+		CommandFn:                 at.command,
+		ProcessResponseFn:         at.processResponse,
+		RetryMode:                 at.retry,
+		Type:                      driver.Write,
+		Client:                    at.session,
+		Clock:                     at.clock,
+		CommandMonitor:            at.monitor,
+		MaxAdaptiveRetries:        at.maxAdaptiveRetries,
+		EnableOverloadRetargeting: at.enableOverloadRetargeting,
+		Database:                  at.database,
+		Deployment:                at.deployment,
+		Selector:                  at.selector,
+		WriteConcern:              at.writeConcern,
+		ServerAPI:                 at.serverAPI,
+		Name:                      driverutil.AbortTransactionOp,
+		Authenticator:             at.authenticator,
+		Logger:                    at.logger,
+	}.Execute(ctx)
+}
+
+func (at *abortTransactionOp) command(dst []byte, _ description.SelectedServer) ([]byte, error) {
+	dst = bsoncore.AppendInt32Element(dst, "abortTransaction", 1)
+	if at.recoveryToken != nil {
+		dst = bsoncore.AppendDocumentElement(dst, "recoveryToken", at.recoveryToken)
+	}
+	return dst, nil
+}

mongo/op_commit_transaction.go

@@ -0,0 +1,78 @@
+// Copyright (C) MongoDB, Inc. 2019-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 mongo
+
+import (
+	"context"
+	"errors"
+
+	"go.mongodb.org/mongo-driver/v2/event"
+	"go.mongodb.org/mongo-driver/v2/internal/driverutil"
+	"go.mongodb.org/mongo-driver/v2/internal/logger"
+	"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"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/description"
+	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/session"
+)
+
+// commitTransactionOp attempts to commit a transaction.
+type commitTransactionOp struct {
+	authenticator             driver.Authenticator
+	recoveryToken             bsoncore.Document
+	session                   *session.Client
+	clock                     *session.ClusterClock
+	monitor                   *event.CommandMonitor
+	database                  string
+	deployment                driver.Deployment
+	selector                  description.ServerSelector
+	writeConcern              *writeconcern.WriteConcern
+	retry                     *driver.RetryMode
+	maxAdaptiveRetries        uint
+	enableOverloadRetargeting bool
+	serverAPI                 *driver.ServerAPIOptions
+	logger                    *logger.Logger
+}
+
+func (ct *commitTransactionOp) processResponse(context.Context, bsoncore.Document, driver.ResponseInfo) error {
+	return nil
+}
+
+// Execute runs this operation and returns an error if the operation did not execute successfully.
+func (ct *commitTransactionOp) Execute(ctx context.Context) error {
+	if ct.deployment == nil {
+		return errors.New("the commitTransaction operation must have a Deployment set before Execute can be called")
+	}
+
+	return driver.Operation{
+		CommandFn:                 ct.command,
+		ProcessResponseFn:         ct.processResponse,
+		RetryMode:                 ct.retry,
+		Type:                      driver.Write,
+		Client:                    ct.session,
+		Clock:                     ct.clock,
+		CommandMonitor:            ct.monitor,
+		MaxAdaptiveRetries:        ct.maxAdaptiveRetries,
+		EnableOverloadRetargeting: ct.enableOverloadRetargeting,
+		Database:                  ct.database,
+		Deployment:                ct.deployment,
+		Selector:                  ct.selector,
+		WriteConcern:              ct.writeConcern,
+		ServerAPI:                 ct.serverAPI,
+		Name:                      driverutil.CommitTransactionOp,
+		Authenticator:             ct.authenticator,
+		Logger:                    ct.logger,
+	}.Execute(ctx)
+}
+
+func (ct *commitTransactionOp) command(dst []byte, _ description.SelectedServer) ([]byte, error) {
+	dst = bsoncore.AppendInt32Element(dst, "commitTransaction", 1)
+	if ct.recoveryToken != nil {
+		dst = bsoncore.AppendDocumentElement(dst, "recoveryToken", ct.recoveryToken)
+	}
+	return dst, nil
+}

mongo/session.go

@@ -19,7 +19,6 @@ import (
 	"go.mongodb.org/mongo-driver/v2/mongo/options"
 	"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/session"
 )
 
@@ -273,13 +272,24 @@ func (s *Session) AbortTransaction(ctx context.Context) error {
 	selector := makePinnedSelector(s.clientSession, &serverselector.Write{})
 
 	s.clientSession.Aborting = true
-	_ = operation.NewAbortTransaction().Session(s.clientSession).ClusterClock(s.client.clock).Database("admin").
-		Deployment(s.deployment).WriteConcern(s.clientSession.CurrentWc).ServerSelector(selector).
-		Retry(driver.RetryOncePerCommand).MaxAdaptiveRetries(s.client.effectiveAdaptiveRetries(true)).
-		EnableOverloadRetargeting(s.client.enableOverloadRetargeting).
-		CommandMonitor(s.client.monitor).
-		RecoveryToken(bsoncore.Document(s.clientSession.RecoveryToken)).ServerAPI(s.client.serverAPI).
-		Authenticator(s.client.authenticator).Logger(s.client.logger).Execute(ctx)
+	retry := driver.RetryOncePerCommand
+	op := abortTransactionOp{
+		session:                   s.clientSession,
+		clock:                     s.client.clock,
+		database:                  "admin",
+		deployment:                s.deployment,
+		writeConcern:              s.clientSession.CurrentWc,
+		selector:                  selector,
+		retry:                     &retry,
+		maxAdaptiveRetries:        s.client.effectiveAdaptiveRetries(true),
+		enableOverloadRetargeting: s.client.enableOverloadRetargeting,
+		monitor:                   s.client.monitor,
+		recoveryToken:             bsoncore.Document(s.clientSession.RecoveryToken),
+		serverAPI:                 s.client.serverAPI,
+		authenticator:             s.client.authenticator,
+		logger:                    s.client.logger,
+	}
+	_ = op.Execute(ctx)
 
 	s.clientSession.Aborting = false
 	_ = s.clientSession.AbortTransaction()
@@ -309,13 +319,23 @@ func (s *Session) CommitTransaction(ctx context.Context) error {
 	selector := makePinnedSelector(s.clientSession, &serverselector.Write{})
 
 	s.clientSession.Committing = true
-	op := operation.NewCommitTransaction().
-		Session(s.clientSession).ClusterClock(s.client.clock).Database("admin").Deployment(s.deployment).
-		WriteConcern(s.clientSession.CurrentWc).ServerSelector(selector).Retry(driver.RetryOncePerCommand).
-		MaxAdaptiveRetries(s.client.effectiveAdaptiveRetries(true)).
-		EnableOverloadRetargeting(s.client.enableOverloadRetargeting).
-		CommandMonitor(s.client.monitor).RecoveryToken(bsoncore.Document(s.clientSession.RecoveryToken)).
-		ServerAPI(s.client.serverAPI).Authenticator(s.client.authenticator).Logger(s.client.logger)
+	retry := driver.RetryOncePerCommand
+	op := commitTransactionOp{
+		session:                   s.clientSession,
+		clock:                     s.client.clock,
+		database:                  "admin",
+		deployment:                s.deployment,
+		writeConcern:              s.clientSession.CurrentWc,
+		selector:                  selector,
+		retry:                     &retry,
+		maxAdaptiveRetries:        s.client.effectiveAdaptiveRetries(true),
+		enableOverloadRetargeting: s.client.enableOverloadRetargeting,
+		monitor:                   s.client.monitor,
+		recoveryToken:             bsoncore.Document(s.clientSession.RecoveryToken),
+		serverAPI:                 s.client.serverAPI,
+		authenticator:             s.client.authenticator,
+		logger:                    s.client.logger,
+	}
 
 	err = op.Execute(ctx)
 	// Return error without updating transaction state if it is a timeout, as the transaction has not