Add transaction support

.travis.yml

@@ -18,10 +18,11 @@ cache:
 
 install:
   - docker version
-  - sudo pip install docker-compose
+  - sudo curl -L "https://github.com/docker/compose/releases/download/1.25.0/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose
   - docker-compose version
   - docker-compose build --build-arg PHP_VERSION=${PHP_VERSION}
   - docker-compose run --rm tests composer install --no-interaction
 
 script:
-  - docker-compose run --rm tests ./vendor/bin/phpunit --coverage-clover ./clover.xml
+  - docker-compose up -d
+  - docker-compose exec tests ./vendor/bin/phpunit

MongoReplsetDockerFile

@@ -0,0 +1,5 @@
+FROM mongo
+
+ADD mongo-replset-init.sh /usr/local/bin/
+
+RUN chmod +x /usr/local/bin/mongo-replset-init.sh

README.md

@@ -13,6 +13,7 @@ Table of contents
 * [Eloquent](#eloquent)
 * [Optional: Alias](#optional-alias)
 * [Query Builder](#query-builder)
+* [Transaction](#transaction)
 * [Schema](#schema)
 * [Extensions](#extensions)
 * [Troubleshooting](#troubleshooting)
@@ -241,6 +242,26 @@ $user = DB::connection('mongodb')->collection('users')->get();
 
 Read more about the query builder on http://laravel.com/docs/queries
 
+Transaction
+-------------
+Transaction requires mongodb version V4.0 or more and deployment replica sets or sharded clusters.
+
+Transaction supports create/insert,update,delete,etc operation.
+
+Transaction supports infinite-level nested transactions, but outside transaction rollbacks do not affect the commit of inside transactions.
+```php
+DB::beginTransaction();
+
+User::create(['name' => 'klinson', 'age' => 20, 'title' => 'admin']);
+
+DB::transaction(function () {
+    DB::collection('users')->where('name', 'klinson')->update(['age' => 20]);
+});
+
+DB::rollBack();
+//DB::commit();
+```
+
 Schema
 ------
 

docker-compose.yml

@@ -12,6 +12,11 @@ services:
         depends_on:
           - mongodb
           - mysql
+          - mongo1
+          - mongo2
+          - mongo3
+        stdin_open: true
+        tty: true
 
     mysql:
         container_name: mysql
@@ -30,3 +35,46 @@ services:
             - 27017:27017
         logging:
             driver: none
+
+    mongo1:
+        hostname: mongo1
+        container_name: mongo1
+        image: mongo
+        restart: always
+        expose:
+            - 27017
+        ports:
+            - 27018:27017
+        command: '--quiet --bind_ip_all --replSet rs0'
+
+    mongo2:
+        hostname: mongo2
+        container_name: mongo2
+        image: mongo
+        restart: always
+        expose:
+            - 27017
+        ports:
+            - 27019:27017
+        command: '--quiet --bind_ip_all --replSet rs0'
+
+    mongo3:
+        hostname: mongo3
+        container_name: mongo3
+        image: mongo
+        restart: always
+        expose:
+            - 27017
+        ports:
+            - 27020:27017
+        command: '--quiet --bind_ip_all --replSet rs0'
+
+    mongoreplret:
+        build:
+            context: .
+            dockerfile: MongoReplsetDockerFile
+        depends_on:
+            - mongo1
+            - mongo2
+            - mongo3
+        entrypoint: ["sh", "-c", "mongo-replset-init.sh"]

mongo-replset-init.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+
+echo "prepare rs initiating"
+
+check_db_status() {
+  mongo1=$(mongo --host mongo1 --port 27017 --eval "db.stats().ok" | tail -n1 | grep -E '(^|\s)1($|\s)')
+  mongo2=$(mongo --host mongo2 --port 27017 --eval "db.stats().ok" | tail -n1 | grep -E '(^|\s)1($|\s)')
+  mongo3=$(mongo --host mongo3 --port 27017 --eval "db.stats().ok" | tail -n1 | grep -E '(^|\s)1($|\s)')
+  if [[ $mongo1 == 1 ]] && [[ $mongo2 == 1 ]] && [[ $mongo3 == 1 ]]; then
+    init_rs
+  else
+    check_db_status
+  fi
+}
+
+init_rs() {
+  ret=$(mongo --host mongo1 --port 27017 --eval "rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: 'mongo1:27017' }, { _id: 1, host: 'mongo2:27017' }, { _id: 2, host: 'mongo3:27017' } ] })" > /dev/null 2>&1)
+}
+
+check_db_status > /dev/null 2>&1
+
+echo "rs initiating finished"
+exit 0

src/Jenssegers/Mongodb/Connection.php

@@ -5,6 +5,9 @@
 use Illuminate\Database\Connection as BaseConnection;
 use Illuminate\Support\Arr;
 use MongoDB\Client;
+use MongoDB\Driver\ReadConcern;
+use MongoDB\Driver\ReadPreference;
+use MongoDB\Driver\WriteConcern;
 
 class Connection extends BaseConnection
 {
@@ -20,6 +23,9 @@ class Connection extends BaseConnection
      */
     protected $connection;
 
+    protected $session_key; // sessions会话列表当前会话数组key 随机生成
+    protected $sessions = []; // 会话列表
+
     /**
      * Create a new database connection instance.
      * @param array $config
@@ -254,4 +260,81 @@ public function __call($method, $parameters)
     {
         return call_user_func_array([$this->db, $method], $parameters);
     }
+
+    /**
+     * create a 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.
+     * To use transactions on MongoDB 4.2 deployments(replica sets and sharded clusters), clients must use MongoDB drivers updated for MongoDB 4.2.
+     *
+     * @see https://docs.mongodb.com/manual/core/transactions/
+     * @author klinson <[email protected]>
+     */
+    public function beginTransaction()
+    {
+        $this->session_key = uniqid();
+        $this->sessions[$this->session_key] = $this->connection->startSession();
+
+        $this->sessions[$this->session_key]->startTransaction([
+            'readPreference' => new ReadPreference(ReadPreference::RP_PRIMARY),
+            'writeConcern' => new WriteConcern(1),
+            'readConcern' => new ReadConcern(ReadConcern::LOCAL)
+        ]);
+    }
+
+    /**
+     * commit transaction in this session and close this session
+     * @author klinson <[email protected]>
+     */
+    public function commit()
+    {
+        if ($session = $this->getSession()) {
+            $session->commitTransaction();
+            $this->setLastSession();
+        }
+    }
+
+    /**
+     * rollback transaction in this session and close this session
+     * @author klinson <[email protected]>
+     */
+    public function rollBack($toLevel = null)
+    {
+        if ($session = $this->getSession()) {
+            $session->abortTransaction();
+            $this->setLastSession();
+        }
+    }
+
+    /**
+     * close this session and get last session key to session_key
+     * Why do it ? Because nested transactions
+     * @author klinson <[email protected]>
+     */
+    protected function setLastSession()
+    {
+        if ($session = $this->getSession()) {
+            $session->endSession();
+            unset($this->sessions[$this->session_key]);
+            if (empty($this->sessions)) {
+                $this->session_key = null;
+            } else {
+                end($this->sessions);
+                $this->session_key = key($this->sessions);
+            }
+        }
+    }
+
+    /**
+     * get now session if it has session
+     * @return \MongoDB\Driver\Session|null
+     * @author klinson <[email protected]>
+     */
+    public function getSession()
+    {
+        return ($this->session_key && isset($this->sessions[$this->session_key]))
+            ? $this->sessions[$this->session_key]
+            : null;
+    }
 }

src/Jenssegers/Mongodb/Query/Builder.php

@@ -294,7 +294,7 @@ public function getFresh($columns = [])
                     }
                 }
             }
-            
+
             // The _id field is mandatory when using grouping.
             if ($group && empty($group['_id'])) {
                 $group['_id'] = null;
@@ -338,6 +338,11 @@ public function getFresh($columns = [])
                 $options = array_merge($options, $this->options);
             }
 
+            // if transaction in session
+            if ($session = $this->connection->getSession()) {
+                $options['session']  =  $session;
+            }
+
             // Execute aggregation
             $results = iterator_to_array($this->collection->aggregate($pipeline, $options));
 
@@ -348,13 +353,15 @@ public function getFresh($columns = [])
             // Return distinct results directly
             $column = isset($this->columns[0]) ? $this->columns[0] : '_id';
 
-            // Execute distinct
-            if ($wheres) {
-                $result = $this->collection->distinct($column, $wheres);
-            } else {
-                $result = $this->collection->distinct($column);
+            $options = [];
+            // if transaction in session
+            if ($session = $this->connection->getSession()) {
+                $options['session']  =  $session;
             }
 
+            // Execute distinct
+            $result = $this->collection->distinct($column, $wheres ?: [], $options);
+
             return $this->useCollections ? new Collection($result) : $result;
         } // Normal query
         else {
@@ -397,6 +404,11 @@ public function getFresh($columns = [])
                 $options = array_merge($options, $this->options);
             }
 
+            // if transaction in session
+            if ($session = $this->connection->getSession()) {
+                $options['session']  =  $session;
+            }
+
             // Execute query and get MongoCursor
             $cursor = $this->collection->find($wheres, $options);
 
@@ -561,7 +573,12 @@ public function insert(array $values)
         }
 
         // Batch insert
-        $result = $this->collection->insertMany($values);
+        $options = [];
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+        $result = $this->collection->insertMany($values, $options);
 
         return (1 == (int) $result->isAcknowledged());
     }
@@ -571,7 +588,12 @@ public function insert(array $values)
      */
     public function insertGetId(array $values, $sequence = null)
     {
-        $result = $this->collection->insertOne($values);
+        $options = [];
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+        $result = $this->collection->insertOne($values, $options);
 
         if (1 == (int) $result->isAcknowledged()) {
             if ($sequence === null) {
@@ -592,6 +614,10 @@ public function update(array $values, array $options = [])
         if (!Str::startsWith(key($values), '$')) {
             $values = ['$set' => $values];
         }
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
 
         return $this->performUpdate($values, $options);
     }
@@ -614,6 +640,11 @@ public function increment($column, $amount = 1, array $extra = [], array $option
             $query->orWhereNotNull($column);
         });
 
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+
         return $this->performUpdate($query, $options);
     }
 
@@ -673,7 +704,14 @@ public function delete($id = null)
         }
 
         $wheres = $this->compileWheres();
-        $result = $this->collection->DeleteMany($wheres);
+
+        $options = [];
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+
+        $result = $this->collection->DeleteMany($wheres, $options);
         if (1 == (int) $result->isAcknowledged()) {
             return $result->getDeletedCount();
         }
@@ -698,7 +736,12 @@ public function from($collection, $as = null)
      */
     public function truncate()
     {
-        $result = $this->collection->drop();
+        $options = [];
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+        $result = $this->collection->drop($options);
 
         return (1 == (int) $result->ok);
     }
@@ -826,6 +869,11 @@ protected function performUpdate($query, array $options = [])
             $options['multiple'] = true;
         }
 
+        // if transaction in session
+        if ($session = $this->connection->getSession()) {
+            $options['session']  =  $session;
+        }
+
         $wheres = $this->compileWheres();
         $result = $this->collection->UpdateMany($wheres, $query, $options);
         if (1 == (int) $result->isAcknowledged()) {

tests/TestCase.php

@@ -53,6 +53,7 @@ protected function getEnvironmentSetUp($app)
         $app['config']->set('database.connections.mongodb', $config['connections']['mongodb']);
         $app['config']->set('database.connections.mongodb2', $config['connections']['mongodb']);
         $app['config']->set('database.connections.dsn_mongodb', $config['connections']['dsn_mongodb']);
+        $app['config']->set('database.connections.mongodb_replset', $config['connections']['mongodb_replset']);
 
         $app['config']->set('auth.model', 'User');
         $app['config']->set('auth.providers.users.model', 'User');