<commit_before>var hl = require('highland');
var _ = require('lodash');
var async = require('async');
var asfy = require('asfy');
var mongoose = require('mongoose');
var helpers = require('./helpers');
var config = require('./config');
var setup = require('./setup');
var models = require('./models')(mongoose);

var options = {
  timeOut: 10000
};

var jsonMapper = require('json-mapper')(options);
var feedMapper = require('feed-mapper')(options);
var siteParser = require('site-parser')(options);

setup.connectToDatabase(mongoose, config.get('database.mongo.url'));

var query = {
  active: true
};

var queryFunction = models.Source.find.bind(models.Source, query);
var isEqual = hl.ncurry(2, _.isEqual);

var sourceStream = hl(helpers.sourceWrapper(queryFunction))
  .ratelimit(1, 1000)
  .take(10)
  .errors(helpers.handleErrors)
  .compact()
  .flatten()

var jsonStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('json'),
      hl.get('type')
    )
  )
  .flatMap(
    hl.wrapCallback(
      jsonMapper.parse.bind(jsonMapper)
    )
  )

var rssStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('rss'),
      hl.get('type')
    )
  )
  .flatMap(
    hl.wrapCallback(
      feedMapper.parse.bind(feedMapper)
    )
  )

var siteStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('site'),
      hl.get('type')
    )
  )
  .flatMap(
    hl.wrapCallback(
      siteParser.parse.bind(siteParser)
    )
  )

var articleStream = hl([jsonStream, rssStream, siteStream])
  .merge()
  .flatten()
  .errors(helpers.handleErrors)

var newArticleStream = articleStream
  .fork()
  .flatFilter(
    hl.wrapCallback(
      async.compose(
        asfy(helpers.isTruthy),
        models.Entry.count,
        asfy(hl.get('guid'))
      )
    )
  )<commit_msg>do mapping from template to entries in parallel (map().parallel(num) vs. flatMap())<commit_after>var hl = require('highland');
var _ = require('lodash');
var async = require('async');
var asfy = require('asfy');
var mongoose = require('mongoose');
var helpers = require('./helpers');
var config = require('./config');
var setup = require('./setup');
var models = require('./models')(mongoose);

var options = {
  timeOut: 10000
};

var jsonMapper = require('json-mapper')(options);
var feedMapper = require('feed-mapper')(options);
var siteParser = require('site-parser')(options);

setup.connectToDatabase(mongoose, config.get('database.mongo.url'));

var query = {
  active: true
};

var queryFunction = models.Source.find.bind(models.Source, query);
var isEqual = hl.ncurry(2, _.isEqual);

var sourceStream = hl(helpers.sourceWrapper(queryFunction))
  .ratelimit(1, 2000)
  .take(10)
  .errors(helpers.handleErrors)
  .compact()
  .flatten()

var jsonStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('json'),
      hl.get('type')
    )
  )
  .map(
    hl.wrapCallback(
      jsonMapper.parse.bind(jsonMapper)
    )
  ).parallel(5)

var rssStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('rss'),
      hl.get('type')
    )
  )
  .map(
    hl.wrapCallback(
      feedMapper.parse.bind(feedMapper)
    )
  ).parallel(5)

var siteStream = sourceStream
  .fork()
  .filter(
    hl.compose(
      isEqual('site'),
      hl.get('type')
    )
  )
  .map(
    hl.wrapCallback(
      siteParser.parse.bind(siteParser)
    )
  ).parallel(5)

var articleStream = hl([jsonStream, rssStream, siteStream])
  .merge()
  .flatten()
  .errors(helpers.handleErrors)

var newArticleStream = articleStream
  .fork()
  .flatFilter(
    hl.wrapCallback(
      async.compose(
        asfy(helpers.isTruthy),
        models.Entry.count,
        asfy(hl.get('guid'))
      )
    )
  )
{'fraction_non_alphanumeric': 0.16022544283413848, 'fraction_numerical': 0.0077831454643048845, 'mean_word_length': 3.244874715261959, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2568744', 'n_tokens_mistral': 1333, 'n_tokens_neox': 1295, 'n_words': 239}