mirror of
https://github.com/Athou/commafeed.git
synced 2026-09-22 12:04:16 +00:00
Merge branch 'perf/batch-guid-check-on-feed-refresh' of https://github.com/ingokegel/commafeed into ingokegel-perf/batch-guid-check-on-feed-refresh
This commit is contained in:
@@ -1,7 +1,10 @@
|
||||
package com.commafeed.backend.dao;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Set;
|
||||
|
||||
import jakarta.inject.Singleton;
|
||||
import jakarta.persistence.EntityManager;
|
||||
@@ -9,6 +12,7 @@ import jakarta.persistence.EntityManager;
|
||||
import com.commafeed.backend.model.Feed;
|
||||
import com.commafeed.backend.model.FeedEntry;
|
||||
import com.commafeed.backend.model.QFeedEntry;
|
||||
import com.google.common.collect.Lists;
|
||||
import com.querydsl.core.Tuple;
|
||||
import com.querydsl.core.types.dsl.NumberExpression;
|
||||
import com.querydsl.jpa.impl.JPAQuery;
|
||||
@@ -17,6 +21,7 @@ import com.querydsl.jpa.impl.JPAQuery;
|
||||
public class FeedEntryDAO extends GenericDAO<FeedEntry> {
|
||||
|
||||
private static final QFeedEntry ENTRY = QFeedEntry.feedEntry;
|
||||
private static final int IN_CLAUSE_BATCH_SIZE = 1000;
|
||||
|
||||
public FeedEntryDAO(EntityManager entityManager) {
|
||||
super(entityManager, FeedEntry.class);
|
||||
@@ -26,6 +31,18 @@ public class FeedEntryDAO extends GenericDAO<FeedEntry> {
|
||||
return query().select(ENTRY).from(ENTRY).where(ENTRY.guidHash.eq(guidHash), ENTRY.feed.eq(feed)).limit(1).fetchOne();
|
||||
}
|
||||
|
||||
public Set<String> findExistingGuids(Feed feed, Set<String> guidHashes) {
|
||||
if (guidHashes.isEmpty()) {
|
||||
return Set.of();
|
||||
}
|
||||
|
||||
Set<String> result = new HashSet<>();
|
||||
for (List<String> batch : Lists.partition(new ArrayList<>(guidHashes), IN_CLAUSE_BATCH_SIZE)) {
|
||||
result.addAll(query().select(ENTRY.guidHash).from(ENTRY).where(ENTRY.feed.eq(feed), ENTRY.guidHash.in(batch)).fetch());
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
public List<FeedCapacity> findFeedsExceedingCapacity(long maxCapacity, long max, boolean keepStarredEntries) {
|
||||
NumberExpression<Long> count = ENTRY.id.count();
|
||||
JPAQuery<Tuple> query = query().select(ENTRY.feed.id, count).from(ENTRY);
|
||||
|
||||
@@ -18,6 +18,7 @@ import org.apache.commons.lang3.StringUtils;
|
||||
import com.codahale.metrics.Meter;
|
||||
import com.codahale.metrics.MetricRegistry;
|
||||
import com.commafeed.backend.Digests;
|
||||
import com.commafeed.backend.dao.FeedEntryDAO;
|
||||
import com.commafeed.backend.dao.FeedSubscriptionDAO;
|
||||
import com.commafeed.backend.dao.UnitOfWork;
|
||||
import com.commafeed.backend.feed.parser.FeedParserResult.Content;
|
||||
@@ -42,6 +43,7 @@ public class FeedRefreshUpdater {
|
||||
private final UnitOfWork unitOfWork;
|
||||
private final FeedService feedService;
|
||||
private final FeedEntryService feedEntryService;
|
||||
private final FeedEntryDAO feedEntryDAO;
|
||||
private final FeedSubscriptionDAO feedSubscriptionDAO;
|
||||
|
||||
private final Striped<Lock> locks;
|
||||
@@ -49,11 +51,12 @@ public class FeedRefreshUpdater {
|
||||
private final Meter feedUpdated;
|
||||
private final Meter entryInserted;
|
||||
|
||||
public FeedRefreshUpdater(UnitOfWork unitOfWork, FeedService feedService, FeedEntryService feedEntryService, MetricRegistry metrics,
|
||||
FeedSubscriptionDAO feedSubscriptionDAO) {
|
||||
public FeedRefreshUpdater(UnitOfWork unitOfWork, FeedService feedService, FeedEntryService feedEntryService, FeedEntryDAO feedEntryDAO,
|
||||
MetricRegistry metrics, FeedSubscriptionDAO feedSubscriptionDAO) {
|
||||
this.unitOfWork = unitOfWork;
|
||||
this.feedService = feedService;
|
||||
this.feedEntryService = feedEntryService;
|
||||
this.feedEntryDAO = feedEntryDAO;
|
||||
this.feedSubscriptionDAO = feedSubscriptionDAO;
|
||||
|
||||
locks = Striped.lazyWeakLock(100000);
|
||||
@@ -126,8 +129,19 @@ public class FeedRefreshUpdater {
|
||||
Map<FeedSubscription, List<FeedEntry>> insertedUnreadEntriesBySubscription = new HashMap<>();
|
||||
|
||||
if (!entries.isEmpty()) {
|
||||
List<FeedSubscription> subscriptions = null;
|
||||
Map<String, Entry> entriesByGuidHash = new HashMap<>();
|
||||
for (Entry entry : entries) {
|
||||
entriesByGuidHash.put(Digests.sha1Hex(entry.guid()), entry);
|
||||
}
|
||||
Set<String> existingGuids = unitOfWork.call(() -> feedEntryDAO.findExistingGuids(feed, entriesByGuidHash.keySet()));
|
||||
List<Entry> newEntries = entriesByGuidHash.entrySet()
|
||||
.stream()
|
||||
.filter(e -> !existingGuids.contains(e.getKey()))
|
||||
.map(Map.Entry::getValue)
|
||||
.toList();
|
||||
|
||||
List<FeedSubscription> subscriptions = null;
|
||||
for (Entry entry : newEntries) {
|
||||
if (subscriptions == null) {
|
||||
subscriptions = unitOfWork.call(() -> feedSubscriptionDAO.findByFeed(feed));
|
||||
}
|
||||
|
||||
@@ -0,0 +1,134 @@
|
||||
package com.commafeed.integration.rest;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
|
||||
import org.awaitility.Awaitility;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Assertions;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockserver.model.HttpRequest;
|
||||
import org.mockserver.model.HttpResponse;
|
||||
import org.mockserver.model.MediaType;
|
||||
|
||||
import com.commafeed.TestConstants;
|
||||
import com.commafeed.frontend.model.Entries;
|
||||
import com.commafeed.frontend.model.Entry;
|
||||
import com.commafeed.frontend.model.request.MarkRequest;
|
||||
import com.commafeed.frontend.resource.CategoryREST;
|
||||
import com.commafeed.integration.BaseIT;
|
||||
|
||||
import io.quarkus.test.junit.QuarkusTest;
|
||||
import io.restassured.RestAssured;
|
||||
import io.restassured.http.ContentType;
|
||||
|
||||
@QuarkusTest
|
||||
class LargeDatasetIT extends BaseIT {
|
||||
|
||||
private static final int FEED_COUNT = 10;
|
||||
private static final int ENTRIES_PER_FEED = 20;
|
||||
private static final int TOTAL_ENTRIES = FEED_COUNT * ENTRIES_PER_FEED;
|
||||
|
||||
private Long firstSubscriptionId;
|
||||
|
||||
@BeforeEach
|
||||
void setup() {
|
||||
initialSetup(TestConstants.ADMIN_USERNAME, TestConstants.ADMIN_PASSWORD);
|
||||
RestAssured.authentication = RestAssured.preemptive().basic(TestConstants.ADMIN_USERNAME, TestConstants.ADMIN_PASSWORD);
|
||||
|
||||
for (int i = 0; i < FEED_COUNT; i++) {
|
||||
String path = "/feed/" + i;
|
||||
getMockServerClient().when(HttpRequest.request().withMethod("GET").withPath(path))
|
||||
.respond(HttpResponse.response().withBody(generateFeed(i)).withContentType(MediaType.APPLICATION_XML));
|
||||
Long subscriptionId = subscribe("http://localhost:" + getMockServerClient().getPort() + path);
|
||||
if (i == 0) {
|
||||
firstSubscriptionId = subscriptionId;
|
||||
}
|
||||
}
|
||||
|
||||
Awaitility.await().atMost(Duration.ofSeconds(60)).until(() -> getAllEntries().getEntries().size(), count -> count >= TOTAL_ENTRIES);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
RestAssured.reset();
|
||||
}
|
||||
|
||||
@Test
|
||||
void markAllAsRead() {
|
||||
Entries before = getAllEntries();
|
||||
Assertions.assertEquals(TOTAL_ENTRIES, before.getEntries().size());
|
||||
Assertions.assertTrue(before.getEntries().stream().noneMatch(Entry::isRead));
|
||||
|
||||
MarkRequest markRequest = new MarkRequest();
|
||||
markRequest.setId(CategoryREST.ALL);
|
||||
markRequest.setRead(true);
|
||||
RestAssured.given().body(markRequest).contentType(ContentType.JSON).post("rest/category/mark").then().statusCode(200);
|
||||
|
||||
Entries after = getAllEntries();
|
||||
Assertions.assertEquals(TOTAL_ENTRIES, after.getEntries().size());
|
||||
Assertions.assertTrue(after.getEntries().stream().allMatch(Entry::isRead));
|
||||
}
|
||||
|
||||
@Test
|
||||
void refreshDoesNotCreateDuplicateEntries() {
|
||||
Assertions.assertEquals(TOTAL_ENTRIES, getAllEntries().getEntries().size());
|
||||
Instant threshold = Instant.now().minus(Duration.ofSeconds(1));
|
||||
forceRefreshAllFeeds();
|
||||
|
||||
Awaitility.await()
|
||||
.atMost(Duration.ofSeconds(15))
|
||||
.until(() -> getSubscription(firstSubscriptionId), f -> f.getLastRefresh().isAfter(threshold));
|
||||
Assertions.assertEquals(TOTAL_ENTRIES, getAllEntries().getEntries().size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void paginationHasMore() {
|
||||
Entries firstPage = RestAssured.given()
|
||||
.get("rest/category/entries?id=all&readType=all&limit=20&offset=0")
|
||||
.then()
|
||||
.statusCode(200)
|
||||
.extract()
|
||||
.as(Entries.class);
|
||||
Assertions.assertEquals(20, firstPage.getEntries().size());
|
||||
Assertions.assertTrue(firstPage.isHasMore());
|
||||
|
||||
Entries lastPage = RestAssured.given()
|
||||
.get("rest/category/entries?id=all&readType=all&limit=20&offset={offset}", TOTAL_ENTRIES - 20)
|
||||
.then()
|
||||
.statusCode(200)
|
||||
.extract()
|
||||
.as(Entries.class);
|
||||
Assertions.assertEquals(20, lastPage.getEntries().size());
|
||||
Assertions.assertFalse(lastPage.isHasMore());
|
||||
}
|
||||
|
||||
private Entries getAllEntries() {
|
||||
return RestAssured.given()
|
||||
.get("rest/category/entries?id=all&readType=all&limit=1000")
|
||||
.then()
|
||||
.statusCode(200)
|
||||
.extract()
|
||||
.as(Entries.class);
|
||||
}
|
||||
|
||||
private String generateFeed(int feedIndex) {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append("<?xml version=\"1.0\" encoding=\"UTF-8\" ?>\n<rss version=\"2.0\">\n<channel>\n");
|
||||
sb.append("<title>Feed ").append(feedIndex).append("</title>\n");
|
||||
sb.append("<link>https://hostname.local/feed/").append(feedIndex).append("</link>\n");
|
||||
sb.append("<description>Test feed ").append(feedIndex).append("</description>\n");
|
||||
Instant base = Instant.parse("2024-01-01T00:00:00Z");
|
||||
for (int i = 0; i < ENTRIES_PER_FEED; i++) {
|
||||
sb.append("<item>\n");
|
||||
sb.append("<title>Feed ").append(feedIndex).append(" Item ").append(i).append("</title>\n");
|
||||
sb.append("<link>https://hostname.local/feed/").append(feedIndex).append("/item/").append(i).append("</link>\n");
|
||||
sb.append("<description>Description for feed ").append(feedIndex).append(" item ").append(i).append("</description>\n");
|
||||
sb.append("<pubDate>").append(base.minus(Duration.ofHours(i))).append("</pubDate>\n");
|
||||
sb.append("</item>\n");
|
||||
}
|
||||
sb.append("</channel>\n</rss>");
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user