Allow chaining on addJob

Author: manobi

examples/fast.js

@@ -1,15 +1,21 @@
 const fetch = require('node-fetch'),
     { worker, createTask } = require('..')(),
     manobi = worker('manobi-fast');
-    
+
 manobi.addJob('compress', (ctx, done) => {
     done(null, {sucesso: 'muleque'});
 });
 
 (async () => {
-    return createTask({
-        date: new Date().toISOString(),
-        job: 'compress',
-        worker: 'manobi-fast'
+    await manobi.start().then(() => {
+        return manobi.stop();
     });
+    // return createTask({
+    //     date: new Date().toISOString(),
+    //     job: 'compress',
+    //     worker: 'manobi-fast'
+    // }).then(() => {
+    //     manobi.tasks.stop();
+    //     return manobi.storage.unsubscribe();
+    // });
 })();

lib/couchdb-adaptor.js

@@ -17,6 +17,7 @@ module.exports = class Adaptor {
     constructor(){
         this.name = undefined;
         this.onChange = undefined;
+        this.feed = {};
     }
 
     /**
@@ -69,10 +70,10 @@ module.exports = class Adaptor {
     */
     async subscribe(){
          // for cold start
-         const docs = await this.looseTasks(); 
-         docs.forEach((doc) => {
-             this.onChange(this.formatDoc(doc));
-         });
+        const docs = await this.looseTasks(); 
+        docs.forEach((doc) => {
+            this.onChange(this.formatDoc(doc));
+        });
 
         const name = this.name;
         this.feed = this.db.follow({
@@ -92,8 +93,17 @@ module.exports = class Adaptor {
             const task = this.formatDoc(doc, deleted);
             this.onChange(task);
         });
+
         // Start the stream
-        this.feed.follow();
+        return this.feed.follow();
+    }
+
+    /**
+    * @method unsubscribe
+    */
+    unsubscribe(){
+        this.feed.off('change', this.onChange);
+        return this.feed.stop();
     }
 
     /**

lib/worker.js

@@ -23,7 +23,21 @@ module.exports = class Worker extends Scheduler {
         this.storage.name = this.name;
         this.storage.onChange = (change) => this.changeHandler.call(this, change);
         this.storage.connect(options);
-        this.storage.subscribe();
+    }
+    
+    /**
+     * @method start
+     */
+    start(){
+        return this.storage.subscribe();
+    }
+
+    /**
+     * @method stop
+     */
+    stop(){
+        this.tasks.stop();
+        return this.storage.unsubscribe();
     }
 
     enqueue(id, fn){
@@ -56,18 +70,11 @@ module.exports = class Worker extends Scheduler {
     addJob(name, job){
         if(name && job && name.length > 0 && (typeof job === 'function')){
             this.jobs[name] = job;
-            return this.jobs;
+            return this;
         }
         throw new Error('"name {String}" and job {Function} are both required');
     }
 
-     /**
-     * @method exit
-     */
-    exit(){
-        this.storage.unsubscribe()
-    }
-
     /**
      * @method createContext
      * @param {Object} task 

spec/worker.spec.js

@@ -35,6 +35,7 @@ describe('Worker', () => {
             done(null, true);
         });
     });
+
     afterEach(() => {
         worker.tasks.stop();
     })
@@ -60,7 +61,7 @@ describe('Worker', () => {
     describe('#exit', () => {
         it('calls adaptor unsubscribe method', () => {
             const spy = sinon.spy(worker.storage, 'unsubscribe');
-            worker.exit()
+            worker.stop();
             expect(spy).to.have.been.called;
         })
     });