Add support for transactions

Co-authored-by: klinson <klinson@163.com>
Co-authored-by: levon80999 <levonb@ucraft.com>

Author: 3 people

README.md

@@ -37,6 +37,7 @@ This package adds functionalities to the Eloquent model and Query builder for Mo
   - [Query Builder](#query-builder)
     - [Basic Usage](#basic-usage-2)
     - [Available operations](#available-operations)
+  - [Transactions](#transactions)
   - [Schema](#schema)
     - [Basic Usage](#basic-usage-3)
     - [Geospatial indexes](#geospatial-indexes)
@@ -968,6 +969,46 @@ If you are familiar with [Eloquent Queries](http://laravel.com/docs/queries), th
 ### Available operations
 To see the available operations, check the [Eloquent](#eloquent) section.
 
+Transactions
+-------
+Transactions require MongoDB version ^4.0 as well as deployment of replica set or sharded clusters. You can find more information [in the MongoDB docs](https://docs.mongodb.com/manual/core/transactions/)
+
+### Basic Usage
+
+Transaction supports all operations.
+
+```php
+DB::transaction(function () {
+    User::create(['name' => 'john', 'age' => 19, 'title' => 'admin', 'email' => 'klinsonup@gmail.com']);
+    DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+    DB::collection('users')->where('name', 'john')->delete();
+});
+```
+
+```php
+// begin a transaction
+DB::beginTransaction();
+User::create(['name' => 'john', 'age' => 19, 'title' => 'admin', 'email' => 'klinsonup@gmail.com']);
+DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+DB::collection('users')->where('name', 'john')->delete();
+
+// you can commit your changes
+DB::commit();
+
+// you can also rollback them
+//DB::rollBack();
+```
+**NOTE:** Transactions in MongoDB cannot be nested. DB::beginTransaction() function will start new transactions in a new created or existing session and will raise the RuntimeException when transactions already exist. See more in MongoDB official docs [Transactions and Sessions](https://www.mongodb.com/docs/manual/core/transactions/#transactions-and-sessions)
+```php
+// This code will raise a RuntimeException
+DB::beginTransaction();
+    User::create(['name' => 'john', 'age' => 20, 'title' => 'admin']);
+    DB::beginTransaction()
+        DB::collection('users')->where('name', 'john')->update(['age' => 20]);
+    DB::commit()
+DB::rollBack();
+```
+
 Schema
 ------
 The database driver also has (limited) schema builder support. You can easily manipulate collections and set indexes.

phpunit.xml.dist

@@ -19,6 +19,9 @@
             <file>tests/QueryBuilderTest.php</file>
             <file>tests/QueryTest.php</file>
         </testsuite>
+        <testsuite name="transaction">
+            <file>tests/TransactionTest.php</file>
+        </testsuite>
         <testsuite name="model">
             <file>tests/ModelTest.php</file>
             <file>tests/RelationsTest.php</file>

src/Concerns/ManagesTransactions.php

@@ -0,0 +1,105 @@
+<?php
+
+namespace Jenssegers\Mongodb\Concerns;
+
+use Closure;
+use MongoDB\Client;
+use MongoDB\Driver\Exception\RuntimeException;
+use MongoDB\Driver\Session;
+use function MongoDB\with_transaction;
+
+trait ManagesTransactions
+{
+    protected ?Session $session = null;
+
+    /**
+     * @return Client
+     */
+    abstract public function getMongoClient();
+
+    public function getSession(): ?Session
+    {
+        return $this->session;
+    }
+
+    private function getSessionOrCreate(): Session
+    {
+        if ($this->session === null) {
+            $this->session = $this->getMongoClient()->startSession();
+        }
+
+        return $this->session;
+    }
+
+    private function getSessionOrThrow(): Session
+    {
+        $session = $this->getSession();
+
+        if ($session === null) {
+            throw new RuntimeException('There is no active session.');
+        }
+
+        return $session;
+    }
+
+    /**
+     * Use the existing or create new session and start a transaction in session.
+     *
+     * In version 4.0, MongoDB supports multi-document transactions on replica sets.
+     * In version 4.2, MongoDB introduces distributed transactions, which adds support for multi-document transactions on sharded clusters and incorporates the existing support for multi-document transactions on replica sets.
+     *
+     * @see https://docs.mongodb.com/manual/core/transactions/
+     */
+    public function beginTransaction(array $options = []): void
+    {
+        $this->getSessionOrCreate()->startTransaction($options);
+        $this->transactions = 1;
+    }
+
+    /**
+     * Commit transaction in this session
+     */
+    public function commit(): void
+    {
+        $this->getSessionOrThrow()->commitTransaction();
+        $this->transactions = 0;
+    }
+
+    /**
+     * Rollback transaction in this session
+     */
+    public function rollBack($toLevel = null): void
+    {
+        $this->getSessionOrThrow()->abortTransaction();
+        $this->transactions = 0;
+    }
+
+    /**
+     * Static transaction function realize the with_transaction functionality provided by MongoDB.
+     *
+     * @param int $attempts
+     */
+    public function transaction(Closure $callback, $attempts = 1, array $options = []): mixed
+    {
+        $attemptsLeft = $attempts;
+        $callbackResult = null;
+
+        $session = $this->getSessionOrCreate();
+
+        $callbackFunction = function (Session $session) use ($callback, &$attemptsLeft, &$callbackResult) {
+            $attemptsLeft--;
+
+            if ($attemptsLeft < 0) {
+                $session->abortTransaction();
+
+                return;
+            }
+
+            $callbackResult = $callback();
+        };
+
+        with_transaction($session, $callbackFunction, $options);
+
+        return $callbackResult;
+    }
+}

src/Connection.php

@@ -5,10 +5,13 @@
 use Illuminate\Database\Connection as BaseConnection;
 use Illuminate\Support\Arr;
 use InvalidArgumentException;
+use Jenssegers\Mongodb\Concerns\ManagesTransactions;
 use MongoDB\Client;
 
 class Connection extends BaseConnection
 {
+    use ManagesTransactions;
+
     /**
      * The MongoDB database handler.
      *

src/Query/Builder.php

@@ -345,6 +345,8 @@ public function getFresh($columns = [], $returnLazy = false)
                 $options = array_merge($options, $this->options);
             }
 
+            $options = $this->inheritConnectionOptions($options);
+
             // Execute aggregation
             $results = iterator_to_array($this->collection->aggregate($pipeline, $options));
 
@@ -355,12 +357,10 @@ public function getFresh($columns = [], $returnLazy = false)
             // Return distinct results directly
             $column = isset($this->columns[0]) ? $this->columns[0] : '_id';
 
+            $options = $this->inheritConnectionOptions();
+
             // Execute distinct
-            if ($wheres) {
-                $result = $this->collection->distinct($column, $wheres);
-            } else {
-                $result = $this->collection->distinct($column);
-            }
+            $result = $this->collection->distinct($column, $wheres ?: [], $options);
 
             return new Collection($result);
         } // Normal query
@@ -406,6 +406,8 @@ public function getFresh($columns = [], $returnLazy = false)
                 $options = array_merge($options, $this->options);
             }
 
+            $options = $this->inheritConnectionOptions($options);
+
             // Execute query and get MongoCursor
             $cursor = $this->collection->find($wheres, $options);
 
@@ -580,8 +582,9 @@ public function insert(array $values)
             $values = [$values];
         }
 
-        // Batch insert
-        $result = $this->collection->insertMany($values);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->insertMany($values, $options);
 
         return 1 == (int) $result->isAcknowledged();
     }
@@ -591,7 +594,9 @@ public function insert(array $values)
      */
     public function insertGetId(array $values, $sequence = null)
     {
-        $result = $this->collection->insertOne($values);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->insertOne($values, $options);
 
         if (1 == (int) $result->isAcknowledged()) {
             if ($sequence === null) {
@@ -613,6 +618,8 @@ public function update(array $values, array $options = [])
             $values = ['$set' => $values];
         }
 
+        $options = $this->inheritConnectionOptions($options);
+
         return $this->performUpdate($values, $options);
     }
 
@@ -634,6 +641,8 @@ public function increment($column, $amount = 1, array $extra = [], array $option
             $query->orWhereNotNull($column);
         });
 
+        $options = $this->inheritConnectionOptions($options);
+
         return $this->performUpdate($query, $options);
     }
 
@@ -695,7 +704,10 @@ public function delete($id = null)
         }
 
         $wheres = $this->compileWheres();
-        $result = $this->collection->DeleteMany($wheres);
+        $options = $this->inheritConnectionOptions();
+
+        $result = $this->collection->deleteMany($wheres, $options);
+
         if (1 == (int) $result->isAcknowledged()) {
             return $result->getDeletedCount();
         }
@@ -720,7 +732,8 @@ public function from($collection, $as = null)
      */
     public function truncate(): bool
     {
-        $result = $this->collection->deleteMany([]);
+        $options = $this->inheritConnectionOptions();
+        $result = $this->collection->deleteMany($options);
 
         return 1 === (int) $result->isAcknowledged();
     }
@@ -853,6 +866,8 @@ protected function performUpdate($query, array $options = [])
             $options['multiple'] = true;
         }
 
+        $options = $this->inheritConnectionOptions($options);
+
         $wheres = $this->compileWheres();
         $result = $this->collection->UpdateMany($wheres, $query, $options);
         if (1 == (int) $result->isAcknowledged()) {
@@ -1247,6 +1262,18 @@ public function options(array $options)
         return $this;
     }
 
+    /**
+     * Apply the connection's session to options if it's not already specified.
+     */
+    private function inheritConnectionOptions(array $options = []): array
+    {
+        if (! isset($options['session']) && ($session = $this->connection->getSession())) {
+            $options['session'] = $session;
+        }
+
+        return $options;
+    }
+
     /**
      * @inheritdoc
      */