old_contents: import axios from 'axios';
  import joi from 'joi';
  
  import db from '../utils/db';
  
  import RSS from '../models/rss';
  import Podcast from '../models/podcast';
  import logger from '../utils/logger';
  import { ProcessStreamQueue } from '../asyncTasks';
  import { timeIt } from '../utils/statsd';
  import { sendFeedToCollections } from '../utils/collections';
  import { setupAxiosRedirectInterceptor } from '../utils/axios';
  
  if (require.main === module) {
  	setupAxiosRedirectInterceptor(axios);
  
  	logger.info('Starting the Stream worker');
  	ProcessStreamQueue(1, streamProcessor);
  }
  
  export async function streamProcessor(job) {
  	logger.info(`Processing Stream feeds for feed ${job.data.rss}`);
  	// just intercept error handling before it goes to Bull
  	try {
  		await handleStream(job);
  	} catch (err) {
  		const tags = { queue: 'stream' };
  		const extra = { JobRSS: job.data.rss, JobArticles: job.data.articles };
  		logger.error('Stream feed job ecountered an error', { err, tags, extra });
  	}
  	logger.info(`Completed processing Stream feeds for feed ${job.data.rss}`);
  }
  
  const joiObjectId = joi.alternatives().try(
  	joi.string().length(12),
  	joi.string().length(24).regex(/^[0-9a-fA-F]{24}$/)
  );
  
  const schema = joi.object().keys({
  	rss: joiObjectId,
  	podcast: joiObjectId,
  }).xor('rss', 'podcast');
  
  export async function handleStream(job) {
  	const validation = joi.validate(job.data, schema);
  	if (!!validation.error) {
  		logger.warn(`Stream job validation failed: ${validation.error.message} for '${JSON.stringify(job.data)}'`);
  		return;
  	}
  
  	const type = 'rss' in job.data ? 'rss' : 'podcast';
  	const model = 'rss' in job.data ? RSS : Podcast;
  
  	await model.update({ _id: job.data[type] }, { "queueState.isSynchronizingWithStream": false });
  
  	const feed = await model.findById(job.data[type]);
  	await timeIt('winds.handle_stream.send_to_collections', () => sendFeedToCollections(type, feed));
  }
  
new_contents: import axios from 'axios';
  import joi from 'joi';
  
  import db from '../utils/db';
  
  import RSS from '../models/rss';
  import Podcast from '../models/podcast';
  import logger from '../utils/logger';
  import { ProcessStreamQueue } from '../asyncTasks';
  import { timeIt } from '../utils/statsd';
  import { sendFeedToCollections } from '../utils/collections';
  import { setupAxiosRedirectInterceptor } from '../utils/axios';
  
  if (require.main === module) {
  	setupAxiosRedirectInterceptor(axios);
  
  	logger.info('Starting the Stream worker');
  	ProcessStreamQueue(2, streamProcessor);
  }
  
  export async function streamProcessor(job) {
  	logger.info(`Processing Stream feeds for feed ${job.data.rss}`);
  	// just intercept error handling before it goes to Bull
  	try {
  		await handleStream(job);
  	} catch (err) {
  		const tags = { queue: 'stream' };
  		const extra = { JobRSS: job.data.rss, JobArticles: job.data.articles };
  		logger.error('Stream feed job ecountered an error', { err, tags, extra });
  	}
  	logger.info(`Completed processing Stream feeds for feed ${job.data.rss}`);
  }
  
  const joiObjectId = joi.alternatives().try(
  	joi.string().length(12),
  	joi.string().length(24).regex(/^[0-9a-fA-F]{24}$/)
  );
  
  const schema = joi.object().keys({
  	rss: joiObjectId,
  	podcast: joiObjectId,
  }).xor('rss', 'podcast');
  
  export async function handleStream(job) {
  	const validation = joi.validate(job.data, schema);
  	if (!!validation.error) {
  		logger.warn(`Stream job validation failed: ${validation.error.message} for '${JSON.stringify(job.data)}'`);
  		return;
  	}
  
  	const type = 'rss' in job.data ? 'rss' : 'podcast';
  	const model = 'rss' in job.data ? RSS : Podcast;
  
  	await model.update({ _id: job.data[type] }, { "queueState.isSynchronizingWithStream": false });
  
  	const feed = await model.findById(job.data[type]);
  	await timeIt('winds.handle_stream.send_to_collections', () => sendFeedToCollections(type, feed));
  }
  
subject: Increase stream worker concurrency