Separated worker tasks from app http work

.gitignore

@@ -3,4 +3,5 @@
 npm-debug.log
 /test/benchmarks
 *.env
+*.swp
 

Procfile_dev

@@ -0,0 +1,2 @@
+web: nodemon lib/server.js
+worker: nodemon lib/worker.js

lib/app/index.js

@@ -10,6 +10,7 @@ var SCRAPE_QUEUE = 'jobs.scrape';
 var VOTE_QUEUE = 'jobs.vote';
 
 function App(config) {
+  logger.log({ type: 'info', msg: 'app/index.App' });
   EventEmitter.call(this);
 
   this.config = config;
@@ -19,6 +20,7 @@ function App(config) {
 }
 
 module.exports = function createApp(config) {
+  logger.log({ type: 'info', msg: 'app/index.createApp' });
   return new App(config);
 };
 
@@ -51,19 +53,11 @@ App.prototype.addArticle = function(userId, url) {
   return Promise.resolve(id);
 };
 
-App.prototype.scrapeArticle = function(userId, id, url) {
-  return this.Article.scrape(userId, id, url);
-};
-
 App.prototype.addUpvote = function(userId, articleId) {
   this.connections.queue.publish(VOTE_QUEUE, { userId: userId, articleId: articleId });
   return Promise.resolve(articleId);
 };
 
-App.prototype.upvoteArticle = function(userId, articleId) {
-  return this.Article.voteFor(userId, articleId);
-};
-
 App.prototype.purgePendingArticles = function() {
   logger.log({ type: 'info', msg: 'app.purgePendingArticles' });
 
@@ -98,48 +92,6 @@ App.prototype.listArticles = function(userId, n, fresh) {
   return this.Article.list(userId, n, fresh);
 };
 
-App.prototype.startScraping = function() {
-  this.connections.queue.handle(SCRAPE_QUEUE, this.handleScrapeJob.bind(this));
-  this.connections.queue.handle(VOTE_QUEUE, this.handleVoteJob.bind(this));
-  return this;
-};
-
-App.prototype.handleScrapeJob = function(job, ack) {
-  logger.log({ type: 'info', msg: 'handling job', queue: SCRAPE_QUEUE, url: job.url });
-
-  this
-    .scrapeArticle(job.userId, job.id, job.url)
-    .then(onSuccess, onError);
-
-  function onSuccess() {
-    logger.log({ type: 'info', msg: 'job complete', status: 'success', url: job.url });
-    ack();
-  }
-
-  function onError() {
-    logger.log({ type: 'info', msg: 'job complete', status: 'failure', url: job.url });
-    ack();
-  }
-};
-
-App.prototype.handleVoteJob = function(job, ack) {
-  logger.log({ type: 'info', msg: 'handling job', queue: VOTE_QUEUE, articleId: job.articleId });
-
-  this
-    .upvoteArticle(job.userId, job.articleId)
-    .then(onSuccess, onError);
-
-  function onSuccess() {
-    logger.log({ type: 'info', msg: 'job complete', queue: VOTE_QUEUE, status: 'success' });
-    ack();
-  }
-
-  function onError(err) {
-    logger.log({ type: 'info', msg: 'job complete', queue: VOTE_QUEUE, status: 'failure', error: err });
-    ack();
-  }
-};
-
 App.prototype.stopScraping = function() {
   this.connections.queue.ignore(SCRAPE_QUEUE);
   this.connections.queue.ignore(VOTE_QUEUE);

lib/config.js

@@ -3,8 +3,8 @@
 module.exports = {
 
   // Services
-  mongo_url: process.env.MONGOLAB_URI || 'mongodb://localhost:27017/appDev',
-  rabbit_url: process.env.CLOUDAMQP_URL || 'amqp://localhost',
+  mongo_url: process.env.MONGOLAB_URI || 'mongodb://192.168.59.103:27017/appDev',
+  rabbit_url: process.env.CLOUDAMQP_URL || 'amqp://192.168.59.103',
   port: int(process.env.PORT) || 5000,
 
   // Security

lib/server.js

@@ -25,7 +25,6 @@ function start() {
   instance.on('lost', abort);
 
   function createServer() {
-    if (config.thrifty) instance.startScraping();
     var server = http.createServer(web(instance, config));
 
     process.on('SIGTERM', shutdown);

lib/work/errors.js

@@ -0,0 +1,32 @@
+function ArticleNotFound() {
+  Error.call(this);
+  Error.captureStackTrace(this, ArticleNotFound);
+  this.name = 'ArticleNotFound';
+  this.message = 'Article Not Found';
+}
+
+ArticleNotFound.prototype = Object.create(Error.prototype);
+
+function VoteNotAllowed() {
+  Error.call(this);
+  Error.captureStackTrace(this, VoteNotAllowed);
+  this.name = 'VoteNotAllowed';
+  this.message = 'Vote Not Allowed';
+}
+
+VoteNotAllowed.prototype = Object.create(Error.prototype);
+
+function ScrapeFailed() {
+  Error.call(this);
+  Error.captureStackTrace(this, ScrapeFailed);
+  this.name = 'ScrapeFailed';
+  this.message = 'Scrape Failed';
+}
+
+ScrapeFailed.prototype = Object.create(Error.prototype);
+
+module.exports = {
+  ArticleNotFound: ArticleNotFound,
+  VoteNotAllowed: VoteNotAllowed,
+  ScrapeFailed: ScrapeFailed
+};

lib/work/index.js

@@ -0,0 +1,111 @@
+var logger = require('logfmt');
+var Promise = require('promise');
+var EventEmitter = require('events').EventEmitter;
+
+var connections = require('../app/connections');
+var ArticleModel = require('../app/article-model');
+
+var SCRAPE_QUEUE = 'jobs.scrape';
+var VOTE_QUEUE = 'jobs.vote';
+
+function Work(config) {
+  logger.log({ type: 'info', msg: 'work/index.Work' });
+  EventEmitter.call(this);
+
+  this.config = config;
+  this.connections = connections(config.mongo_url, config.rabbit_url);
+  this.connections.once('ready', this.onConnected.bind(this));
+  this.connections.once('lost', this.onLost.bind(this));
+}
+
+module.exports = function createApp(config) {
+  return new Work(config);
+};
+
+Work.prototype = Object.create(EventEmitter.prototype);
+
+Work.prototype.onConnected = function() {
+  logger.log({ type: 'info', msg: 'work/index.onConnected' });
+  var queues = 0;
+  this.Article = ArticleModel(this.connections.db, this.config.mongo_cache);
+  this.connections.queue.create(SCRAPE_QUEUE, { prefetch: 5 }, onCreate.bind(this));
+  this.connections.queue.create(VOTE_QUEUE, { prefetch: 5 }, onCreate.bind(this));
+
+  function onCreate() {
+    if (++queues === 2) this.onReady();
+  }
+};
+
+Work.prototype.onReady = function() {
+  logger.log({ type: 'info', msg: 'app.ready' });
+  this.emit('ready');
+};
+
+Work.prototype.onLost = function() {
+  logger.log({ type: 'info', msg: 'app.lost' });
+  this.emit('lost');
+};
+
+Work.prototype.scrapeArticle = function(userId, id, url) {
+  logger.log({ type: 'info', msg: 'work/index.scrapeArticle' });
+  return this.Article.scrape(userId, id, url);
+};
+
+Work.prototype.upvoteArticle = function(userId, articleId) {
+  logger.log({ type: 'info', msg: 'work/index.upvoteArticle' });
+  return this.Article.voteFor(userId, articleId);
+};
+
+Work.prototype.startScraping = function() {
+  logger.log({ type: 'info', msg: 'work/index.startScraping' });
+  this.connections.queue.handle(SCRAPE_QUEUE, this.handleScrapeJob.bind(this));
+  this.connections.queue.handle(VOTE_QUEUE, this.handleVoteJob.bind(this));
+  return this;
+};
+
+// KEEP
+Work.prototype.handleScrapeJob = function(job, ack) {
+  logger.log({ type: 'info', msg: 'work/index.handleScrapeJob' });
+  logger.log({ type: 'info', msg: 'handling job', queue: SCRAPE_QUEUE, url: job.url });
+
+  this
+    .scrapeArticle(job.userId, job.id, job.url)
+    .then(onSuccess, onError);
+
+  function onSuccess() {
+    logger.log({ type: 'info', msg: 'job complete', status: 'success', url: job.url });
+    ack();
+  }
+
+  function onError() {
+    logger.log({ type: 'info', msg: 'job complete', status: 'failure', url: job.url });
+    ack();
+  }
+};
+
+// KEEP
+Work.prototype.handleVoteJob = function(job, ack) {
+  logger.log({ type: 'info', msg: 'handling job', queue: VOTE_QUEUE, articleId: job.articleId });
+
+  this
+    .upvoteArticle(job.userId, job.articleId)
+    .then(onSuccess, onError);
+
+  function onSuccess() {
+    logger.log({ type: 'info', msg: 'job complete', queue: VOTE_QUEUE, status: 'success' });
+    ack();
+  }
+
+  function onError(err) {
+    logger.log({ type: 'info', msg: 'job complete', queue: VOTE_QUEUE, status: 'failure', error: err });
+    ack();
+  }
+};
+
+// KEEP
+Work.prototype.stopScraping = function() {
+  this.connections.queue.ignore(SCRAPE_QUEUE);
+  this.connections.queue.ignore(VOTE_QUEUE);
+  return this;
+};
+

lib/worker.js

@@ -3,7 +3,7 @@ var logger = require('logfmt');
 var throng = require('throng');
 
 var config = require('./config');
-var app = require('./app');
+var app = require('./work');
 
 http.globalAgent.maxSockets = Infinity;
 throng(start, { workers: config.worker_concurrency });

package.json

@@ -4,6 +4,7 @@
   "description": "A minimum viable node app for production deployment",
   "main": "index.js",
   "scripts": {
+    "dev-start": "script/dev-start",
     "start": "nf start",
     "reset": "script/reset",
     "benchmark": "script/benchmark"
@@ -33,7 +34,9 @@
     "superagent": "^0.18.2",
     "throng": "^1.0.0"
   },
-  "devDependencies": {},
+  "devDependencies": {
+    "nodemon": "^1.3.7"
+  },
   "engines": {
     "node": "0.11.x"
   }

readme.md

@@ -39,7 +39,7 @@ cd node-articles-nlp
 
 heroku create
 
-heroku addons:add mongohq
+heroku addons:add mongolab
 heroku addons:add cloudamqp
 
 heroku config:set NODE_ENV=production

script/dev-start

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+nf start -j Procfile_dev