How to rule a data feed with node.js

http-csv-objectmapper-json-solr.js

console.time('Zuneta-stream');
/**
 * Modules dependencies
 */

var csv = require('csv-stream'),
	request = require('request'),
	objectMapper = require('object-mapper-stream'),
	solr = require('solr-client'),
	crypto = require('crypto');

var client = solr.createClient();

// Extend methods available through `format` for objectMapper
objectMapper.format.createMD5Hash = function(s){
	return crypto.createHash('md5').update(s,'utf-8').digest('hex');
}

// Options for the object mapper
var optionsMapper = {
	map : {
		id : { to : 'id' , format : 'createMD5Hash'},
		title : { to : 'title'},
		description : 'description',
		attribute : 'category',
		image_link : 'image1',
		brand : 'brand_t',
		price : 'price',
		link : 'url_display',
		default_affiliate : { to : 'affiliate_name_t' , bydefault : 'Zuneta'},
		default_merchant : { to : 'merchant_name_t', bydefault : 'Zuneta'},
		default_merchant_id : { to : 'merchant_id_t' , copy : 'merchant_name,affiliate', format : 'createMD5Hash'}
	}
};

console.time('delay-http-get');
var first = true;
request('http://www.zuneta.com/feeds/productBrands.txt')
	.on('error',function(err){
		console.error(err);
	})
	.on('data',function(){
		if(first) console.timeEnd('delay-http-get');
		first = false;
	})
	.pipe(csv.createStream({ delimiter : '\t'}))
	.on('error',function(err){
		console.log(err);
	})
	.pipe(objectMapper.createStream(optionsMapper))
	.on('error',function(err){
		console.error(err);
	})
	.pipe(client.createAddStream())
	.on('error',function(err){
		console.error(err);
	})
	.on('data',function(buffer,encoding){
		console.log(buffer.toString(encoding))
	})
	.on('end',function(){
		console.log('end!');
		console.timeEnd('Zuneta-stream');
	})