replace homemade threadpool framework with rxjava

commafeed-server/src/main/java/com/commafeed/backend/service/DatabaseStartupService.java

@@ -25,7 +25,7 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 @RequiredArgsConstructor(onConstructor = @__({ @Inject }))
 @Singleton
-public class StartupService implements Managed {
+public class DatabaseStartupService implements Managed {
 
 	private final SessionFactory sessionFactory;
 	private final UserDAO userDAO;

commafeed-server/src/main/java/com/commafeed/backend/service/FeedEntryService.java

@@ -7,18 +7,24 @@ import java.util.List;
 import javax.inject.Inject;
 import javax.inject.Singleton;
 
+import org.apache.commons.codec.digest.DigestUtils;
+
 import com.commafeed.backend.cache.CacheService;
 import com.commafeed.backend.dao.FeedEntryDAO;
 import com.commafeed.backend.dao.FeedEntryStatusDAO;
 import com.commafeed.backend.dao.FeedSubscriptionDAO;
 import com.commafeed.backend.feed.FeedEntryKeyword;
+import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedEntry;
+import com.commafeed.backend.model.FeedEntryContent;
 import com.commafeed.backend.model.FeedEntryStatus;
 import com.commafeed.backend.model.FeedSubscription;
 import com.commafeed.backend.model.User;
 
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
 
+@Slf4j
 @RequiredArgsConstructor(onConstructor = @__({ @Inject }))
 @Singleton
 public class FeedEntryService {
@@ -26,8 +32,45 @@ public class FeedEntryService {
 	private final FeedSubscriptionDAO feedSubscriptionDAO;
 	private final FeedEntryDAO feedEntryDAO;
 	private final FeedEntryStatusDAO feedEntryStatusDAO;
+	private final FeedEntryContentService feedEntryContentService;
+	private final FeedEntryFilteringService feedEntryFilteringService;
 	private final CacheService cache;
 
+	/**
+	 * this is NOT thread-safe
+	 */
+	public boolean addEntry(Feed feed, FeedEntry entry, List<FeedSubscription> subscriptions) {
+
+		Long existing = feedEntryDAO.findExisting(entry.getGuid(), feed);
+		if (existing != null) {
+			return false;
+		}
+
+		FeedEntryContent content = feedEntryContentService.findOrCreate(entry.getContent(), feed.getLink());
+		entry.setGuidHash(DigestUtils.sha1Hex(entry.getGuid()));
+		entry.setContent(content);
+		entry.setInserted(new Date());
+		entry.setFeed(feed);
+		feedEntryDAO.saveOrUpdate(entry);
+
+		// if filter does not match the entry, mark it as read
+		for (FeedSubscription sub : subscriptions) {
+			boolean matches = true;
+			try {
+				matches = feedEntryFilteringService.filterMatchesEntry(sub.getFilter(), entry);
+			} catch (FeedEntryFilteringService.FeedEntryFilterException e) {
+				log.error("could not evaluate filter {}", sub.getFilter(), e);
+			}
+			if (!matches) {
+				FeedEntryStatus status = new FeedEntryStatus(sub.getUser(), sub, entry);
+				status.setRead(true);
+				feedEntryStatusDAO.saveOrUpdate(status);
+			}
+		}
+
+		return true;
+	}
+
 	public void markEntry(User user, Long entryId, boolean read) {
 
 		FeedEntry entry = feedEntryDAO.findById(entryId);

commafeed-server/src/main/java/com/commafeed/backend/service/FeedRefreshEngine.java

@@ -0,0 +1,120 @@
+package com.commafeed.backend.service;
+
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+
+import org.apache.commons.lang3.time.DateUtils;
+import org.hibernate.SessionFactory;
+
+import com.codahale.metrics.Meter;
+import com.codahale.metrics.MetricRegistry;
+import com.commafeed.CommaFeedConfiguration;
+import com.commafeed.backend.dao.FeedDAO;
+import com.commafeed.backend.dao.UnitOfWork;
+import com.commafeed.backend.feed.FeedRefreshUpdater;
+import com.commafeed.backend.feed.FeedRefreshWorker;
+import com.commafeed.backend.model.Feed;
+
+import io.dropwizard.lifecycle.Managed;
+import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.disposables.Disposable;
+import io.reactivex.rxjava3.processors.PublishProcessor;
+import io.reactivex.rxjava3.schedulers.Schedulers;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Singleton
+public class FeedRefreshEngine implements Managed {
+
+	private final SessionFactory sessionFactory;
+	private final FeedDAO feedDAO;
+	private final FeedRefreshWorker worker;
+	private final FeedRefreshUpdater updater;
+	private final CommaFeedConfiguration config;
+	private final Meter refill;
+
+	private final PublishProcessor<Feed> priorityQueue;
+	private Disposable flow;
+
+	@Inject
+	public FeedRefreshEngine(SessionFactory sessionFactory, FeedDAO feedDAO, FeedRefreshWorker worker, FeedRefreshUpdater updater,
+			CommaFeedConfiguration config, MetricRegistry metrics) {
+		this.sessionFactory = sessionFactory;
+		this.feedDAO = feedDAO;
+		this.worker = worker;
+		this.updater = updater;
+		this.config = config;
+		this.refill = metrics.meter(MetricRegistry.name(getClass(), "refill"));
+		this.priorityQueue = PublishProcessor.create();
+	}
+
+	@Override
+	public void start() throws Exception {
+		Flowable<Feed> database = Flowable.fromCallable(() -> findNextUpdatableFeeds(getBatchSize(), getLastLoginThreshold()))
+				.onErrorResumeNext(e -> {
+					log.error("error while fetching next updatable feeds", e);
+					return Flowable.empty();
+				})
+				// repeat query 15s after the flowable has been emptied
+				// https://github.com/ReactiveX/RxJava/issues/448#issuecomment-233244964
+				.repeatWhen(o -> o.concatMap(v -> Flowable.timer(15, TimeUnit.SECONDS)))
+				.flatMap(Flowable::fromIterable);
+		Flowable<Feed> source = Flowable.merge(priorityQueue, database);
+
+		this.flow = source.subscribeOn(Schedulers.io())
+				// feed fetching
+				.parallel(config.getApplicationSettings().getBackgroundThreads())
+				.runOn(Schedulers.io())
+				.flatMap(f -> Flowable.fromCallable(() -> worker.update(f)).onErrorResumeNext(e -> {
+					log.error("error while fetching feed", e);
+					return Flowable.empty();
+				}))
+				.sequential()
+				// database updating
+				.parallel(config.getApplicationSettings().getDatabaseUpdateThreads())
+				.runOn(Schedulers.io())
+				.flatMap(fae -> Flowable.fromCallable(() -> updater.update(fae.getFeed(), fae.getEntries())).onErrorResumeNext(e -> {
+					log.error("error while updating database", e);
+					return Flowable.empty();
+				}))
+				.sequential()
+				// end flow
+				.subscribe();
+	}
+
+	public void refreshImmediately(Feed feed) {
+		priorityQueue.onNext(feed);
+	}
+
+	private List<Feed> findNextUpdatableFeeds(int max, Date lastLoginThreshold) {
+		refill.mark();
+
+		return UnitOfWork.call(sessionFactory, () -> {
+			List<Feed> list = feedDAO.findNextUpdatable(max, lastLoginThreshold);
+
+			// set the disabledDate as we use it in feedDAO.findNextUpdatable() to decide what to refresh next
+			Date nextRefreshDate = DateUtils.addMinutes(new Date(), config.getApplicationSettings().getRefreshIntervalMinutes());
+			list.forEach(f -> f.setDisabledUntil(nextRefreshDate));
+			feedDAO.saveOrUpdate(list);
+
+			return list;
+		});
+	}
+
+	private int getBatchSize() {
+		return Math.min(Flowable.bufferSize(), 3 * config.getApplicationSettings().getBackgroundThreads());
+	}
+
+	private Date getLastLoginThreshold() {
+		return Boolean.TRUE.equals(config.getApplicationSettings().getHeavyLoad()) ? DateUtils.addDays(new Date(), -30) : null;
+	}
+
+	@Override
+	public void stop() throws Exception {
+		flow.dispose();
+	}
+}

commafeed-server/src/main/java/com/commafeed/backend/service/FeedService.java

@@ -50,6 +50,14 @@ public class FeedService {
 		return feed;
 	}
 
+	public void save(Feed feed) {
+		String normalized = FeedUtils.normalizeURL(feed.getUrl());
+		feed.setNormalizedUrl(normalized);
+		feed.setNormalizedUrlHash(DigestUtils.sha1Hex(normalized));
+		feed.setLastUpdated(new Date());
+		feedDAO.saveOrUpdate(feed);
+	}
+
 	public Favicon fetchFavicon(Feed feed) {
 
 		Favicon icon = null;

commafeed-server/src/main/java/com/commafeed/backend/service/FeedSubscriptionService.java

@@ -14,7 +14,6 @@ import com.commafeed.backend.cache.CacheService;
 import com.commafeed.backend.dao.FeedDAO;
 import com.commafeed.backend.dao.FeedEntryStatusDAO;
 import com.commafeed.backend.dao.FeedSubscriptionDAO;
-import com.commafeed.backend.feed.FeedQueues;
 import com.commafeed.backend.feed.FeedUtils;
 import com.commafeed.backend.model.Feed;
 import com.commafeed.backend.model.FeedCategory;
@@ -35,7 +34,7 @@ public class FeedSubscriptionService {
 	private final FeedEntryStatusDAO feedEntryStatusDAO;
 	private final FeedSubscriptionDAO feedSubscriptionDAO;
 	private final FeedService feedService;
-	private final FeedQueues queues;
+	private final FeedRefreshEngine feedRefreshEngine;
 	private final CacheService cache;
 	private final CommaFeedConfiguration config;
 
@@ -76,7 +75,7 @@ public class FeedSubscriptionService {
 		sub.setTitle(FeedUtils.truncate(title, 128));
 		feedSubscriptionDAO.saveOrUpdate(sub);
 
-		queues.add(feed, true);
+		feedRefreshEngine.refreshImmediately(feed);
 		cache.invalidateUserRootCategory(user);
 		return sub.getId();
 	}
@@ -96,7 +95,7 @@ public class FeedSubscriptionService {
 		List<FeedSubscription> subs = feedSubscriptionDAO.findAll(user);
 		for (FeedSubscription sub : subs) {
 			Feed feed = sub.getFeed();
-			queues.add(feed, true);
+			feedRefreshEngine.refreshImmediately(feed);
 		}
 	}
 

commafeed-server/src/main/java/com/commafeed/backend/service/FeedUpdateService.java

@@ -1,67 +0,0 @@
-package com.commafeed.backend.service;
-
-import java.util.Date;
-import java.util.List;
-
-import javax.inject.Inject;
-import javax.inject.Singleton;
-
-import org.apache.commons.codec.digest.DigestUtils;
-
-import com.commafeed.backend.dao.FeedEntryDAO;
-import com.commafeed.backend.dao.FeedEntryStatusDAO;
-import com.commafeed.backend.model.Feed;
-import com.commafeed.backend.model.FeedEntry;
-import com.commafeed.backend.model.FeedEntryContent;
-import com.commafeed.backend.model.FeedEntryStatus;
-import com.commafeed.backend.model.FeedSubscription;
-import com.commafeed.backend.service.FeedEntryFilteringService.FeedEntryFilterException;
-
-import lombok.RequiredArgsConstructor;
-import lombok.extern.slf4j.Slf4j;
-
-@Slf4j
-@RequiredArgsConstructor(onConstructor = @__({ @Inject }))
-@Singleton
-public class FeedUpdateService {
-
-	private final FeedEntryDAO feedEntryDAO;
-	private final FeedEntryStatusDAO feedEntryStatusDAO;
-	private final FeedEntryContentService feedEntryContentService;
-	private final FeedEntryFilteringService feedEntryFilteringService;
-
-	/**
-	 * this is NOT thread-safe
-	 */
-	public boolean addEntry(Feed feed, FeedEntry entry, List<FeedSubscription> subscriptions) {
-
-		Long existing = feedEntryDAO.findExisting(entry.getGuid(), feed);
-		if (existing != null) {
-			return false;
-		}
-
-		FeedEntryContent content = feedEntryContentService.findOrCreate(entry.getContent(), feed.getLink());
-		entry.setGuidHash(DigestUtils.sha1Hex(entry.getGuid()));
-		entry.setContent(content);
-		entry.setInserted(new Date());
-		entry.setFeed(feed);
-		feedEntryDAO.saveOrUpdate(entry);
-
-		// if filter does not match the entry, mark it as read
-		for (FeedSubscription sub : subscriptions) {
-			boolean matches = true;
-			try {
-				matches = feedEntryFilteringService.filterMatchesEntry(sub.getFilter(), entry);
-			} catch (FeedEntryFilterException e) {
-				log.error("could not evaluate filter {}", sub.getFilter(), e);
-			}
-			if (!matches) {
-				FeedEntryStatus status = new FeedEntryStatus(sub.getUser(), sub, entry);
-				status.setRead(true);
-				feedEntryStatusDAO.saveOrUpdate(status);
-			}
-		}
-
-		return true;
-	}
-}

commafeed-server/src/main/java/com/commafeed/backend/service/PubSubService.java

@@ -17,10 +17,11 @@ import org.apache.http.client.methods.HttpPost;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.util.EntityUtils;
+import org.hibernate.SessionFactory;
 
 import com.commafeed.CommaFeedConfiguration;
 import com.commafeed.backend.HttpGetter;
-import com.commafeed.backend.feed.FeedQueues;
+import com.commafeed.backend.dao.UnitOfWork;
 import com.commafeed.backend.feed.FeedUtils;
 import com.commafeed.backend.model.Feed;
 import com.commafeed.frontend.resource.PubSubHubbubCallbackREST;
@@ -38,7 +39,8 @@ import lombok.extern.slf4j.Slf4j;
 public class PubSubService {
 
 	private final CommaFeedConfiguration config;
-	private final FeedQueues queues;
+	private final FeedService feedService;
+	private final SessionFactory sessionFactory;
 
 	public void subscribe(Feed feed) {
 		String hub = feed.getPushHub();
@@ -73,7 +75,7 @@ public class PubSubService {
 				if (code == 400 && StringUtils.contains(message, pushpressError)) {
 					String[] tokens = message.split(" ");
 					feed.setPushTopic(tokens[tokens.length - 1]);
-					queues.giveBack(feed);
+					UnitOfWork.run(sessionFactory, () -> feedService.save(feed));
 					log.debug("handled pushpress subfeed {} : {}", topic, feed.getPushTopic());
 				} else {
 					throw new Exception(