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

return to async beans, jms is too much overhead atm

Browse Source
This commit is contained in:
Athou
2013-04-25 17:26:07 +02:00
parent 84f9b4e10e
commit 2af55553a2
4 changed files with 55 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,81 +0,0 @@
package com.commafeed.backend.feeds;
import java.io.Serializable;
import java.util.Collection;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.MessageDriven;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.inject.Inject;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.commafeed.backend.dao.FeedDAO;
import com.commafeed.backend.model.Feed;
import com.commafeed.backend.model.FeedEntry;
import com.commafeed.backend.services.FeedUpdateService;
@MessageDriven(name = "FeedRefreshUpdater", activationConfig = {
@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
@ActivationConfigProperty(propertyName = "destination", propertyValue = "jms/refreshQueue") })
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public class FeedRefreshUpdater implements MessageListener {
private static Logger log = LoggerFactory
.getLogger(FeedRefreshUpdater.class);
@Inject
FeedDAO feedDAO;
@Inject
FeedUpdateService feedUpdateService;
@Override
public void onMessage(Message message) {
if (message instanceof ObjectMessage) {
ObjectMessage objectMessage = (ObjectMessage) message;
try {
FeedRefreshTask task = (FeedRefreshTask) objectMessage
.getObject();
if (task.getEntries() != null) {
feedUpdateService.updateEntries(task.getFeed(),
task.getEntries());
} else {
feedDAO.update(task.getFeed());
}
} catch (JMSException e) {
log.error(e.getMessage(), e);
}
}
}
public static class FeedRefreshTask implements Serializable {
private static final long serialVersionUID = 1L;
private Feed feed;
private Collection<FeedEntry> entries;
public FeedRefreshTask(Feed feed, Collection<FeedEntry> entries) {
this.feed = feed;
this.entries = entries;
}
public Feed getFeed() {
return feed;
}
public Collection<FeedEntry> getEntries() {
return entries;
}
}
}

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

@@ -4,16 +4,8 @@ import java.util.Calendar;
import java.util.Collection; import java.util.Collection;
import java.util.Date; import java.util.Date;
import javax.annotation.Resource;
import javax.inject.Inject; import javax.inject.Inject;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException; import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.Session;
import javax.transaction.HeuristicMixedException; import javax.transaction.HeuristicMixedException;
import javax.transaction.HeuristicRollbackException; import javax.transaction.HeuristicRollbackException;
import javax.transaction.NotSupportedException; import javax.transaction.NotSupportedException;
@@ -28,20 +20,17 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.commafeed.backend.HttpGetter.NotModifiedException; import com.commafeed.backend.HttpGetter.NotModifiedException;
import com.commafeed.backend.feeds.FeedRefreshUpdater.FeedRefreshTask;
import com.commafeed.backend.model.Feed; import com.commafeed.backend.model.Feed;
import com.commafeed.backend.model.FeedEntry; import com.commafeed.backend.model.FeedEntry;
import com.commafeed.backend.services.FeedUpdateService;
public class FeedRefreshWorker { public class FeedRefreshWorker {
private static Logger log = LoggerFactory private static Logger log = LoggerFactory
.getLogger(FeedRefreshWorker.class); .getLogger(FeedRefreshWorker.class);
@Resource(name = "jms/refreshQueue") @Inject
private Queue queue; FeedUpdateService feedUpdateService;
@Resource
private ConnectionFactory connectionFactory;
@Inject @Inject
FeedFetcher fetcher; FeedFetcher fetcher;
@@ -140,25 +129,10 @@ public class FeedRefreshWorker {
feed.setEtagHeader(fetchedFeed.getFeed().getEtagHeader()); feed.setEtagHeader(fetchedFeed.getFeed().getEtagHeader());
entries = fetchedFeed.getEntries(); entries = fetchedFeed.getEntries();
} }
FeedRefreshTask task = new FeedRefreshTask(feed, entries); feedUpdateService.updateEntries(feed, entries);
send(task);
} }
private void send(FeedRefreshTask task) throws JMSException {
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
MessageProducer producer = session.createProducer(queue);
producer.setDisableMessageID(true);
producer.setDisableMessageTimestamp(true);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
ObjectMessage message = session.createObjectMessage(task);
producer.send(message);
connection.close();
}
private Feed getNextFeed() { private Feed getNextFeed() {
return taskGiver.take(); return taskGiver.take();
} }

89
src/main/java/com/commafeed/backend/services/FeedUpdateService.java
View File

@@ -5,9 +5,13 @@ import java.util.Calendar;
import java.util.Collection; import java.util.Collection;
import java.util.List; import java.util.List;
import javax.ejb.Asynchronous;
import javax.ejb.Stateless; import javax.ejb.Stateless;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.inject.Inject; import javax.inject.Inject;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.ObjectUtils; import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
@@ -24,6 +28,7 @@ import com.commafeed.backend.model.FeedSubscription;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
@Stateless @Stateless
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
public class FeedUpdateService { public class FeedUpdateService {
@Inject @Inject
@@ -38,59 +43,63 @@ public class FeedUpdateService {
@Inject @Inject
FeedEntryStatusDAO feedEntryStatusDAO; FeedEntryStatusDAO feedEntryStatusDAO;
@Asynchronous
public void updateEntries(Feed feed, Collection<FeedEntry> entries) { public void updateEntries(Feed feed, Collection<FeedEntry> entries) {
List<FeedEntry> existingEntries = getExistingEntries(entries); if (CollectionUtils.isNotEmpty(entries)) {
List<FeedSubscription> subscriptions = feedSubscriptionDAO List<FeedEntry> existingEntries = getExistingEntries(entries);
.findByFeed(feed); List<FeedSubscription> subscriptions = feedSubscriptionDAO
.findByFeed(feed);
List<FeedEntry> entryUpdateList = Lists.newArrayList(); List<FeedEntry> entryUpdateList = Lists.newArrayList();
List<FeedEntryStatus> statusUpdateList = Lists.newArrayList(); List<FeedEntryStatus> statusUpdateList = Lists.newArrayList();
for (FeedEntry entry : entries) { for (FeedEntry entry : entries) {
FeedEntry foundEntry = findEntry(existingEntries, entry); FeedEntry foundEntry = findEntry(existingEntries, entry);
if (foundEntry == null) { if (foundEntry == null) {
FeedEntryContent content = entry.getContent(); FeedEntryContent content = entry.getContent();
content.setContent(FeedUtils.handleContent(content.getContent())); content.setContent(FeedUtils.handleContent(content
String title = FeedUtils.handleContent(content.getTitle()); .getContent()));
if (title != null) { String title = FeedUtils.handleContent(content.getTitle());
content.setTitle(title.substring(0, if (title != null) {
Math.min(2048, title.length()))); content.setTitle(title.substring(0,
} Math.min(2048, title.length())));
}
entry.setInserted(Calendar.getInstance().getTime()); entry.setInserted(Calendar.getInstance().getTime());
entry.getFeeds().add(feed); entry.getFeeds().add(feed);
entryUpdateList.add(entry); entryUpdateList.add(entry);
} else { } else {
boolean foundFeed = false; boolean foundFeed = false;
for (Feed existingFeed : foundEntry.getFeeds()) { for (Feed existingFeed : foundEntry.getFeeds()) {
if (ObjectUtils.equals(existingFeed.getId(), feed.getId())) { if (ObjectUtils.equals(existingFeed.getId(),
foundFeed = true; feed.getId())) {
break; foundFeed = true;
break;
}
}
if (!foundFeed) {
foundEntry.getFeeds().add(feed);
entryUpdateList.add(entry);
} }
} }
}
if (!foundFeed) { for (FeedEntry entry : entryUpdateList) {
foundEntry.getFeeds().add(feed); for (FeedSubscription sub : subscriptions) {
entryUpdateList.add(entry); FeedEntryStatus status = new FeedEntryStatus();
status.setEntry(entry);
status.setSubscription(sub);
statusUpdateList.add(status);
} }
} }
}
for (FeedEntry entry : entryUpdateList) {
for (FeedSubscription sub : subscriptions) {
FeedEntryStatus status = new FeedEntryStatus();
status.setEntry(entry);
status.setSubscription(sub);
statusUpdateList.add(status);
}
}
feedEntryDAO.saveOrUpdate(entryUpdateList);
feedEntryStatusDAO.saveOrUpdate(statusUpdateList);
}
feedDAO.update(feed); feedDAO.update(feed);
feedEntryDAO.saveOrUpdate(entryUpdateList);
feedEntryStatusDAO.saveOrUpdate(statusUpdateList);
} }
private FeedEntry findEntry(List<FeedEntry> existingEntries, FeedEntry entry) { private FeedEntry findEntry(List<FeedEntry> existingEntries, FeedEntry entry) {

4
src/main/tomee/conf/system.properties
View File

@@ -46,5 +46,5 @@
# javax.persistence.jtaDataSource = # javax.persistence.jtaDataSource =
# javax.persistence.nonJtaDataSource = # javax.persistence.nonJtaDataSource =
#AsynchronousPool.CorePoolSize = 1 AsynchronousPool.CorePoolSize = 1
#AsynchronousPool.MaximumPoolSize = 100 AsynchronousPool.MaximumPoolSize = 100