<commit_before>var DAY = 1000 * 60 * 60 * 24;
var daysStoreArticles = 3.0;
var updateInterval = 1000 * 60 * 15;
var intervalProcesses = {};
var articlePubLimit = 150;
var maxArticlesFromSource = 25;
var feedSubscriber;
var keepLimitDate = function(){
  return new Date( new Date().getTime() - ( DAY * daysStoreArticles ));
};

Articles._ensureIndex( { feed_id: 1, date: -1} );

Accounts.config({sendVerificationEmail: true});

Facts.setUserIdFilter(function ( userId ) {
  var user = Meteor.users.findOne(userId);
  return user && user.admin;
});

/************* null publication should be sent without fastRender.onAllRoutes()
FastRender.onAllRoutes( function( ) {
  var self = this,
      feed_ids,
      visibleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, clicks: 1, readCount: 1};

  if ( self.userId )
    feed_ids = Meteor.users.findOne( self.userId, {fields: {feedList: 1}} ).feedList;
  else
    feed_ids = Feeds.find( {subscribers: null}, {fields: {_id: 1}}).map ( function( feed ) { return feed._id;});

  //self.subscribe( 'feeds' );
  //self.subscribe( "articles", feed_ids, 20);
});
****************/
function stoppablePublisher( sub ){
  var self = this,
      handle,
      name;

  function subHasId( id ){
    return sub._documents && sub._documents[ name ] && sub._documents[ name ][ id ];
  };

  self.ids = () => {
    return _( sub._documents && sub._documents[ name ] || {} ).keys();
  }

  function observeAndPublish( cursor ){
    var oldIds = self.ids(),
        newIds;

    handle = cursor.observeChanges({
      added( id, doc ){
        if ( subHasId( id ) ){
          oldIds.splice( oldIds.indexOf( id ), 1);
        } else{
          sub.added( name, id, doc );
        }
      },
      removed ( id ){
        sub.removed( name, id );
      },
      changed ( id, doc ){
        sub.changed( name, id, doc );
      }
    });

    if ( sub._documents && oldIds.length)
      oldIds.forEach( ( id ) => sub.removed ( name, id) );
  }

  self.start = ( cursor ) =>{
    if ( handle ) handle.stop();
    if ( cursor._cursorDescription.collectionName !== name ){
      if ( ! name )
        name = cursor._cursorDescription.collectionName;
      else
        throw new Error( 'stoppablePublisher can not handle cursors from different collections. ',
         name, ' to ', cursor._cursorDescription.collectionName);
    }
    observeAndPublish( cursor );
  };

  self.stop = () => {
    handle && handle.stop();
  }
}


//prepare userdata and feedlist for all clients ASAP
Meteor.publish( null, function() {
  var self = this;
  var userId = self.userId || 'nullUser';
  var feedOptions = {fields: {_id: 1, title: 1, url: 1, last_date:1}};
  var articleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, feed_id: 1};
  var articleOptions = {fields: articleFields, limit: 100, sort: {date: -1, _id: 1}};
  var feedPublisher, articlePublisher, userObserver, nullUserObserver, nullUserCursor;

  feedPublisher = new stoppablePublisher( self );
  articlePublisher = new stoppablePublisher( self );

  function startFeedsAndArticles( id, doc){
    var feedCursor = Feeds.find( {_id:{ $in: doc.feedList}}, feedOptions);
    feedPublisher.start( feedCursor );
    var articleCursor = Articles.find( {feed_id: {$in: doc.feedList}, date: {$gt: keepLimitDate()}}, articleOptions);
    articlePublisher.start( articleCursor );
  }

  userObserver = Meteor.users.find( userId, {fields: {feedList: 1}} ).observeChanges({
    added: startFeedsAndArticles,
    changed: startFeedsAndArticles
  });


  self.onStop( () => {
    userObserver.stop();
    feedPublisher.stop();
    delete self.feedPublisher;
    articlePublisher.stop();
    delete self.articlePublisher;
  });

  return Meteor.users.find( userId, {fields: {admin: 1}});
});

Meteor.publish( 'articles', function( feed_ids, limit ){
  var self = this;
  check( feed_ids, [String] );
  check( limit, Number );
  var startDate = keepLimitDate();
  var visibleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, feed_id: 1};
  return Articles.find( {feed_id: {$in: feed_ids}, date: {$gt: startDate}}, {fields: visibleFields, limit: limit, sort: {date: -1, _id: 1}});
});

Meteor.startup( () => {

  Meteor.call('findArticles', {} );
  Meteor.call('removeOldArticles');


  intervalProcesses["removeOldArticles"] = Meteor.setInterval(
    () => Meteor.call('removeOldArticles'),
    DAY
  );


  intervalProcesses[ "findArticles"] = Meteor.setInterval(
    () => Meteor.call('findArticles', { hub: null} ),
    updateInterval
  );


  var options = {
    callbackPath: "hubbub",  //leave slash off since this will be argument to eteor.AbsoluteUrl()
    secret: Random.id()
  };

//if no ROOT_URL was set assume we are on my server
  if ( Meteor.absoluteUrl() === "http://localhost:3000/"){
    options.callbackUrl = "http://localhost:3000/" + options.callbackPath;
  }

  var openSubscriptions = {};
  feedSubscriber = new FeedSubscriber ( options );

  feedSubscriber.on(
    'feed',
    Meteor.bindEnvironment( FeedParser.readAndInsertArticles, ( error ) => console.log( error ) )
  );

  var stopAllSubscriptions = () => {
    var waitingOn = openSubscriptions.map( sub => {
      var fut = new Future();
      var cb = ( err, res ) => {
        delete openSubscriptions[ sub.url ];
        fut.return( res );
      };
      sub => feedSubscriber.unsubscribe( sub.url, sub.hub , feedSubscriber.callbackUrl, cb );
      return fut;
    });<commit_msg>no errors but no pubsub articles<commit_after>var Future = Npm.require( 'fibers/future');
var DAY = 1000 * 60 * 60 * 24;
var daysStoreArticles = 3.0;
var updateInterval = 1000 * 60 * 15;
var intervalProcesses = {};
var articlePubLimit = 150;
var maxArticlesFromSource = 25;
var feedSubscriber;
var keepLimitDate = function(){
  return new Date( new Date().getTime() - ( DAY * daysStoreArticles ));
};

Articles._ensureIndex( { feed_id: 1, date: -1} );

Accounts.config({sendVerificationEmail: true});

Facts.setUserIdFilter(function ( userId ) {
  var user = Meteor.users.findOne(userId);
  return user && user.admin;
});

/************* null publication should be sent without fastRender.onAllRoutes()
FastRender.onAllRoutes( function( ) {
  var self = this,
      feed_ids,
      visibleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, clicks: 1, readCount: 1};

  if ( self.userId )
    feed_ids = Meteor.users.findOne( self.userId, {fields: {feedList: 1}} ).feedList;
  else
    feed_ids = Feeds.find( {subscribers: null}, {fields: {_id: 1}}).map ( function( feed ) { return feed._id;});

  //self.subscribe( 'feeds' );
  //self.subscribe( "articles", feed_ids, 20);
});
****************/
function stoppablePublisher( sub ){
  var self = this,
      handle,
      name;

  function subHasId( id ){
    return sub._documents && sub._documents[ name ] && sub._documents[ name ][ id ];
  };

  self.ids = () => {
    return _( sub._documents && sub._documents[ name ] || {} ).keys();
  }

  function observeAndPublish( cursor ){
    var oldIds = self.ids(),
        newIds;

    handle = cursor.observeChanges({
      added( id, doc ){
        if ( subHasId( id ) ){
          oldIds.splice( oldIds.indexOf( id ), 1);
        } else{
          sub.added( name, id, doc );
        }
      },
      removed ( id ){
        sub.removed( name, id );
      },
      changed ( id, doc ){
        sub.changed( name, id, doc );
      }
    });

    if ( sub._documents && oldIds.length)
      oldIds.forEach( ( id ) => sub.removed ( name, id) );
  }

  self.start = ( cursor ) =>{
    if ( handle ) handle.stop();
    if ( cursor._cursorDescription.collectionName !== name ){
      if ( ! name )
        name = cursor._cursorDescription.collectionName;
      else
        throw new Error( 'stoppablePublisher can not handle cursors from different collections. ',
         name, ' to ', cursor._cursorDescription.collectionName);
    }
    observeAndPublish( cursor );
  };

  self.stop = () => {
    handle && handle.stop();
  }
}


//prepare userdata and feedlist for all clients ASAP
Meteor.publish( null, function() {
  var self = this;
  var userId = self.userId || 'nullUser';
  var feedOptions = {fields: {_id: 1, title: 1, url: 1, last_date:1}};
  var articleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, feed_id: 1};
  var articleOptions = {fields: articleFields, limit: 100, sort: {date: -1, _id: 1}};
  var feedPublisher, articlePublisher, userObserver, nullUserObserver, nullUserCursor;

  feedPublisher = new stoppablePublisher( self );
  articlePublisher = new stoppablePublisher( self );

  function startFeedsAndArticles( id, doc){
    var feedCursor = Feeds.find( {_id:{ $in: doc.feedList}}, feedOptions);
    feedPublisher.start( feedCursor );
    var articleCursor = Articles.find( {feed_id: {$in: doc.feedList}, date: {$gt: keepLimitDate()}}, articleOptions);
    articlePublisher.start( articleCursor );
  }

  userObserver = Meteor.users.find( userId, {fields: {feedList: 1}} ).observeChanges({
    added: startFeedsAndArticles,
    changed: startFeedsAndArticles
  });


  self.onStop( () => {
    userObserver.stop();
    feedPublisher.stop();
    delete self.feedPublisher;
    articlePublisher.stop();
    delete self.articlePublisher;
  });

  return Meteor.users.find( userId, {fields: {admin: 1}});
});

Meteor.publish( 'articles', function( feed_ids, limit ){
  var self = this;
  check( feed_ids, [String] );
  check( limit, Number );
  var startDate = keepLimitDate();
  var visibleFields = {_id: 1, title: 1, source: 1, date: 1, summary: 1, link: 1, feed_id: 1};
  return Articles.find( {feed_id: {$in: feed_ids}, date: {$gt: startDate}}, {fields: visibleFields, limit: limit, sort: {date: -1, _id: 1}});
});

Meteor.startup( () => {

  Meteor.call('findArticles', {} );
  Meteor.call('removeOldArticles');


  intervalProcesses["removeOldArticles"] = Meteor.setInterval(
    () => Meteor.call('removeOldArticles'),
    DAY
  );


  intervalProcesses[ "findArticles"] = Meteor.setInterval(
    () => Meteor.call('findArticles', { hub: null} ),
    updateInterval
  );


  var options = {
    callbackUrl: Meteor.absoluteUrl("hubbub"),  //leave slash off since this will be argument to eteor.AbsoluteUrl()
    secret: Random.id()
  };

  var openSubscriptions = {};
  feedSubscriber = new FeedSubscriber ( options );

  feedSubscriber.on("error", function(error){
    console.log("Error");
    console.log(error);
  });

  feedSubscriber.on(
    'feed', data => console.log( data ));
    //Meteor.bindEnvironment( FeedParser.readAndInsertArticles, ( error ) => console.log( error ) )
  //);

  var stopAllSubscriptions = () => {
    var waitingOn = _( openSubscriptions ).map( sub => {
      var fut = new Future();
      var cb = ( err, res ) => {
        delete openSubscriptions[ sub.url ];
        console.log( 'unsubscribed: ', res);
        fut.return( res );
      };
      feedSubscriber.unsubscribe( sub.url, sub.hub , feedSubscriber.callbackUrl, cb );
      return fut;
    });
{'fraction_non_alphanumeric': 0.15547703180212014, 'fraction_numerical': 0.012865814985956328, 'mean_word_length': 3.294941634241245, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 29, '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': '4300965', 'n_tokens_mistral': 3782, 'n_tokens_neox': 3500, 'n_words': 926}