Archives/Athou_commafeed
1
0
Fork 1
mirror of https://github.com/Athou/commafeed.git synced 2026-03-21 21:37:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

wip: allow parallel update of feeds

Browse Source
This commit is contained in:
Athou
2013-05-22 00:07:13 +02:00
parent f75baf5cff
commit 15b7d685fe
7 changed files with 177 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/java/com/commafeed/backend/feeds/FeedRefreshTaskGiver.java
View File

@@ -6,10 +6,8 @@ import java.util.List;
import java.util.Queue;
import javax.annotation.PostConstruct;
import javax.ejb.Lock;
import javax.ejb.LockType;
import javax.ejb.Singleton;
import javax.inject.Inject;
import javax.inject.Singleton;
import org.apache.commons.lang3.time.DateUtils;
@@ -34,7 +32,8 @@ public class FeedRefreshTaskGiver {
private int backgroundThreads;
private Queue<Feed> addQueue = Queues.newConcurrentLinkedQueue();
private Queue<Feed> queue = Queues.newConcurrentLinkedQueue();
private Queue<Feed> takeQueue = Queues.newConcurrentLinkedQueue();
private Queue<Feed> giveBackQueue = Queues.newConcurrentLinkedQueue();
@PostConstruct
public void init() {
@@ -42,7 +41,6 @@ public class FeedRefreshTaskGiver {
.getBackgroundThreads();
}
@Lock(LockType.WRITE)
public void add(Feed feed) {
Date now = Calendar.getInstance().getTime();
boolean heavyLoad = applicationSettingsService.get().isHeavyLoad();
@@ -52,31 +50,34 @@ public class FeedRefreshTaskGiver {
feed.setEtagHeader(null);
feed.setLastModifiedHeader(null);
}
addQueue.add(feed);
}
@Lock(LockType.WRITE)
public Feed take() {
Feed feed = queue.poll();
public synchronized Feed take() {
Feed feed = takeQueue.poll();
if (feed == null) {
int count = Math.min(100, 5 * backgroundThreads);
List<Feed> feeds = feedDAO.findNextUpdatable(count);
int addQueueSize = queue.size();
for (int i = 0; i < addQueueSize; i++) {
feeds.add(addQueue.poll());
}
feeds.addAll(addQueue);
for (Feed f : feeds) {
queue.add(f);
takeQueue.add(f);
f.setLastUpdated(Calendar.getInstance().getTime());
}
feeds.addAll(giveBackQueue);
feedDAO.update(feeds);
feed = queue.poll();
feed = takeQueue.poll();
}
metricsBean.feedRefreshed();
return feed;
}
public void giveBack(Feed feed) {
giveBackQueue.add(feed);
}
}

47
src/main/java/com/commafeed/backend/feeds/FeedRefreshUpdater.java
View File

@@ -1,69 +1,68 @@
package com.commafeed.backend.feeds;
import java.util.Collection;
import java.util.List;
import javax.ejb.Asynchronous;
import javax.ejb.Stateless;
import javax.inject.Inject;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.commafeed.backend.LockMap;
import com.commafeed.backend.dao.FeedDAO;
import com.commafeed.backend.dao.FeedSubscriptionDAO;
import com.commafeed.backend.model.Feed;
import com.commafeed.backend.model.FeedEntry;
import com.commafeed.backend.model.FeedEntryContent;
import com.commafeed.backend.model.FeedPushInfo;
import com.commafeed.backend.model.FeedSubscription;
import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
import com.commafeed.backend.services.ApplicationSettingsService;
import com.commafeed.backend.services.FeedUpdateService;
@Stateless
public class FeedRefreshUpdater {
protected static Logger log = LoggerFactory
.getLogger(FeedRefreshUpdater.class);
private static LockMap<String> lockMap = new LockMap<String>();
@Inject
FeedUpdateService feedUpdateService;
@Inject
SubscriptionHandler handler;
@Inject
FeedRefreshTaskGiver taskGiver;
@Inject
FeedDAO feedDAO;
@Inject
ApplicationSettingsService applicationSettingsService;
@Asynchronous
public void updateEntries(Feed feed, Collection<FeedEntry> entries) {
if (CollectionUtils.isNotEmpty(entries)) {
@Inject
FeedSubscriptionDAO feedSubscriptionDAO;
public void updateFeed(Feed feed, Collection<FeedEntry> entries) {
taskGiver.giveBack(feed);
if (entries != null) {
List<FeedSubscription> subscriptions = feedSubscriptionDAO
.findByFeed(feed);
for (FeedEntry entry : entries) {
handleEntry(feed, entry);
updateEntry(feed, entry, subscriptions);
}
feedUpdateService.updateEntries(feed, entries);
}
feedDAO.update(feed);
if (applicationSettingsService.get().isPubsubhubbub()) {
handlePubSub(feed);
}
}
private void handleEntry(Feed feed, FeedEntry entry) {
String baseUri = feed.getLink();
FeedEntryContent content = entry.getContent();
content.setContent(FeedUtils.handleContent(content.getContent(),
baseUri));
String title = FeedUtils.handleContent(content.getTitle(), baseUri);
if (title != null) {
content.setTitle(title.substring(0, Math.min(2048, title.length())));
}
String author = entry.getAuthor();
if (author != null) {
entry.setAuthor(author.substring(0, Math.min(128, author.length())));
private void updateEntry(Feed feed, FeedEntry entry,
List<FeedSubscription> subscriptions) {
synchronized (lockMap.get(entry.getGuid())) {
feedUpdateService.updateEntry(feed, entry, subscriptions);
}
}

13
src/main/java/com/commafeed/backend/feeds/FeedRefreshWorker.java
View File

@@ -5,12 +5,6 @@ import java.util.Collection;
import java.util.Date;
import javax.inject.Inject;
import javax.jms.JMSException;
import javax.transaction.HeuristicMixedException;
import javax.transaction.HeuristicRollbackException;
import javax.transaction.NotSupportedException;
import javax.transaction.RollbackException;
import javax.transaction.SystemException;
import org.apache.commons.lang.mutable.MutableBoolean;
import org.apache.commons.lang3.StringUtils;
@@ -77,10 +71,7 @@ public class FeedRefreshWorker {
}
}
private void update(Feed feed) throws NotSupportedException,
SystemException, SecurityException, IllegalStateException,
RollbackException, HeuristicMixedException,
HeuristicRollbackException, JMSException {
private void update(Feed feed) {
FetchedFeed fetchedFeed = null;
Collection<FeedEntry> entries = null;
@@ -130,7 +121,7 @@ public class FeedRefreshWorker {
feed.setMessage(message);
feed.setDisabledUntil(disabledUntil);
feedRefreshUpdater.updateEntries(feed, entries);
feedRefreshUpdater.updateFeed(feed, entries);
}