Merge pull request #122 from talis/ent-71-regenerate-composites

ENT-71: regenerate composites without clearing collection first

.editorconfig

@@ -0,0 +1,15 @@
+# This file is for unifying the coding style for different editors and IDEs
+# editorconfig.org
+
+# PHP PSR-2 Coding Standards
+# http://www.php-fig.org/psr/psr-2/
+
+root = true
+
+[*.php]
+charset = utf-8
+end_of_line = lf
+insert_final_newline = true
+trim_trailing_whitespace = true
+indent_style = space
+indent_size = 4

.gitignore

@@ -1,5 +1,6 @@
 .idea
 composer.phar
+composer.lock
 output
 vendor
 tags

.travis.yml

@@ -7,8 +7,10 @@ php:
 - 5.4
 services:
 - redis
+- docker
 before_install:
-- sudo apt-get install -y mongodb-org=2.6.9 mongodb-org-server=2.6.9 mongodb-org-shell=2.6.9 mongodb-org-mongos=2.6.9 mongodb-org-tools=2.6.9
+- docker pull rossfsinger/mongo-2.6.12
+- docker run -d -p 127.0.0.1:27017:27017 -v ~/data:/data/db rossfsinger/mongo-2.6.12:latest
 - echo "extension = mongodb.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
 - sleep 15
 install:

composer.json

@@ -18,10 +18,11 @@
         }
     ],
     "require": {
+        "php" : ">=5.4",
         "semsol/arc2": "v2.2.4",
         "chrisboulton/php-resque": "dev-master#98fde571db008a8b48e73022599d1d1c07d4a7b5",
         "monolog/monolog" : "~1.13",
-        "mongodb/mongodb": "^1.0.0"
+        "mongodb/mongodb": "1.0.4"
     },
     "require-dev": {
         "phpunit/phpunit": "4.1.*"

docker-compose.yml

@@ -0,0 +1,4 @@
+mongo:
+  image: mongo/2.6.12:latest
+  ports:
+      - "27017:27017"

src/mongo/Config.class.php

@@ -2054,6 +2054,19 @@ public function getCollectionForManualRollbackAudit($storeName, $readPreference
         );
     }
 
+    /**
+     * @param string $storeName
+     * @param string $readPreference
+     * @return Collection
+     */
+    public function getCollectionForJobGroups($storeName, $readPreference = ReadPreference::RP_PRIMARY_PREFERRED)
+    {
+        return $this->getMongoCollection(
+            $this->getDatabase($storeName, $this->dbConfig[$storeName]['data_source'], $readPreference),
+            OPERATION_GROUPS_COLLECTION
+        );
+    }
+
     /**
      * @param $readPreference
      * @return Database

src/mongo/JobGroup.php

@@ -0,0 +1,86 @@
+<?php
+
+namespace Tripod\Mongo;
+
+use \MongoDB\BSON\ObjectId;
+
+class JobGroup
+{
+    private $id;
+    private $collection;
+    private $storeName;
+
+    /**
+     * Constructor method
+     * @param string                        $storeName Tripod store (database) name
+     * @param string|ObjectId $groupId   Optional tracking ID, will assign a new one if omitted
+     */
+    public function __construct($storeName, $groupId = null)
+    {
+        $this->storeName = $storeName;
+        if (!$groupId) {
+            $groupId = new ObjectId();
+        } elseif (!$groupId instanceof ObjectId) {
+            $groupId = new ObjectId($groupId);
+        }
+        $this->id = $groupId;
+    }
+
+    /**
+     * Update the number of jobs
+     *
+     * @param integer $count Number of jobs in group
+     * @return void
+     */
+    public function setJobCount($count)
+    {
+        $this->getMongoCollection()->updateOne(
+            ['_id' => $this->getId()],
+            ['$set' => ['count' => $count]],
+            ['upsert' => true]
+        );
+    }
+
+    /**
+     * Update the number of jobs by $inc.  To decrement, use a negative integer
+     *
+     * @param integer $inc Number to increment or decrement by
+     * @return integer Updated job count
+     */
+    public function incrementJobCount($inc = 1)
+    {
+        $updateResult = $this->getMongoCollection()->findOneAndUpdate(
+            ['_id' => $this->getId()],
+            ['$inc' => ['count' => $inc]],
+            ['upsert' => true, 'returnDocument' => \MongoDB\Operation\FindOneAndUpdate::RETURN_DOCUMENT_AFTER]
+        );
+        if (\is_array($updateResult)) {
+            return $updateResult['count'];
+        } elseif (isset($updateResult->count)) {
+            return $updateResult->count;
+        }
+    }
+
+    /**
+     * @return ObjectId
+     */
+    public function getId()
+    {
+        return $this->id;
+    }
+
+    /**
+     * For mocking
+     *
+     * @return \MongoDB\Collection
+     */
+    protected function getMongoCollection()
+    {
+        if (!isset($this->collection)) {
+            $config = Config::getInstance();
+
+            $this->collection = $config->getCollectionForJobGroups($this->storeName);
+        }
+        return $this->collection;
+    }
+}

src/mongo/MongoTripodConstants.php

@@ -6,6 +6,7 @@
 define('VIEWS_COLLECTION', 'views');
 define('LOCKS_COLLECTION', 'locks');
 define('AUDIT_MANUAL_ROLLBACKS_COLLECTION','audit_manual_rollbacks');
+define('OPERATION_GROUPS_COLLECTION', 'job_groups');
 
 // search
 define('SEARCH_INDEX_COLLECTION', 'search');

src/mongo/delegates/SearchDocuments.class.php

@@ -111,7 +111,7 @@ public function generateSearchDocumentBasedOnSpecId($specId, $resource, $context
         $this->debugLog("Processing {$specId}");
 
         // build the document
-        $generatedDocument = array();
+        $generatedDocument = [\_CREATED_TS => DateUtil::getMongoDate()];
         $this->addIdToImpactIndex($_id, $generatedDocument);
 
         $_id['type'] = $specId;

src/mongo/delegates/SearchIndexer.class.php

@@ -10,6 +10,8 @@
 use Tripod\Mongo\Config;
 use Tripod\Mongo\ImpactedSubject;
 use Tripod\Mongo\Labeller;
+use Tripod\Mongo\Jobs\ApplyOperation;
+use Tripod\Mongo\JobGroup;
 use \MongoDB\Driver\ReadPreference;
 use \MongoDB\Collection;
 
@@ -166,6 +168,7 @@ public function generateAndIndexSearchDocuments($resourceUri, $context, $podName
      * @param string|null $resourceUri
      * @param string|null $context
      * @param string|null $queueName
+     * @return array|null Will return an array with a count and group id, if $queueName is sent and $resourceUri is null
      */
     public function generateSearchDocuments($searchDocumentType, $resourceUri=null, $context=null, $queueName=null)
     {
@@ -174,9 +177,9 @@ public function generateSearchDocuments($searchDocumentType, $resourceUri=null,
         // default the context
         $contextAlias = $this->getContextAlias($context);
         $spec = \Tripod\Mongo\Config::getInstance()->getSearchDocumentSpecification($this->getStoreName(), $searchDocumentType);
-        
+
         if($resourceUri)
-        {            
+        {
             $this->generateAndIndexSearchDocuments($resourceUri, $contextAlias, $spec['from'], $searchDocumentType);
             return;
         }
@@ -205,31 +208,30 @@ public function generateSearchDocuments($searchDocumentType, $resourceUri=null,
             $filter["_id"] = array(_ID_RESOURCE=>$this->labeller->uri_to_alias($resource),_ID_CONTEXT=>$contextAlias);
         }
 
+        $count = $this->config->getCollectionForCBD($this->getStoreName(), $from)->count($filter);
         $docs = $this->config->getCollectionForCBD($this->getStoreName(), $from)->find($filter, array(
             'maxTimeMS' => $this->config->getMongoCursorTimeout()
         ));
-        foreach ($docs as $doc)
-        {
-            if($queueName && !$resourceUri)
-            {
+
+        $jobOptions = [];
+        if ($queueName && !$resourceUri) {
+            $jobOptions['statsConfig'] = $this->getStatsConfig();
+            $jobGroup = new JobGroup($this->storeName);
+            $jobOptions[ApplyOperation::TRACKING_KEY] = $jobGroup->getId()->__toString();
+            $jobGroup->setJobCount($count);
+        }
+        foreach ($docs as $doc) {
+            if ($queueName && !$resourceUri) {
                 $subject = new ImpactedSubject(
                     $doc['_id'],
                     OP_SEARCH,
                     $this->storeName,
                     $from,
                     array($searchDocumentType)
                 );
-                $jobOptions = array();
-
-                if($this->stat || !empty($this->statsConfig))
-                {
-                    $jobOptions['statsConfig'] = $this->getStatsConfig();
-                }
 
                 $this->getApplyOperation()->createJob(array($subject), $queueName, $jobOptions);
-            }
-            else
-            {
+            } else {
                 $this->generateAndIndexSearchDocuments(
                     $doc[_ID_KEY][_ID_RESOURCE],
                     $doc[_ID_KEY][_ID_CONTEXT],
@@ -246,6 +248,12 @@ public function generateSearchDocuments($searchDocumentType, $resourceUri=null,
             'filter'=>$filter,
             'from'=>$from));
         $this->getStat()->timer(MONGO_CREATE_SEARCH_DOC.".$searchDocumentType",$t->result());
+
+        $stat = ['count' => $count];
+        if (isset($jobOptions[ApplyOperation::TRACKING_KEY])) {
+            $stat[ApplyOperation::TRACKING_KEY] = $jobOptions[ApplyOperation::TRACKING_KEY];
+        }
+        return $stat;
     }
 
     /**
@@ -266,7 +274,7 @@ public function deleteSearchDocumentsByTypeId($typeId)
     {
     	return $this->getSearchProvider()->deleteSearchDocumentsByTypeId($typeId);
     }
-    
+
 
     /**
      * @return \Tripod\ISearchProvider
@@ -300,4 +308,4 @@ protected function deDupe(Array $input)
         }
         return $output;
     }
-}
+}