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

reformat using google-java-format 2.0

Browse Source
This commit is contained in:
Athou
2026-10-30 19:39:53 +01:00
parent 8f9be28002
commit f979f5426f
23 changed files with 382 additions and 441 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
commafeed-server/src/main/java/com/commafeed/backend/HttpClientFactory.java
View File

@@ -66,11 +66,10 @@ public class HttpClientFactory {
String userAgent =
config.httpClient()
.userAgent()
.orElseGet(
() ->
String.format(
"CommaFeed/%s (https://github.com/Athou/commafeed)",
version.getVersion()));
.orElseGet(() ->
String.format(
"CommaFeed/%s (https://github.com/Athou/commafeed)",
version.getVersion()));
return newClient(config, connectionManager, userAgent);
}

5
commafeed-server/src/main/java/com/commafeed/backend/HttpGetter.java
View File

@@ -283,9 +283,8 @@ public class HttpGetter {
}
return CacheBuilder.newBuilder()
.weigher(
(HttpRequest key, HttpResponse value) ->
value.content() != null ? value.content().length : 0)
.weigher((HttpRequest key, HttpResponse value) ->
value.content() != null ? value.content().length : 0)
.maximumWeight(cacheConfig.maximumMemorySize().asLongValue())
.expireAfterWrite(cacheConfig.expiration())
.build();

6
commafeed-server/src/main/java/com/commafeed/backend/dao/FeedSubscriptionDAO.java
View File

@@ -124,10 +124,8 @@ public class FeedSubscriptionDAO extends GenericDAO<FeedSubscription> {
Set<Long> categoryIds =
categories.stream().map(AbstractModel::getId).collect(Collectors.toSet());
return findAll(user).stream()
.filter(
s ->
s.getCategory() != null
&& categoryIds.contains(s.getCategory().getId()))
.filter(s ->
s.getCategory() != null && categoryIds.contains(s.getCategory().getId()))
.toList();
}

191
commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshEngine.java
View File

@@ -104,56 +104,52 @@ public class FeedRefreshEngine {
private void startFeedProcessingLoop() {
// take a feed from the queue, process it, rince, repeat
feedProcessingLoopExecutor.submit(
() -> {
while (!feedProcessingLoopExecutor.isShutdown()) {
try {
// take() is blocking until a feed is available from the queue
Feed feed = queue.take();
feedProcessingLoopExecutor.submit(() -> {
while (!feedProcessingLoopExecutor.isShutdown()) {
try {
// take() is blocking until a feed is available from the queue
Feed feed = queue.take();
// send the feed to be processed
log.debug(
"got feed {} from the queue, send it for processing",
feed.getId());
processFeedAsync(feed);
// send the feed to be processed
log.debug("got feed {} from the queue, send it for processing", feed.getId());
processFeedAsync(feed);
// we removed a feed from the queue, try to refill it as it may now be
// empty
if (queue.isEmpty()) {
log.debug("took the last feed from the queue, try to refill");
refillQueueAsync();
}
} catch (InterruptedException e) {
log.debug("interrupted while waiting for a feed in the queue");
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
// we removed a feed from the queue, try to refill it as it may now be
// empty
if (queue.isEmpty()) {
log.debug("took the last feed from the queue, try to refill");
refillQueueAsync();
}
});
} catch (InterruptedException e) {
log.debug("interrupted while waiting for a feed in the queue");
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
});
}
private void startRefillLoop() {
// refill the queue at regular intervals if it's empty
refillLoopExecutor.submit(
() -> {
while (!refillLoopExecutor.isShutdown()) {
try {
if (queue.isEmpty()) {
log.debug("refilling queue");
refillQueueAsync();
}
log.debug("sleeping for 15s");
TimeUnit.SECONDS.sleep(15);
} catch (InterruptedException e) {
log.debug("interrupted while sleeping");
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
refillLoopExecutor.submit(() -> {
while (!refillLoopExecutor.isShutdown()) {
try {
if (queue.isEmpty()) {
log.debug("refilling queue");
refillQueueAsync();
}
});
log.debug("sleeping for 15s");
TimeUnit.SECONDS.sleep(15);
} catch (InterruptedException e) {
log.debug("interrupted while sleeping");
Thread.currentThread().interrupt();
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
});
}
public void refreshImmediately(Feed feed) {
@@ -184,65 +180,53 @@ public class FeedRefreshEngine {
}
},
refillExecutor)
.whenComplete(
(data, ex) -> {
if (ex != null) {
log.error("error while refilling the queue", ex);
}
});
.whenComplete((data, ex) -> {
if (ex != null) {
log.error("error while refilling the queue", ex);
}
});
}
private void processFeedAsync(Feed feed) {
CompletableFuture.supplyAsync(() -> worker.update(feed), workerExecutor)
.thenApplyAsync(r -> updater.update(r.feed(), r.entries()), databaseUpdaterExecutor)
.thenCompose(
r -> {
List<CompletableFuture<Void>> futures =
r.insertedUnreadEntriesBySubscription().entrySet().stream()
.map(
e -> {
FeedSubscription sub = e.getKey();
List<FeedEntry> entries = e.getValue();
.thenCompose(r -> {
List<CompletableFuture<Void>> futures =
r.insertedUnreadEntriesBySubscription().entrySet().stream()
.map(e -> {
FeedSubscription sub = e.getKey();
List<FeedEntry> entries = e.getValue();
notifier.notifyOverWebsocket(sub, entries);
return CompletableFuture.runAsync(
() ->
notifier
.sendPushNotifications(
sub,
entries),
notifierExecutor);
})
.toList();
return CompletableFuture.allOf(
futures.toArray(CompletableFuture[]::new));
})
.exceptionally(
ex -> {
log.error("error while processing feed {}", feed.getUrl(), ex);
return null;
});
notifier.notifyOverWebsocket(sub, entries);
return CompletableFuture.runAsync(
() -> notifier.sendPushNotifications(sub, entries),
notifierExecutor);
})
.toList();
return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new));
})
.exceptionally(ex -> {
log.error("error while processing feed {}", feed.getUrl(), ex);
return null;
});
}
private List<Feed> getNextUpdatableFeeds(int max) {
return unitOfWork.call(
() -> {
Instant lastLoginThreshold =
config.feedRefresh().userInactivityPeriod().isZero()
? null
: Instant.now()
.minus(config.feedRefresh().userInactivityPeriod());
List<Feed> feeds = feedDAO.findNextUpdatable(max, lastLoginThreshold);
if (!feeds.isEmpty()) {
// update disabledUntil to prevent feeds from being returned again by
// feedDAO.findNextUpdatable()
Instant nextUpdateDate =
Instant.now().plus(config.feedRefresh().interval());
feedDAO.setDisabledUntil(
feeds.stream().map(AbstractModel::getId).toList(), nextUpdateDate);
}
return feeds;
});
return unitOfWork.call(() -> {
Instant lastLoginThreshold =
config.feedRefresh().userInactivityPeriod().isZero()
? null
: Instant.now().minus(config.feedRefresh().userInactivityPeriod());
List<Feed> feeds = feedDAO.findNextUpdatable(max, lastLoginThreshold);
if (!feeds.isEmpty()) {
// update disabledUntil to prevent feeds from being returned again by
// feedDAO.findNextUpdatable()
Instant nextUpdateDate = Instant.now().plus(config.feedRefresh().interval());
feedDAO.setDisabledUntil(
feeds.stream().map(AbstractModel::getId).toList(), nextUpdateDate);
}
return feeds;
});
}
private int getBatchSize() {
@@ -292,19 +276,18 @@ public class FeedRefreshEngine {
ThreadPoolExecutor pool =
new ThreadPoolExecutor(
threads, threads, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>());
pool.setRejectedExecutionHandler(
(r, e) -> {
if (e.isShutdown()) {
return;
}
pool.setRejectedExecutionHandler((r, e) -> {
if (e.isShutdown()) {
return;
}
try {
e.getQueue().put(r);
} catch (InterruptedException ex) {
log.debug("interrupted while waiting for a slot in the queue.", ex);
Thread.currentThread().interrupt();
}
});
try {
e.getQueue().put(r);
} catch (InterruptedException ex) {
log.debug("interrupted while waiting for a slot in the queue.", ex);
Thread.currentThread().interrupt();
}
});
return pool;
}
}

47
commafeed-server/src/main/java/com/commafeed/backend/feed/FeedRefreshUpdater.java
View File

@@ -91,24 +91,22 @@ public class FeedRefreshUpdater {
if (locked1 && locked2) {
processed = true;
insertedEntry =
unitOfWork.call(
() -> {
if (feedEntryService.find(feed, entry) != null) {
// entry already exists, nothing to do
return null;
}
unitOfWork.call(() -> {
if (feedEntryService.find(feed, entry) != null) {
// entry already exists, nothing to do
return null;
}
FeedEntry feedEntry = feedEntryService.create(feed, entry);
entryInserted.mark();
for (FeedSubscription sub : subscriptions) {
boolean unread =
feedEntryService.applyFilter(sub, feedEntry);
if (unread) {
subscriptionsForWhichEntryIsUnread.add(sub);
}
}
return feedEntry;
});
FeedEntry feedEntry = feedEntryService.create(feed, entry);
entryInserted.mark();
for (FeedSubscription sub : subscriptions) {
boolean unread = feedEntryService.applyFilter(sub, feedEntry);
if (unread) {
subscriptionsForWhichEntryIsUnread.add(sub);
}
}
return feedEntry;
});
} else {
log.error("lock timeout for {} - {}", feed.getUrl(), key1);
}
@@ -147,14 +145,13 @@ public class FeedRefreshUpdater {
AddEntryResult addEntryResult = addEntry(feed, entry, subscriptions);
processed &= addEntryResult.processed;
inserted += addEntryResult.insertedEntry != null ? 1 : 0;
addEntryResult.subscriptionsForWhichEntryIsUnread.forEach(
sub -> {
if (addEntryResult.insertedEntry != null) {
insertedUnreadEntriesBySubscription
.computeIfAbsent(sub, k -> new ArrayList<>())
.add(addEntryResult.insertedEntry);
}
});
addEntryResult.subscriptionsForWhichEntryIsUnread.forEach(sub -> {
if (addEntryResult.insertedEntry != null) {
insertedUnreadEntriesBySubscription
.computeIfAbsent(sub, k -> new ArrayList<>())
.add(addEntryResult.insertedEntry);
}
});
}
if (inserted == 0) {

13
commafeed-server/src/main/java/com/commafeed/backend/opml/OPMLExporter.java
View File

@@ -65,20 +65,17 @@ public class OPMLExporter {
for (FeedCategory child :
categories.stream()
.filter(
c ->
c.getParent() != null
&& c.getParent().getId().equals(cat.getId()))
.filter(c ->
c.getParent() != null && c.getParent().getId().equals(cat.getId()))
.toList()) {
outline.getChildren().add(buildCategoryOutline(child, categories, subscriptions));
}
for (FeedSubscription sub :
subscriptions.stream()
.filter(
s ->
s.getCategory() != null
&& s.getCategory().getId().equals(cat.getId()))
.filter(s ->
s.getCategory() != null
&& s.getCategory().getId().equals(cat.getId()))
.toList()) {
outline.getChildren().add(buildSubscriptionOutline(sub));
}

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

@@ -148,26 +148,23 @@ public class FeedEntryService {
List<FeedEntryStatus> statusesToMark =
statuses.stream()
.filter(FeedEntryStatus::isMarkable)
.filter(
s -> {
Instant entryDate = s.getEntry().getPublished();
return olderThan == null
|| entryDate == null
|| entryDate.isBefore(olderThan);
})
.filter(
s -> {
Instant insertedDate = s.getEntry().getInserted();
return insertedBefore == null
|| insertedDate == null
|| insertedDate.isBefore(insertedBefore);
})
.filter(s -> {
Instant entryDate = s.getEntry().getPublished();
return olderThan == null
|| entryDate == null
|| entryDate.isBefore(olderThan);
})
.filter(s -> {
Instant insertedDate = s.getEntry().getInserted();
return insertedBefore == null
|| insertedDate == null
|| insertedDate.isBefore(insertedBefore);
})
.toList();
statusesToMark.forEach(
s -> {
s.setRead(true);
feedEntryStatusDAO.merge(s);
});
statusesToMark.forEach(s -> {
s.setRead(true);
feedEntryStatusDAO.merge(s);
});
}
}

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

@@ -47,14 +47,13 @@ public class FeedSubscriptionService {
// we need to use this hook because the feed needs to have been persisted before being
// processed
// by the feed engine
feedSubscriptionDAO.onPostCommitInsert(
sub -> {
Feed feed = sub.getFeed();
if (feed.getDisabledUntil() == null
|| feed.getDisabledUntil().isBefore(Instant.now())) {
feedRefreshEngine.refreshImmediately(feed);
}
});
feedSubscriptionDAO.onPostCommitInsert(sub -> {
Feed feed = sub.getFeed();
if (feed.getDisabledUntil() == null
|| feed.getDisabledUntil().isBefore(Instant.now())) {
feedRefreshEngine.refreshImmediately(feed);
}
});
}
public long subscribe(

36
commafeed-server/src/main/java/com/commafeed/backend/service/db/DatabaseCleaningService.java
View File

@@ -79,13 +79,10 @@ public class DatabaseCleaningService {
} while (entriesDeleted > 0);
}
deleted =
unitOfWork.call(
() ->
feedDAO.delete(
feedDAO.findByIds(
feeds.stream()
.map(AbstractModel::getId)
.toList())));
unitOfWork.call(() ->
feedDAO.delete(
feedDAO.findByIds(
feeds.stream().map(AbstractModel::getId).toList())));
total += deleted;
log.debug("removed {} feeds without subscriptions", total);
} while (deleted != 0);
@@ -119,10 +116,9 @@ public class DatabaseCleaningService {
}
List<FeedCapacity> feeds =
unitOfWork.call(
() ->
feedEntryDAO.findFeedsExceedingCapacity(
maxFeedCapacity, batchSize, keepStarredEntries));
unitOfWork.call(() ->
feedEntryDAO.findFeedsExceedingCapacity(
maxFeedCapacity, batchSize, keepStarredEntries));
if (feeds.isEmpty()) {
break;
}
@@ -139,12 +135,11 @@ public class DatabaseCleaningService {
final long rem = remaining;
deleted =
unitOfWork.call(
() ->
feedEntryDAO.deleteOldEntries(
feed.id(),
Math.min(batchSize, rem),
keepStarredEntries));
unitOfWork.call(() ->
feedEntryDAO.deleteOldEntries(
feed.id(),
Math.min(batchSize, rem),
keepStarredEntries));
entriesDeletedMeter.mark(deleted);
total += deleted;
remaining -= deleted;
@@ -166,10 +161,9 @@ public class DatabaseCleaningService {
}
deleted =
unitOfWork.call(
() ->
feedEntryDAO.deleteEntriesOlderThan(
olderThan, batchSize, keepStarredEntries));
unitOfWork.call(() ->
feedEntryDAO.deleteEntriesOlderThan(
olderThan, batchSize, keepStarredEntries));
entriesDeletedMeter.mark(deleted);
total += deleted;
log.debug("removed {} old entries", total);

17
commafeed-server/src/main/java/com/commafeed/backend/task/DemoAccountCleanupTask.java
View File

@@ -31,16 +31,15 @@ public class DemoAccountCleanupTask extends ScheduledTask {
}
log.info("recreating demo user account");
unitOfWork.run(
() -> {
User demoUser = userDAO.findByName(CommaFeedConstants.USERNAME_DEMO);
if (demoUser == null) {
return;
}
unitOfWork.run(() -> {
User demoUser = userDAO.findByName(CommaFeedConstants.USERNAME_DEMO);
if (demoUser == null) {
return;
}
userService.unregister(demoUser);
userService.createDemoUser();
});
userService.unregister(demoUser);
userService.createDemoUser();
});
}
@Override