Transactions事务

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:本示例重点介绍了事务API的关键组件。特别是，它使用回调API。回调API：

• starts a transaction启动事务
• executes the specified operations执行指定的操作
• commits the result or ends the transaction on error提交结果或在出现错误时结束事务

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. 服务器端操作中的错误，如DuplicateKeyError，可能会结束事务，并导致命令错误，提醒用户事务已结束。This behavior is expected and occurs even if the client never calls Session.abortTransaction(). 即使客户端从未调用Session.abortTransaction()，也会发生这种行为。To incorporate custom error handling, use the Core API on your transaction.若要合并自定义错误处理，请在事务中使用Core API。

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

static bool
with_transaction_example(bson_error_t *error)
{
   mongoc_client_t *client = NULL;
   mongoc_write_concern_t *wc = NULL;
   mongoc_collection_t *coll = NULL;
   bool success = false;
   bool ret = false;
   bson_t *doc = NULL;
   bson_t *insert_opts = NULL;
   mongoc_client_session_t *session = NULL;
   mongoc_transaction_opt_t *txn_opts = NULL;

   /* For a replica set, include the replica set name and a seedlist of the members in the URI string;对于副本集，在URI字符串中包含副本集名称和成员的种子列表；
    *  e.g.
    * uri_repl = "mongodb://mongodb0.example.com:27017,mongodb1.example.com:" \
    *    "27017/?replicaSet=myRepl";
    * client = mongoc_client_new (uri_repl);
    * For a sharded cluster, connect to the mongos instances;对于分片集群，连接到mongos实例； e.g.
    * uri_sharded =
    * "mongodb://mongos0.example.com:27017,mongos1.example.com:27017/";
    * client = mongoc_client_new (uri_sharded);
    */

   client = get_client();

   /* Prereq: Create collections. Note Atlas connection strings include a majority write
    * concern by default.
    */
   wc = mongoc_write_concern_new();
   mongoc_write_concern_set_wmajority(wc, 1000);
   insert_opts = bson_new();
   mongoc_write_concern_append(wc, insert_opts);
   coll = mongoc_client_get_collection(client, "mydb1", "foo");
   doc = BCON_NEW("abc", BCON_INT32(0));
   ret = mongoc_collection_insert_one(coll, doc, insert_opts, NULL /* reply */, error);
   if (!ret) {
      goto fail;
   }
   bson_destroy(doc);
   mongoc_collection_destroy(coll);
   coll = mongoc_client_get_collection(client, "mydb2", "bar");
   doc = BCON_NEW("xyz", BCON_INT32(0));
   ret = mongoc_collection_insert_one(coll, doc, insert_opts, NULL /* reply */, error);
   if (!ret) {
      goto fail;
   }

   /* Step 1: Start a client session.第1步：启动客户端会话。 */
   session = mongoc_client_start_session(client, NULL /* opts */, error);
   if (!session) {
      goto fail;
   }

   /* Step 2: Optional. Define options to use for the transaction.第2步：可选。定义用于事务的选项。 */
   txn_opts = mongoc_transaction_opts_new();
   mongoc_transaction_opts_set_write_concern(txn_opts, wc);

   /* Step 3: Use mongoc_client_session_with_transaction to start a transaction,
    * execute the callback, and commit (or abort on error). */
   ret = mongoc_client_session_with_transaction(session, callback, txn_opts, NULL /* ctx */, NULL /* reply */, error);
   if (!ret) {
      goto fail;
   }

   success = true;
fail:
   bson_destroy(doc);
   mongoc_collection_destroy(coll);
   bson_destroy(insert_opts);
   mongoc_write_concern_destroy(wc);
   mongoc_transaction_opts_destroy(txn_opts);
   mongoc_client_session_destroy(session);
   mongoc_client_destroy(client);
   return success;
}

/* Define the callback that specifies the sequence of operations to perform
 * inside the transactions. */
static bool
callback(mongoc_client_session_t *session, void *ctx, bson_t **reply, bson_error_t *error)
{
   mongoc_client_t *client = NULL;
   mongoc_collection_t *coll = NULL;
   bson_t *doc = NULL;
   bool success = false;
   bool ret = false;

   BSON_UNUSED(ctx);

   client = mongoc_client_session_get_client(session);
   coll = mongoc_client_get_collection(client, "mydb1", "foo");
   doc = BCON_NEW("abc", BCON_INT32(1));
   ret = mongoc_collection_insert_one(coll, doc, NULL /* opts */, *reply, error);
   if (!ret) {
      goto fail;
   }
   bson_destroy(doc);
   mongoc_collection_destroy(coll);
   coll = mongoc_client_get_collection(client, "mydb2", "bar");
   doc = BCON_NEW("xyz", BCON_INT32(999));
   ret = mongoc_collection_insert_one(coll, doc, NULL /* opts */, *reply, error);
   if (!ret) {
      goto fail;
   }

   success = true;
fail:
   mongoc_collection_destroy(coll);
   bson_destroy(doc);
   return success;
}

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

// The mongocxx::instance constructor and destructor initialize and shut down the driver,
// respectively. Therefore, a mongocxx::instance must be created before using the driver and
// must remain alive for as long as the driver is in use.
mongocxx::instance inst{};

// For a replica set, include the replica set name and a seedlist of the members in the URI
// string; e.g.
// uriString =
// 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
// For a sharded cluster, connect to the mongos instances; e.g.
// uriString = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'
mongocxx::client client{mongocxx::uri{}};

// Prepare to set majority write explicitly. Note: on Atlas deployments this won't always be
// needed. The suggested Atlas connection string includes majority write concern by default.
write_concern wc_majority{};
wc_majority.acknowledge_level(write_concern::level::k_majority);

// Prereq: Create collections.

auto foo = client["mydb1"]["foo"];
auto bar = client["mydb2"]["bar"];

try {
    options::insert opts;
    opts.write_concern(wc_majority);

    foo.insert_one(make_document(kvp("abc", 0)), opts);
    bar.insert_one(make_document(kvp("xyz", 0)), opts);
} catch (mongocxx::exception const& e) {
    std::cout << "An exception occurred while inserting: " << e.what() << std::endl;
    return EXIT_FAILURE;
}

// Step 1: Define the callback that specifies the sequence of operations to perform inside the
// transactions.
client_session::with_transaction_cb callback = [&](client_session* session) {
    // Important::  You must pass the session to the operations.
    foo.insert_one(*session, make_document(kvp("abc", 1)));
    bar.insert_one(*session, make_document(kvp("xyz", 999)));
};

// Step 2: Start a client session
auto session = client.start_session();

// Step 3: Use with_transaction to start a transaction, execute the callback,
// and commit (or abort on error).
try {
    options::transaction opts;
    opts.write_concern(wc_majority);

    session.with_transaction(callback, opts);
} catch (mongocxx::exception const& e) {
    std::cout << "An exception occurred: " << e.what() << std::endl;
    return EXIT_FAILURE;
}

return EXIT_SUCCESS;

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

// For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
// string uri = "mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl";
// For a sharded cluster, connect to the mongos instances; e.g.
// string uri = "mongodb://mongos0.example.com:27017,mongos1.example.com:27017/";
var client = new MongoClient(connectionString);

// Prereq: Create collections.
var database1 = client.GetDatabase("mydb1");
var collection1 = database1.GetCollection<BsonDocument>("foo").WithWriteConcern(WriteConcern.WMajority);
collection1.InsertOne(new BsonDocument("abc", 0));

var database2 = client.GetDatabase("mydb2");
var collection2 = database2.GetCollection<BsonDocument>("bar").WithWriteConcern(WriteConcern.WMajority);
collection2.InsertOne(new BsonDocument("xyz", 0));

// Step 1: Start a client session.步骤1：启动客户端会话。
using (var session = client.StartSession())
{
    // Step 2: Optional. Define options to use for the transaction.步骤2：可选。定义用于事务的选项。
    var transactionOptions = new TransactionOptions(
        writeConcern: WriteConcern.WMajority);

    // Step 3: Define the sequence of operations to perform inside the transactions步骤3：定义在事务中执行的操作顺序
    var cancellationToken = CancellationToken.None; // normally a real token would be used通常会使用真实的令牌
    result = session.WithTransaction(
        (s, ct) =>
        {
            try
            {
                collection1.InsertOne(s, new BsonDocument("abc", 1), cancellationToken: ct);
                collection2.InsertOne(s, new BsonDocument("xyz", 999), cancellationToken: ct);
            }
            catch (MongoWriteException)
            {
                // Do something in response to the exception
                throw; // NOTE: You must rethrow the exception otherwise an infinite loop can occur.
            }
            return "Inserted into collections in different databases";
        },
        transactionOptions,
        cancellationToken);
}

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

// WithTransactionExample is an example of using the Session.WithTransaction function.
func WithTransactionExample(ctx context.Context) error {
	// For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
	// uri := "mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl"
	// For a sharded cluster, connect to the mongos instances; e.g.
	// uri := "mongodb://mongos0.example.com:27017,mongos1.example.com:27017/"
	uri := mtest.ClusterURI()

	clientOpts := options.Client().ApplyURI(uri)
	client, err := mongo.Connect(clientOpts)
	if err != nil {
		return err
	}
	defer func() { _ = client.Disconnect(ctx) }()

	// Prereq: Create collections.
	wcMajority := writeconcern.Majority()
	wcMajorityCollectionOpts := options.Collection().SetWriteConcern(wcMajority)
	fooColl := client.Database("mydb1").Collection("foo", wcMajorityCollectionOpts)
	barColl := client.Database("mydb1").Collection("bar", wcMajorityCollectionOpts)

	// Step 1: Define the callback that specifies the sequence of operations to perform inside the transaction.
	callback := func(sesctx context.Context) (any, error) {
		// Important: You must pass sesctx as the Context parameter to the operations for them to be executed in the
		// transaction.
		if _, err := fooColl.InsertOne(sesctx, bson.D{{"abc", 1}}); err != nil {
			return nil, err
		}
		if _, err := barColl.InsertOne(sesctx, bson.D{{"xyz", 999}}); err != nil {
			return nil, err
		}

		return nil, nil
	}

	// Step 2: Start a session and run the callback using WithTransaction.
	session, err := client.StartSession()
	if err != nil {
		return err
	}
	defer session.EndSession(ctx)

	result, err := session.WithTransaction(ctx, callback)
	if err != nil {
		return err
	}
	log.Printf("result: %v\n", result)
	return nil
}

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

/*
  For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
  String uri = "mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/admin?replicaSet=myRepl";
  For a sharded cluster, connect to the mongos instances.
  For example:
  String uri = "mongodb://mongos0.example.com:27017,mongos1.example.com:27017:27017/admin";
 */

final MongoClient client = MongoClients.create(uri);

/*
    Create collections.
 */

client.getDatabase("mydb1").getCollection("foo")
        .withWriteConcern(WriteConcern.MAJORITY).insertOne(new Document("abc", 0));
client.getDatabase("mydb2").getCollection("bar")
        .withWriteConcern(WriteConcern.MAJORITY).insertOne(new Document("xyz", 0));

/* Step 1: Start a client session.步骤1：启动客户端会话。 */

final ClientSession clientSession = client.startSession();

/* Step 2: Optional. Define options to use for the transaction.步骤2：可选。定义用于事务的选项。 */

TransactionOptions txnOptions = TransactionOptions.builder()
        .writeConcern(WriteConcern.MAJORITY)
        .build();

/* Step 3: Define the sequence of operations to perform inside the transactions.步骤3：定义要在事务内部执行的操作顺序。 */

TransactionBody txnBody = new TransactionBody<String>() {
    public String execute() {
        MongoCollection<Document> coll1 = client.getDatabase("mydb1").getCollection("foo");
        MongoCollection<Document> coll2 = client.getDatabase("mydb2").getCollection("bar");

        /*
           Important:: You must pass the session to the operations.
         */
        coll1.insertOne(clientSession, new Document("abc", 1));
        coll2.insertOne(clientSession, new Document("xyz", 999));
        return "Inserted into collections in different databases";
    }
};
try {
    /*
       Step 4: Use .withTransaction() to start a transaction,
       execute the callback, and commit (or abort on error).执行回调，并提交（或在出错时中止）。
    */

    clientSession.withTransaction(txnBody, txnOptions);
} catch (RuntimeException e) {
    // some error handling
} finally {
    clientSession.close();
}

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:本示例重点介绍了事务API的关键组件。特别是，它使用回调API。回调API：

• starts a transaction启动事务
• executes the specified operations执行指定的操作
• commits the result or ends the transaction on error提交结果或在出现错误时结束事务

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

# For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
# uriString = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
# For a sharded cluster, connect to the mongos instances; e.g.
# uriString = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'

client = AsyncIOMotorClient(uriString)
wc_majority = WriteConcern("majority", wtimeout=1000)

# Prereq: Create collections.
await client.get_database("mydb1", write_concern=wc_majority).foo.insert_one({"abc": 0})
await client.get_database("mydb2", write_concern=wc_majority).bar.insert_one({"xyz": 0})

# Step 1: Define the callback that specifies the sequence of operations to perform inside the transactions.
async def callback(my_session):
    collection_one = my_session.client.mydb1.foo
    collection_two = my_session.client.mydb2.bar

    # Important:: You must pass the session to the operations.
    await collection_one.insert_one({"abc": 1}, session=my_session)
    await collection_two.insert_one({"xyz": 999}, session=my_session)

# Step 2: Start a client session.
async with await client.start_session() as session:
    # Step 3: Use with_transaction to start a transaction, execute the callback, and commit (or abort on error).
    await session.with_transaction(
        callback,
        read_concern=ReadConcern("local"),
        write_concern=wc_majority,
        read_preference=ReadPreference.PRIMARY,
    )

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:本示例重点介绍了事务API的关键组件。特别是，它使用回调API。回调API：

• starts a transaction启动事务
• executes the specified operations执行指定的操作
• commits the result or ends the transaction on error提交结果或在出现错误时结束事务

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). 服务器端操作中的错误，如DuplicateKeyError，可能会结束事务，并导致命令错误，提醒用户事务已结束。即使客户端从未调用Session.abortTransaction()，也会发生这种行为。To incorporate custom error handling, use the Core API on your transaction.若要合并自定义错误处理，请在事务中使用Core API。

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.回调API包含了针对某些错误的重试逻辑。驱动程序在TransientTransactionError或UnknownTransactionCommitResult提交错误后尝试重新运行事务。

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.从MongoDB 6.2开始，如果服务器收到TransactionTooLargeForCache错误，则不会重试事务。

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.从MongoDB 8.1开始，如果在多文档事务中运行upsert操作，则当遇到重复键错误时，upsert不会重试。

  // For a replica set, include the replica set name and a seedlist of the members in the URI string;对于副本集，在URI字符串中包含副本集名称和成员的种子列表； e.g.
  // const uri = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
  // For a sharded cluster, connect to the mongos instances;对于分片集群，连接到mongos实例； e.g.
  // const uri = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'

  const client = new MongoClient(uri);
  await client.connect();

  // Prereq: Create collections.Prereq：创建集合。

  await client
    .db('mydb1')
    .collection('foo')
    .insertOne({ abc: 0 }, { writeConcern: { w: 'majority' } });

  await client
    .db('mydb2')
    .collection('bar')
    .insertOne({ xyz: 0 }, { writeConcern: { w: 'majority' } });

  // Step 1: Start a Client Session
  const session = client.startSession();

  // Step 2: Optional. Define options to use for the transaction步骤2：可选。定义用于事务的选项
  const transactionOptions = {
    readPreference: 'primary',
    readConcern: { level: 'local' },
    writeConcern: { w: 'majority' }
  };

  // Step 3: Use withTransaction to start a transaction, execute the callback, and commit (or abort on error)步骤3：使用withTransaction启动事务，执行回调，并提交（或在出错时中止）
  // Note: The callback for withTransaction MUST be async and/or return a Promise.注意：withTransaction的回调必须是异步的和/或返回一个Promise。
  try {
    await session.withTransaction(async () => {
      const coll1 = client.db('mydb1').collection('foo');
      const coll2 = client.db('mydb2').collection('bar');

      // Important:: You must pass the session to the operations重要提示：您必须将会话传递给操作

      await coll1.insertOne({ abc: 1 }, { session });
      await coll2.insertOne({ xyz: 999 }, { session });
    }, transactionOptions);
  } finally {
    await session.endSession();
    await client.close();
  }

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

sub runTransactionWithRetry {
    my ( $txnFunc, $session ) = @_;

    LOOP: {
        eval {
            $txnFunc->($session); # performs transaction
        };
        if ( my $error = $@ ) {
            print("Transaction aborted-> Caught exception during transaction.\n");
            # If transient error, retry the whole transaction
            if ( $error->has_error_label("TransientTransactionError") ) {
                print("TransientTransactionError, retrying transaction ->..\n");
                redo LOOP;
            }
            else {
                die $error;
            }
        }
    }

    return;
}

sub commitWithRetry {
    my ($session) = @_;

    LOOP: {
        eval {
            $session->commit_transaction(); # Uses write concern set at transaction start.
            print("Transaction committed->\n");
        };
        if ( my $error = $@ ) {
            # Can retry commit
            if ( $error->has_error_label("UnknownTransactionCommitResult") ) {
                print("UnknownTransactionCommitResult, retrying commit operation ->..\n");
                redo LOOP;
            }
            else {
                print("Error during commit ->..\n");
                die $error;
            }
        }
    }

    return;
}

# Updates two collections in a transactions

sub updateEmployeeInfo {
    my ($session) = @_;
    my $employeesCollection = $session->client->ns("hr.employees");
    my $eventsCollection    = $session->client->ns("reporting.events");

    $session->start_transaction(
        {
            readConcern  => { level => "snapshot" },
            writeConcern => { w     => "majority" },
            readPreference => 'primary',
        }
    );

    eval {
        $employeesCollection->update_one(
            { employee => 3 }, { '$set' => { status => "Inactive" } },
            { session => $session},
        );
        $eventsCollection->insert_one(
            { employee => 3, status => { new => "Inactive", old => "Active" } },
            { session => $session},
        );
    };
    if ( my $error = $@ ) {
        print("Caught exception during transaction, aborting->\n");
        $session->abort_transaction();
        die $error;
    }

    commitWithRetry($session);
}

# Start a session
my $session = $client->start_session();

eval {
    runTransactionWithRetry(\&updateEmployeeInfo, $session);
};
if ( my $error = $@ ) {
    # Do something with error
}

$session->end_session();

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

/*
 * For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
 * uriString = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
 * For a sharded cluster, connect to the mongos instances; e.g.
 * uriString = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'
 */

$client = new \MongoDB\Client($uriString);

// Prerequisite: Create collections.
$client->selectCollection(
    'mydb1',
    'foo',
    [
        'writeConcern' => new \MongoDB\Driver\WriteConcern(\MongoDB\Driver\WriteConcern::MAJORITY, 1000),
    ],
)->insertOne(['abc' => 0]);

$client->selectCollection(
    'mydb2',
    'bar',
    [
        'writeConcern' => new \MongoDB\Driver\WriteConcern(\MongoDB\Driver\WriteConcern::MAJORITY, 1000),
    ],
)->insertOne(['xyz' => 0]);

// Step 1: Define the callback that specifies the sequence of operations to perform inside the transactions.

$callback = function (\MongoDB\Driver\Session $session) use ($client): void {
    $client
        ->selectCollection('mydb1', 'foo')
        ->insertOne(['abc' => 1], ['session' => $session]);

    $client
        ->selectCollection('mydb2', 'bar')
        ->insertOne(['xyz' => 999], ['session' => $session]);
};

// Step 2: Start a client session.

$session = $client->startSession();

// Step 3: Use with_transaction to start a transaction, execute the callback, and commit (or abort on error).

\MongoDB\with_transaction($session, $callback);

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction启动事务
• executes the specified operations执行指定的操作
• commits the result or ends the transaction on error提交结果或在出现错误时结束事务

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

# For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
# uriString = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
# For a sharded cluster, connect to the mongos instances; e.g.
# uriString = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'

client = MongoClient(uriString)
wc_majority = WriteConcern("majority", wtimeout=1000)

# Prereq: Create collections.
client.get_database("mydb1", write_concern=wc_majority).foo.insert_one({"abc": 0})
client.get_database("mydb2", write_concern=wc_majority).bar.insert_one({"xyz": 0})

# Step 1: Define the callback that specifies the sequence of operations to perform inside the transactions.
def callback(session):
    collection_one = session.client.mydb1.foo
    collection_two = session.client.mydb2.bar

    # Important:: You must pass the session to the operations.
    collection_one.insert_one({"abc": 1}, session=session)
    collection_two.insert_one({"xyz": 999}, session=session)

# Step 2: Start a client session.
with client.start_session() as session:
    # Step 3: Use with_transaction to start a transaction, execute the callback, and commit (or abort on error).
    session.with_transaction(callback)

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

# For a replica set, include the replica set name and a seedlist of the members in the URI string; e.g.
# uriString = 'mongodb://mongodb0.example.com:27017,mongodb1.example.com:27017/?replicaSet=myRepl'
# For a sharded cluster, connect to the mongos instances; e.g.
# uri_string = 'mongodb://mongos0.example.com:27017,mongos1.example.com:27017/'

client = Mongo::Client.new(uri_string, write_concern: {w: :majority, wtimeout: 1000})

# Prereq: Create collections.

client.use('mydb1')['foo'].insert_one(abc: 0)
client.use('mydb2')['bar'].insert_one(xyz: 0)

# Step 1: Define the callback that specifies the sequence of operations to perform inside the transactions.

callback = Proc.new do |my_session|
  collection_one = client.use('mydb1')['foo']
  collection_two = client.use('mydb2')['bar']

  # Important: You must pass the session to the operations.

  collection_one.insert_one({'abc': 1}, session: my_session)
  collection_two.insert_one({'xyz': 999}, session: my_session)
end

#. Step 2: Start a client session.

session = client.start_session

# Step 3: Use with_transaction to start a transaction, execute the callback, and commit (or abort on error).

session.with_transaction(
  read_concern: {level: :local},
  write_concern: {w: :majority, wtimeout: 1000},
  read: {mode: :primary},
  &callback)

This example highlights the key components of the transactions API. In particular, it uses the callback API. The callback API:

• starts a transaction
• executes the specified operations
• commits the result or ends the transaction on error

Errors in server-side operations, such as the DuplicateKeyError, can end the transaction and result in a command error to alert the user that the transaction has ended. This behavior is expected and occurs even if the client never calls Session.abortTransaction(). To incorporate custom error handling, use the Core API on your transaction.

The callback API incorporates retry logic for certain errors. The driver tries to rerun the transaction after a TransientTransactionError or UnknownTransactionCommitResult commit error.

Starting in MongoDB 6.2, the server does not retry the transaction if it receives a TransactionTooLargeForCache error.

Starting in MongoDB 8.1, if an upsert operation runs in a multidocument transaction, then the upsert does not retry when it encounters a duplicate key error.

// Prereq: Create collections. CRUD operations in transactions must be on existing collections.

client
    .database("mydb1")
    .collection::<Document>("foo")
    .insert_one(doc! { "abc": 0})
    .await?;
client
    .database("mydb2")
    .collection::<Document>("bar")
    .insert_one(doc! { "xyz": 0})
    .await?;

// Step 1: Define the callback that specifies the sequence of operations to perform inside the
// transaction.
async fn callback(session: &mut ClientSession) -> Result<()> {
    let collection_one = session
        .client()
        .database("mydb1")
        .collection::<Document>("foo");
    let collection_two = session
        .client()
        .database("mydb2")
        .collection::<Document>("bar");

    // Important: You must pass the session to the operations.
    collection_one
        .insert_one(doc! { "abc": 1 })
        .session(&mut *session)
        .await?;
    collection_two
        .insert_one(doc! { "xyz": 999 })
        .session(session)
        .await?;

    Ok(())
}

// Step 2: Start a client session.
let mut session = client.start_session().await?;

// Step 3: Use and_run2 to start a transaction, execute the callback, and commit (or
// abort on error).
session.start_transaction().and_run2(callback).await?;

This example uses the core API. Because the core API does not incorporate retry logic for the TransientTransactionError or UnknownTransactionCommitResult commit errors, the example includes explicit logic to retry the transaction for these errors:

/*
 * Copyright 2008-present MongoDB, Inc.
 *
 * 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
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.mongodb.scala

import org.mongodb.scala.model.{Filters, Updates}
import org.mongodb.scala.result.UpdateResult

import scala.concurrent.Await
import scala.concurrent.duration.Duration


//scalastyle:off magic.number
class DocumentationTransactionsExampleSpec extends RequiresMongoDBISpec {

  // Implicit functions that execute the Observable and return the results
  val waitDuration = Duration(5, "seconds")
  implicit class ObservableExecutor[T](observable: Observable[T]) {
    def execute(): Seq[T] = Await.result(observable.toFuture(), waitDuration)
  }

  implicit class SingleObservableExecutor[T](observable: SingleObservable[T]) {
    def execute(): T = Await.result(observable.toFuture(), waitDuration)
  }
  // end implicit functions

  "The Scala driver" should "be able to commit a transaction" in withClient { client =>
    assume(serverVersionAtLeast(List(4, 0, 0)) && !hasSingleHost())
    client.getDatabase("hr").drop().execute()
    client.getDatabase("hr").createCollection("employees").execute()
    client.getDatabase("hr").createCollection("events").execute()

    updateEmployeeInfoWithRetry(client).execute() should equal(Completed())
    client.getDatabase("hr").drop().execute() should equal(Completed())
  }

  def updateEmployeeInfo(database: MongoDatabase, observable: SingleObservable[ClientSession]): SingleObservable[ClientSession] = {
    observable.map(clientSession => {
      val employeesCollection = database.getCollection("employees")
      val eventsCollection = database.getCollection("events")

      val transactionOptions = TransactionOptions.builder()
        .readPreference(ReadPreference.primary())
        .readConcern(ReadConcern.SNAPSHOT)
        .writeConcern(WriteConcern.MAJORITY)
        .build()
      clientSession.startTransaction(transactionOptions)
      employeesCollection.updateOne(clientSession, Filters.eq("employee", 3), Updates.set("status", "Inactive"))
        .subscribe((res: UpdateResult) => println(res))
      eventsCollection.insertOne(clientSession, Document("employee" -> 3, "status" -> Document("new" -> "Inactive", "old" -> "Active")))
        .subscribe((res: Completed) => println(res))

      clientSession
    })
  }

  def commitAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
    observable.recoverWith({
      case e: MongoException if e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL) => {
        println("UnknownTransactionCommitResult, retrying commit operation ...")
        commitAndRetry(observable)
      }
      case e: Exception => {
        println(s"Exception during commit ...: $e")
        throw e
      }
    })
  }

  def runTransactionAndRetry(observable: SingleObservable[Completed]): SingleObservable[Completed] = {
    observable.recoverWith({
      case e: MongoException if e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL) => {
        println("TransientTransactionError, aborting transaction and retrying ...")
        runTransactionAndRetry(observable)
      }
    })
  }

  def updateEmployeeInfoWithRetry(client: MongoClient): SingleObservable[Completed] = {

    val database = client.getDatabase("hr")
    val updateEmployeeInfoObservable: Observable[ClientSession] = updateEmployeeInfo(database, client.startSession())
    val commitTransactionObservable: SingleObservable[Completed] =
      updateEmployeeInfoObservable.flatMap(clientSession => clientSession.commitTransaction())
    val commitAndRetryObservable: SingleObservable[Completed] = commitAndRetry(commitTransactionObservable)

    runTransactionAndRetry(commitAndRetryObservable)
  }
}