<commit_before>function publish(name, getFeed, collectReferences, getParams={}) {
	var publication;

	function streamSubscriptionHandle(data) {
		var refs = collectReferences(data.new);

		_(refs).each((ids, model) => {
			var collection = Mongo.Collection.get(model);

			if(! collection instanceof Mongo.Collection) {
		        throw new Meteor.Error('non-collection', `couldn\'t find collection with name ${ref}`);
			}

			_(ids).each((id) => publication.added(model, id, collection.findOne(id)));
		});

		_(data.new).each(activity => publication.added(name, activity.id, activity));

		_(data.deleted).each(activityId => publication.removed(name, activityId));

		publication.ready();
	}

	Meteor.publish(name, function(limit=20, userId=undefined /* subscribe params */) {
		publication = this;

		if (!this.userId) {
		  throw new Meteor.Error('not-authorized', 'You can only subscribe to feeds when authenticated');
		}

		_(getParams).defaults({
			limit: limit
		});

		var streamFeed = getFeed(userId || this.userId),
			feed = Stream.await(streamFeed.get(getParams)),
		    activities = feed.results;

		activities.forEach(activity => {
			publication.added(name, activity.id, activity);
		});

		var refs = collectReferences(activities);
		var cursors = [];

		_(refs).each((ids, model) => {
			var collection = Mongo.Collection.get(model);

			if(! collection instanceof Mongo.Collection) {
		        throw new Meteor.Error('non-collection', `couldn\'t find collection with name ${ref}`);
			}

			cursors.push(collection.find({ _id: { $in: ids } }));
		});

		var boundStreamSubscriptionHandle = Meteor.bindEnvironment(streamSubscriptionHandle);
		var streamSubscription = streamFeed.subscribe(boundStreamSubscriptionHandle);

		this.onStop(function() {
			streamSubscription.cancel();
		});

		return cursors;
	});
}

function getNewsFeed(feedGroup, userId) {
	return Stream.feedManager.getNewsFeeds(userId)[feedGroup];
}

function getUserFeed(userId) {
	return Stream.feedManager.getUserFeed(userId);
}

function getNotificationFeed(userId) {
	return Stream.feedManager.getNotificationFeed(userId);
}

function aggregatedCollectReferences(aggregated) {
	aggregated = _.chain(aggregated)
		.map(aggregation => aggregation['activities'])
		.flatten()
		.value();

	return Stream.backend.collectReferences(aggregated);
}

_(Stream._settings.newsFeeds).each((feedType, feedGroup) => {
	var collect = Stream.backend.collectReferences.bind(Stream.backend);

	if(feedType === 'aggregated') {
		collect = aggregatedCollectReferences;
	}

	publish(`Stream.feeds.${feedGroup}`, _.partial(getNewsFeed, feedGroup), collect);	
});

publish('Stream.feeds.notification', getNotificationFeed, aggregatedCollectReferences);
publish('Stream.feeds.user', getUserFeed, Stream.backend.collectReferences.bind(Stream.backend));<commit_msg>Set publication status to ready when no userId is supplied<commit_after>function publish(name, getFeed, collectReferences, getParams={}) {
	var publication;

	function streamSubscriptionHandle(data) {
		var refs = collectReferences(data.new);

		_(refs).each((ids, model) => {
			var collection = Mongo.Collection.get(model);

			if(! collection instanceof Mongo.Collection) {
		        throw new Meteor.Error('non-collection', `couldn\'t find collection with name ${ref}`);
			}

			_(ids).each((id) => publication.added(model, id, collection.findOne(id)));
		});

		_(data.new).each(activity => publication.added(name, activity.id, activity));

		_(data.deleted).each(activityId => publication.removed(name, activityId));

		publication.ready();
	}

	Meteor.publish(name, function(limit=20, userId=undefined /* subscribe params */) {
		userId = userId || this.userId;
		publication = this;

		if(! userId) {
			return publication.ready();
		}

		if (!this.userId) {
		  throw new Meteor.Error('not-authorized', 'You can only subscribe to feeds when authenticated');
		}

		_(getParams).defaults({
			limit: limit
		});

		var streamFeed = getFeed(userId),
			feed = Stream.await(streamFeed.get(getParams)),
		    activities = feed.results;

		activities.forEach(activity => {
			publication.added(name, activity.id, activity);
		});

		var refs = collectReferences(activities);
		var cursors = [];

		_(refs).each((ids, model) => {
			var collection = Mongo.Collection.get(model);

			if(! collection instanceof Mongo.Collection) {
		        throw new Meteor.Error('non-collection', `couldn\'t find collection with name ${ref}`);
			}

			cursors.push(collection.find({ _id: { $in: ids } }));
		});

		var boundStreamSubscriptionHandle = Meteor.bindEnvironment(streamSubscriptionHandle);
		var streamSubscription = streamFeed.subscribe(boundStreamSubscriptionHandle);

		this.onStop(function() {
			streamSubscription.cancel();
		});

		return cursors;
	});
}

function getNewsFeed(feedGroup, userId) {
	return Stream.feedManager.getNewsFeeds(userId)[feedGroup];
}

function getUserFeed(userId) {
	return Stream.feedManager.getUserFeed(userId);
}

function getNotificationFeed(userId) {
	return Stream.feedManager.getNotificationFeed(userId);
}

function aggregatedCollectReferences(aggregated) {
	aggregated = _.chain(aggregated)
		.map(aggregation => aggregation['activities'])
		.flatten()
		.value();

	return Stream.backend.collectReferences(aggregated);
}

_(Stream._settings.newsFeeds).each((feedType, feedGroup) => {
	var collect = Stream.backend.collectReferences.bind(Stream.backend);

	if(feedType === 'aggregated') {
		collect = aggregatedCollectReferences;
	}

	publish(`Stream.feeds.${feedGroup}`, _.partial(getNewsFeed, feedGroup), collect);	
});

publish('Stream.feeds.notification', getNotificationFeed, aggregatedCollectReferences);
publish('Stream.feeds.user', getUserFeed, Stream.backend.collectReferences.bind(Stream.backend));
{'fraction_non_alphanumeric': 0.14265951928065018, 'fraction_numerical': 0.0006916825177243646, 'mean_word_length': 5.127118644067797, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '881935', 'n_tokens_mistral': 1887, 'n_tokens_neox': 1694, 'n_words': 326}