Download worker

download.js

/** REQUIREMENTS **/
var Kue			= require('kue')
,	Http		= require('http-get')
,	Messenger  	= require('../lib/messenger')
,	parse		= require('../lib/parser')
,	models		= require('../models')
;

/** APP VARS **/
var	jobs 	= Kue.createQueue()
,	Article = models.Article
,	messenger = new Messenger()
;

var download = function(uri, callback) {
	// callback = function(err, buffer);
	Http.get({ url: uri }, function(err, result) {
		if(err) {
			callback(err, null);
		} else if(result.code !== 200) {
			callback("HTTP/1.1 " + result.code, null);
		} else {
			callback(null, result.buffer);
		}
	});
};

var saveArticle = function(a, callback) {
	var article = new Article({
		date: new Date(a.pubDate),
		uid: a.uid,
		title: a.title,
		url: a.link,
		description: a.description,
		html: '',
		feed: a.feed.uid,
		category: a.category.uid,
		language: a.language.uid
	});
	
	// Save the article to DB.
	article.save(function(err) {
		if(err) callback(err);
		else callback(null);
	});
};

var feedExists = function(uri) {
	models.Source.find({ url: uri }, function(err, docs) {
		if(err) {
			return false;
		} else {
			if(typeof docs !== 'object') return false;
			if(docs.length < 1) return false;

			return true;
		}
	});
};

jobs.process('download', 12, function(job, done) {
	var data = job.data;

	if(feedExists(data.url)) {
		console.log(data.url + " exists.");
		done();
	} else {
		download(data.url, function(err, xml) {
			if(err) { 
				done(err);
			} else {
				parse(data, xml, function(err, articles) {
					if(err) {
						done(err);
					} else if(!articles || articles.length < 1) {
						done("Articles is empty.");
					} else {
						var cursor = 0;
						var total = articles.length;

						var save = function() {
							saveArticle(articles[cursor], function(err) {
								if(err) {
									done(err);
								} else {
									messenger.publish({
										sender: 'worker:download', 
										event: 'article downloaded', 
										uid: articles[cursor].uid, 
										url: articles[cursor].link 
									});

									cursor++;

									if(cursor >= total) {
										done();
									} else {
										save(articles[cursor]);
									}
								}
							});
						};

						// Save the article.
						save();
					}
				});
			}
		});
	}
});