Merge branch 'master' of github.com:GetStream/Winds

Author: astrotars

api/src/workers/podcast.js

@@ -182,7 +182,7 @@ export async function handlePodcast(job) {
 		tasks.push(OgQueueAdd({ type: 'episode', podcast: podcastID, urls: updatedEpisodes.map(e => e.link) }, queueOpts));
 	}
 	const allowedLanguage = [null, undefined, '', 'eng'].includes(podcast.language);
-	if (await tryCreateQueueFlag('stream', 'podcast', podcastID, streamTTL) && allowedLanguage) {
+	if (allowedLanguage) {
 		tasks.push(StreamQueueAdd({ podcast: podcastID, contentIds: updatedEpisodes.map(e => e._id) }, queueOpts));
 	}
 	statsd.increment('winds.handle_podcast.result.updates');

api/src/workers/rss.js

@@ -203,7 +203,7 @@ export async function handleRSS(job) {
 		tasks.push(OgQueueAdd({ type: 'article', rss: rssID, urls: updatedArticles.map(a => a.url) }, queueOpts));
 	}
 	const allowedLanguage = [null, undefined, '', 'eng'].includes(rss.language);
-	if (await tryCreateQueueFlag('stream', 'rss', rssID, streamTTL) && allowedLanguage) {
+	if (allowedLanguage) {
 		tasks.push(StreamQueueAdd({ rss: rssID, contentIds: updatedArticles.map(a => a._id)  }, queueOpts));
 	}
 	statsd.increment('winds.handle_rss.result.updates');

api/src/workers/stream.js

@@ -11,7 +11,6 @@ import logger from '../utils/logger';
 import { ProcessStreamQueue, ShutDownStreamQueue } from '../asyncTasks';
 import { timeIt } from '../utils/statsd';
 import { sendFeedToCollections } from '../utils/collections';
-import { removeQueueFlag } from '../utils/queue';
 import { startSampling } from '../utils/watchdog';
 
 if (require.main === module) {
@@ -30,6 +29,8 @@ export async function streamProcessor(job) {
 		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 });
+		//XXX: we have to retry to not lose updates
+		await job.retry();
 	}
 	logger.info(`Completed processing Stream feeds for feed ${job.data.rss}`);
 }
@@ -55,8 +56,6 @@ export async function handleStream(job) {
 	const [type, model, contentModel] =
 		'rss' in job.data ? ['rss', RSS, Article] : ['podcast', Podcast, Episode];
 
-	await removeQueueFlag('stream', type, job.data[type]);
-
 	const feed = await model.findById(job.data[type]);
 	const content = await contentModel.find({ _id: { $in: job.data.contentIds } })
 		.sort({ publicationDate: -1 })

app/package.json

@@ -1,6 +1,6 @@
 {
   "name": "Winds",
-  "version": "2.1.171",
+  "version": "2.1.172",
   "description": "Winds is a beatiful open source RSS Reader and Podcast app. Wind 2.0 was created using React/Redux/Node.\nFor more information, visit https://getstream.io/winds/.\nTo contribute or run your own version head over to Github: https://github.com/getstream/winds",
   "private": true,
   "author": "Winds Team <[email protected]>",

app/src/components/RSSArticleList.js

@@ -87,10 +87,12 @@ class RSSArticleList extends React.Component {
 			);
 
 			this.unsubscribeFromStreamFeed();
-			this.subscribeToStreamFeed(
-				nextProps.rssFeed._id,
-				nextProps.rssFeed.streamToken,
-			);
+			if (nextProps.rssFeed) {
+				this.subscribeToStreamFeed(
+					nextProps.rssFeed._id,
+					nextProps.rssFeed.streamToken,
+				);
+			}
 		}
 
 		if (!this.props.rssFeed && nextProps.rssFeed) {