forked from Archives/Athou_commafeed
15
pom.xml
15
pom.xml
@@ -31,16 +31,6 @@
|
|||||||
<enabled>true</enabled>
|
<enabled>true</enabled>
|
||||||
</snapshots>
|
</snapshots>
|
||||||
</repository>
|
</repository>
|
||||||
<repository>
|
|
||||||
<id>jklm.releases</id>
|
|
||||||
<url>http://mvn.jkeylockmanager.de</url>
|
|
||||||
<releases>
|
|
||||||
<enabled>true</enabled>
|
|
||||||
</releases>
|
|
||||||
<snapshots>
|
|
||||||
<enabled>false</enabled>
|
|
||||||
</snapshots>
|
|
||||||
</repository>
|
|
||||||
</repositories>
|
</repositories>
|
||||||
|
|
||||||
<build>
|
<build>
|
||||||
@@ -244,11 +234,6 @@
|
|||||||
<artifactId>joda-time</artifactId>
|
<artifactId>joda-time</artifactId>
|
||||||
<version>2.2</version>
|
<version>2.2</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
<dependency>
|
|
||||||
<groupId>de.jkeylockmanager</groupId>
|
|
||||||
<artifactId>jkeylockmanager</artifactId>
|
|
||||||
<version>1.0.0</version>
|
|
||||||
</dependency>
|
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>net.java.dev.rome</groupId>
|
<groupId>net.java.dev.rome</groupId>
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ import org.apache.http.conn.ssl.X509HostnameVerifier;
|
|||||||
import org.apache.http.impl.client.DecompressingHttpClient;
|
import org.apache.http.impl.client.DecompressingHttpClient;
|
||||||
import org.apache.http.impl.client.DefaultHttpClient;
|
import org.apache.http.impl.client.DefaultHttpClient;
|
||||||
import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
|
import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
|
||||||
|
import org.apache.http.impl.client.SystemDefaultHttpClient;
|
||||||
import org.apache.http.params.HttpConnectionParams;
|
import org.apache.http.params.HttpConnectionParams;
|
||||||
import org.apache.http.params.HttpParams;
|
import org.apache.http.params.HttpParams;
|
||||||
import org.apache.http.params.HttpProtocolParams;
|
import org.apache.http.params.HttpProtocolParams;
|
||||||
@@ -182,7 +183,7 @@ public class HttpGetter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public static HttpClient newClient() {
|
public static HttpClient newClient() {
|
||||||
DefaultHttpClient client = new DefaultHttpClient();
|
DefaultHttpClient client = new SystemDefaultHttpClient();
|
||||||
|
|
||||||
SSLSocketFactory ssf = new SSLSocketFactory(SSL_CONTEXT, VERIFIER);
|
SSLSocketFactory ssf = new SSLSocketFactory(SSL_CONTEXT, VERIFIER);
|
||||||
ClientConnectionManager ccm = client.getConnectionManager();
|
ClientConnectionManager ccm = client.getConnectionManager();
|
||||||
|
|||||||
@@ -76,9 +76,9 @@ public class FeedParser {
|
|||||||
|
|
||||||
entry.setGuid(FeedUtils.truncate(item.getUri(), 2048));
|
entry.setGuid(FeedUtils.truncate(item.getUri(), 2048));
|
||||||
entry.setGuidHash(DigestUtils.sha1Hex(item.getUri()));
|
entry.setGuidHash(DigestUtils.sha1Hex(item.getUri()));
|
||||||
entry.setUrl(FeedUtils.toAbsoluteUrl(
|
entry.setUrl(FeedUtils.truncate(
|
||||||
FeedUtils.truncate(item.getLink(), 2048),
|
FeedUtils.toAbsoluteUrl(item.getLink(), feed.getLink()),
|
||||||
feed.getLink()));
|
2048));
|
||||||
entry.setUpdated(validateDate(getUpdateDate(item)));
|
entry.setUpdated(validateDate(getUpdateDate(item)));
|
||||||
entry.setAuthor(FeedUtils.truncate(item.getAuthor(), 128));
|
entry.setAuthor(FeedUtils.truncate(item.getAuthor(), 128));
|
||||||
|
|
||||||
|
|||||||
@@ -3,9 +3,11 @@ package com.commafeed.backend.feeds;
|
|||||||
import java.util.Collection;
|
import java.util.Collection;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import java.util.concurrent.ArrayBlockingQueue;
|
import java.util.concurrent.ArrayBlockingQueue;
|
||||||
|
import java.util.concurrent.BlockingQueue;
|
||||||
import java.util.concurrent.RejectedExecutionHandler;
|
import java.util.concurrent.RejectedExecutionHandler;
|
||||||
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.ThreadPoolExecutor;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.locks.Lock;
|
||||||
|
|
||||||
import javax.annotation.PostConstruct;
|
import javax.annotation.PostConstruct;
|
||||||
import javax.annotation.PreDestroy;
|
import javax.annotation.PreDestroy;
|
||||||
@@ -26,10 +28,7 @@ import com.commafeed.backend.model.FeedSubscription;
|
|||||||
import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
|
import com.commafeed.backend.pubsubhubbub.SubscriptionHandler;
|
||||||
import com.commafeed.backend.services.ApplicationSettingsService;
|
import com.commafeed.backend.services.ApplicationSettingsService;
|
||||||
import com.commafeed.backend.services.FeedUpdateService;
|
import com.commafeed.backend.services.FeedUpdateService;
|
||||||
|
import com.google.common.util.concurrent.Striped;
|
||||||
import de.jkeylockmanager.manager.KeyLockManager;
|
|
||||||
import de.jkeylockmanager.manager.KeyLockManagers;
|
|
||||||
import de.jkeylockmanager.manager.LockCallback;
|
|
||||||
|
|
||||||
@Singleton
|
@Singleton
|
||||||
public class FeedRefreshUpdater {
|
public class FeedRefreshUpdater {
|
||||||
@@ -37,8 +36,6 @@ public class FeedRefreshUpdater {
|
|||||||
protected static Logger log = LoggerFactory
|
protected static Logger log = LoggerFactory
|
||||||
.getLogger(FeedRefreshUpdater.class);
|
.getLogger(FeedRefreshUpdater.class);
|
||||||
|
|
||||||
private static final KeyLockManager lockManager = KeyLockManagers.newLock();
|
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
FeedUpdateService feedUpdateService;
|
FeedUpdateService feedUpdateService;
|
||||||
|
|
||||||
@@ -53,7 +50,7 @@ public class FeedRefreshUpdater {
|
|||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
ApplicationSettingsService applicationSettingsService;
|
ApplicationSettingsService applicationSettingsService;
|
||||||
|
|
||||||
@Inject
|
@Inject
|
||||||
MetricsBean metricsBean;
|
MetricsBean metricsBean;
|
||||||
|
|
||||||
@@ -61,15 +58,18 @@ public class FeedRefreshUpdater {
|
|||||||
FeedSubscriptionDAO feedSubscriptionDAO;
|
FeedSubscriptionDAO feedSubscriptionDAO;
|
||||||
|
|
||||||
private ThreadPoolExecutor pool;
|
private ThreadPoolExecutor pool;
|
||||||
|
private BlockingQueue<Runnable> queue;
|
||||||
|
private Striped<Lock> locks;
|
||||||
|
|
||||||
@PostConstruct
|
@PostConstruct
|
||||||
public void init() {
|
public void init() {
|
||||||
ApplicationSettings settings = applicationSettingsService.get();
|
ApplicationSettings settings = applicationSettingsService.get();
|
||||||
int threads = Math.max(settings.getDatabaseUpdateThreads(), 1);
|
int threads = Math.max(settings.getDatabaseUpdateThreads(), 1);
|
||||||
log.info("Creating database pool with {} threads", threads);
|
log.info("Creating database pool with {} threads", threads);
|
||||||
|
locks = Striped.lazyWeakLock(threads);
|
||||||
pool = new ThreadPoolExecutor(threads, threads, 0,
|
pool = new ThreadPoolExecutor(threads, threads, 0,
|
||||||
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(
|
TimeUnit.MILLISECONDS,
|
||||||
100 * threads));
|
queue = new ArrayBlockingQueue<Runnable>(100 * threads));
|
||||||
pool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
|
pool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
|
||||||
@Override
|
@Override
|
||||||
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
|
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
|
||||||
@@ -130,12 +130,16 @@ public class FeedRefreshUpdater {
|
|||||||
|
|
||||||
private void updateEntry(final Feed feed, final FeedEntry entry,
|
private void updateEntry(final Feed feed, final FeedEntry entry,
|
||||||
final List<FeedSubscription> subscriptions) {
|
final List<FeedSubscription> subscriptions) {
|
||||||
lockManager.executeLocked(entry.getGuid(), new LockCallback() {
|
Lock lock = locks.get(entry.getGuid());
|
||||||
@Override
|
try {
|
||||||
public void doInLock() throws Exception {
|
lock.tryLock(1, TimeUnit.MINUTES);
|
||||||
feedUpdateService.updateEntry(feed, entry, subscriptions);
|
feedUpdateService.updateEntry(feed, entry, subscriptions);
|
||||||
}
|
} catch (InterruptedException e) {
|
||||||
});
|
log.error("interrupted while waiting for lock for " + feed.getUrl()
|
||||||
|
+ " : " + e.getMessage(), e);
|
||||||
|
} finally {
|
||||||
|
lock.unlock();
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void handlePubSub(final Feed feed) {
|
private void handlePubSub(final Feed feed) {
|
||||||
@@ -150,4 +154,8 @@ public class FeedRefreshUpdater {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public int getQueueSize() {
|
||||||
|
return queue.size();
|
||||||
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -36,6 +36,7 @@ import com.commafeed.backend.dao.UserRoleDAO;
|
|||||||
import com.commafeed.backend.dao.UserSettingsDAO;
|
import com.commafeed.backend.dao.UserSettingsDAO;
|
||||||
import com.commafeed.backend.feeds.FeedFetcher;
|
import com.commafeed.backend.feeds.FeedFetcher;
|
||||||
import com.commafeed.backend.feeds.FeedRefreshTaskGiver;
|
import com.commafeed.backend.feeds.FeedRefreshTaskGiver;
|
||||||
|
import com.commafeed.backend.feeds.FeedRefreshUpdater;
|
||||||
import com.commafeed.backend.feeds.OPMLExporter;
|
import com.commafeed.backend.feeds.OPMLExporter;
|
||||||
import com.commafeed.backend.feeds.OPMLImporter;
|
import com.commafeed.backend.feeds.OPMLImporter;
|
||||||
import com.commafeed.backend.model.User;
|
import com.commafeed.backend.model.User;
|
||||||
@@ -116,6 +117,9 @@ public abstract class AbstractREST {
|
|||||||
@Inject
|
@Inject
|
||||||
FeedRefreshTaskGiver taskGiver;
|
FeedRefreshTaskGiver taskGiver;
|
||||||
|
|
||||||
|
@Inject
|
||||||
|
FeedRefreshUpdater feedRefreshUpdater;
|
||||||
|
|
||||||
@PostConstruct
|
@PostConstruct
|
||||||
public void init() {
|
public void init() {
|
||||||
CommaFeedApplication app = CommaFeedApplication.get();
|
CommaFeedApplication app = CommaFeedApplication.get();
|
||||||
|
|||||||
@@ -179,7 +179,8 @@ public class AdminREST extends AbstractResourceREST {
|
|||||||
Map<String, ? extends Object> map = ImmutableMap.of("lastMinute",
|
Map<String, ? extends Object> map = ImmutableMap.of("lastMinute",
|
||||||
metricsBean.getLastMinute(), "lastHour",
|
metricsBean.getLastMinute(), "lastHour",
|
||||||
metricsBean.getLastHour(), "backlog",
|
metricsBean.getLastHour(), "backlog",
|
||||||
feedDAO.getUpdatableCount());
|
feedDAO.getUpdatableCount(), "queue",
|
||||||
|
feedRefreshUpdater.getQueueSize());
|
||||||
return Response.ok(map).build();
|
return Response.ok(map).build();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user