Mongo backed queue

packages/collection-observers/.travis.yml

@@ -0,0 +1,5 @@
+language: node_js
+node_js:
+  - "0.10"
+before_install:
+  - "curl -L http://git.io/s0Zu-w | /bin/sh"

packages/collection-observers/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog
+
+## vCurrent
+
+#### 29/04/15 by Rhys Bartels-Waller @rhyslbw
+- Initial commit
+

packages/collection-observers/README.md

@@ -0,0 +1,15 @@
+cfs:collection-observers
+=========================
+
+This is a Meteor package used by
+[CollectionFS](https://github.com/CollectionFS/Meteor-CollectionFS).
+
+You don't need to manually add this package to your app. It is added when you
+add the `cfs:standard-packages` package.
+
+## Overview
+This package triggers an FSCollection to emit events by observing for specific changes. In a multi-instance system this package needs to be running from a single instance to avoid duplicate events.
+
+## Observer Reference:
+- FSCollection document removed
+- All stores complete

packages/collection-observers/collection-observers-tests.js

@@ -0,0 +1,5 @@
+// Write your tests here!
+// Here is an example.
+Tinytest.add('example', function (test) {
+  test.equal(true, true);
+});

packages/collection-observers/collection-observers.js

@@ -0,0 +1,79 @@
+FS.CollectionObservers = {};
+
+FS.CollectionObservers.register = function(fsCollection){
+
+  // Emit "removed" event on collection
+  fsCollection.files.find().observe({
+    removed: function(fsFile) {
+      FS.debug && console.log('Collection Observer:', fsFile._id, 'removed from collection', fsCollection.name);
+      fsCollection.emit('removed', fsFile);
+    }
+  });
+
+  // Observe files that have been stored so we can delete any temp files
+  fsCollection.files.find(getDoneQuery(fsCollection.options.stores)).observe({
+    added: function(fsFile) {
+      FS.debug && console.log('Collection Observer: All stores complete for', fsFile._id, 'on collection', fsCollection.name, 'and temp data exists');
+      fsCollection.emit('allStoresComplete', fsFile);
+    }
+  });
+
+}
+
+/**
+ *  @method getDoneQuery
+ *  @private
+ *  @param {Array} stores - The stores array from the FS.Collection options
+ *
+ *  Returns a selector that will be used to identify files where all
+ *  stores have successfully save or have failed the
+ *  max number of times. The resulting selector
+ *  should be something like this:
+ *
+ *  {
+ *    $and: [
+ *      {
+ *        $or: [
+ *          {
+ *            $and: [
+ *              {
+ *                'copies.storeName': {$ne: null}
+ *              },
+ *              {
+ *                'copies.storeName': {$ne: false}
+ *              }
+ *            ]
+ *          },
+ *          {
+ *            'failures.copies.storeName.doneTrying': true
+ *          }
+ *        ]
+ *      },
+ *      REPEATED FOR EACH STORE
+ *    ]
+ *  }
+ *
+ */
+function getDoneQuery(stores) {
+  var selector = {
+    $and: [{"tempFileAvailable": true}]
+  };
+
+  // Add conditions for all defined stores
+  FS.Utility.each(stores, function(store) {
+    var storeName = store.name;
+    var copyCond = {$or: [{$and: []}]};
+    var tempCond = {};
+    tempCond["copies." + storeName] = {$ne: null};
+    copyCond.$or[0].$and.push(tempCond);
+    tempCond = {};
+    tempCond["copies." + storeName] = {$ne: false};
+    copyCond.$or[0].$and.push(tempCond);
+    tempCond = {};
+    tempCond['failures.copies.' + storeName + '.doneTrying'] = true;
+    copyCond.$or.push(tempCond);
+    selector.$and.push(copyCond);
+  })
+
+  return selector;
+}

packages/collection-observers/package.js

@@ -0,0 +1,23 @@
+Package.describe({
+  name: 'cfs:collection-observers',
+  version: '0.1.0',
+  summary: 'CollectionFS observers trigger the Collection to emit events based on changes made outside the application. Run on a single instance in a multi-instance system',
+  git: '',
+  documentation: 'README.md'
+});
+
+Package.onUse(function(api) {
+  api.versionsFrom('1.1.0.2');
+
+  api.use([
+    'cfs:base-package@0.0.30'
+  ]);
+
+  api.addFiles('collection-observers.js', 'server');
+});
+
+Package.onTest(function(api) {
+  api.use('tinytest');
+  api.use('cfs:collection-observers');
+  api.addFiles('collection-observers-tests.js');
+});

packages/collection/api.common.js

@@ -29,7 +29,12 @@ FS.Collection.prototype.insert = function(fileRef, callback) {
     // so that it is available when FileWorker calls saveCopies.
     // This will also trigger file handling from collection observes.
     else if (Meteor.isServer) {
-      fileObj.createReadStream().pipe(FS.TempStore.createWriteStream(fileObj));
+      // XXX: Intermediate refactor. Wanted to minimise changes until client-side beginStorage is refactored
+
+      var emitted = self.emit('inserted', fileObj, 'server', new Date());
+      if (FS.debug && !emitted) {
+        console.log(fileObj.name() + ' was successfully inserted into the Mongo Collection. You are seeing this informational message because you enabled debugging and you have not defined any listeners for the "inserted" event on the ' + self.name + ' collection.');
+      }
     }
   }
 

packages/collection/common.js

@@ -125,24 +125,20 @@ FS.Collection = function(name, options) {
   // Save the collection reference (we want it without the 'cfs.' prefix and '.filerecord' suffix)
   FS._collections[name] = this;
 
-  // Set up observers
-  Meteor.isServer && FS.FileWorker && FS.FileWorker.observe(this);
+  // Register with Job Manager
+  Meteor.isServer && FS.JobManager && FS.JobManager.register(this);
 
-  // Emit "removed" event on collection
-  self.files.find().observe({
-    removed: function(fileObj) {
-      self.emit('removed', fileObj);
-    }
-  });
+  // Register with Collection Observers
+  Meteor.isServer && FS.CollectionObservers && FS.CollectionObservers.register(this);
 
   // Emit events based on TempStore events
   if (FS.TempStore) {
     FS.TempStore.on('stored', function (fileObj, result) {
-      // When a file is successfully stored into the temp store, we emit an "uploaded" event on the FS.Collection only if the file belongs to this collection
+      // When a file is successfully stored into the temp store, we emit an "tempStoreTransferComplete" event on the FS.Collection only if the file belongs to this collection
       if (fileObj.collectionName === name) {
-        var emitted = self.emit('uploaded', fileObj);
+        var emitted = self.emit('tempStoreTransferComplete', fileObj);
         if (FS.debug && !emitted) {
-          console.log(fileObj.name() + ' was successfully uploaded. You are seeing this informational message because you enabled debugging and you have not defined any listeners for the "uploaded" event on the ' + name + ' collection.');
+          console.log(fileObj.name() + ' was successfully uploaded. You are seeing this informational message because you enabled debugging and you have not defined any listeners for the "tempStoreTransferComplete" event on the ' + name + ' collection.');
         }
       }
     });

packages/collection/package.js

@@ -1,6 +1,6 @@
 Package.describe({
   name: 'cfs:collection',
-  version: '0.5.5',
+  version: '0.5.6',
   summary: 'CollectionFS, FS.Collection object',
   git: 'https://github.com/CollectionFS/Meteor-cfs-collection.git'
 });

packages/file/fsFile-server.js

@@ -1,3 +1,19 @@
+/**
+ * @method FS.File.prototype.beginStorage
+ * @public
+ * @return {undefined}
+ *
+ */
+FS.File.prototype.beginStorage = function() {
+  var self = this;
+
+  // Save the binary to a single chunk temp file,
+  // so that it is available when FileWorker calls saveCopies.
+  // This will also trigger file handling from event listeners.
+  self.createReadStream().pipe(FS.TempStore.createWriteStream(self));
+
+}
+
 /**
  * Notes a details about a storage adapter failure within the file record
  * @param {string} storeName

packages/file/package.js

@@ -1,7 +1,7 @@
 Package.describe({
   git: 'https://github.com/CollectionFS/Meteor-cfs-file.git',
   name: 'cfs:file',
-  version: '0.1.17',
+  version: '0.1.18',
   summary: 'CollectionFS, FS.File object'
 });
 

packages/job-manager/.travis.yml

@@ -0,0 +1,5 @@
+language: node_js
+node_js:
+  - "0.10"
+before_install:
+  - "curl -L http://git.io/s0Zu-w | /bin/sh"

packages/job-manager/CHANGELOG.md

@@ -0,0 +1,7 @@
+# Changelog
+
+## vCurrent
+
+#### 09/04/15 by Rhys Bartels-Waller
+- Initial commit
+

packages/job-manager/README.md

@@ -0,0 +1,19 @@
+cfs:job-manager
+=========================
+
+This is a Meteor package used by
+[CollectionFS](https://github.com/CollectionFS/Meteor-CollectionFS).
+
+You don't need to manually add this package to your app. It is added when you
+add the `cfs:standard-packages` package.
+
+## Overview
+
+The job and queue functionality is an implementation of [vsivsi:job-collection](https://github.com/vsivsi/meteor-job-collection), a "powerful and easy to use job manager designed and built for Meteor.js."
+
+Job Manager creates jobs by listening to events emitted in other cfs packages. These jobs are completed by worker groups established by the [cfs:worker](https://github.com/CollectionFS/Meteor-CollectionFS/tree/master/packages/worker) package.
+
+## Job Task Reference:
+- saveCopy
+- removeTempFile
+- removeStoredData

packages/job-manager/common.js

@@ -0,0 +1,11 @@
+/**
+ * @public
+ * @type Object
+ */
+FS.JobManager = {
+  _registeredJobTypes: [],
+  _registeredJobWorkers: []
+};
+
+// TODO: Allow custom options
+FS.JobManager.jobCollection =  new JobCollection('cfs_jobManager');

packages/job-manager/package.js

@@ -0,0 +1,40 @@
+Package.describe({
+  name: 'cfs:job-manager',
+  version: '0.1.0',
+  summary: 'CollectionFS queue job management add-on',
+  git: '',
+  documentation: 'README.md'
+});
+
+Package.onUse(function(api) {
+  api.versionsFrom('1.0.3.1');
+
+
+  api.use([
+    'cfs:base-package@0.0.29',
+    'cfs:tempstore@0.1.5'
+  ]);
+
+  api.use([
+    'vsivsi:job-collection@1.1.0'
+  ]);
+
+  api.use([
+    'random'
+  ], 'server');
+
+  api.addFiles([
+    'common.js'
+  ], ['client', 'server']);
+
+  api.addFiles([
+    'server.js'
+  ], 'server');
+
+});
+
+Package.onTest(function(api) {
+  api.use('tinytest');
+  api.use('cfs:job-manager');
+  //api.addFiles('tests/server-tests.js');
+});