RDBKnownUriFilter implements KnownUriFilter, Closeable, UriHashCustodian { @Override public List<CrawleableUri> getOutdatedUris() { long generalRecrawlTime = Math.max(FrontierImpl.DEFAULT_GENERAL_RECRAWL_TIME, FrontierImpl.getGeneralRecrawlTime()); Cursor<HashMap> cursor = r.db(DATABASE_NAME) .table(TABLE_NAME) .filter(doc -> doc.getField(COLUMN_TIMESTAMP_NEXT_CRAWL).le(System.currentTimeMillis()).and( (doc.getField(COLUMN_CRAWLING_IN_PROCESS).eq(false)) .or(doc.getField(COLUMN_TIMESTAMP_LAST_CRAWL).le(System.currentTimeMillis() - generalRecrawlTime * 3)))) .run(connector.connection); List<CrawleableUri> urisToRecrawl = new ArrayList<>(); while (cursor.hasNext()) { try { HashMap row = cursor.next(); String ipString = (String) row.get(COLUMN_IP); if (ipString.contains("/")) { ipString = ipString.split("/")[1]; } urisToRecrawl.add(new CrawleableUri(new URI((String) row.get(COLUMN_URI)), InetAddress.getByName(ipString))); } catch (URISyntaxException | UnknownHostException e) { LOGGER.warn(e.toString()); } } for (CrawleableUri uri : urisToRecrawl) { r.db(DATABASE_NAME).table(TABLE_NAME).filter(doc -> doc.getField(COLUMN_URI).eq(uri.getUri().toString())). update(r.hashMap(COLUMN_CRAWLING_IN_PROCESS, true)).run(connector.connection); } cursor.close(); return urisToRecrawl; } RDBKnownUriFilter(String hostname, Integer port, boolean frontierDoesRecrawling);  RDBKnownUriFilter(String hostname, Integer port); @SuppressWarnings("unused")  RDBKnownUriFilter(RDBConnector connector, RethinkDB r, boolean frontierDoesRecrawling); void open(); void openConnector(); @Override List<CrawleableUri> getOutdatedUris(); boolean knownUriFilterTableExists(); void close(); @Override void add(CrawleableUri uri, long nextCrawlTimestamp); @Override void add(CrawleableUri uri, long lastCrawlTimestamp, long nextCrawlTimestamp); @Override Set<CrawleableUri> getUrisWithSameHashValues(Set<HashValue> hashValuesForComparison); @Override void addHashValuesForUris(List<CrawleableUri> uris); @Override boolean isUriGood(CrawleableUri uri); void purge(); @Override long count(); static final String DATABASE_NAME; static final String TABLE_NAME; static final String COLUMN_TIMESTAMP_LAST_CRAWL; static final String COLUMN_URI; static final String COLUMN_CRAWLING_IN_PROCESS; static final String COLUMN_TIMESTAMP_NEXT_CRAWL; static final String COLUMN_IP; static final String COLUMN_TYPE; static final String COLUMN_HASH_VALUE; }
@Test public void testGetOutdatedUris() throws URISyntaxException, UnknownHostException { this.connector.open(); r.dbCreate(RDBKnownUriFilter.DATABASE_NAME).run(this.connector.connection); r.db(RDBKnownUriFilter.DATABASE_NAME).tableCreate(RDBKnownUriFilter.TABLE_NAME).run(this.connector.connection); r.db(RDBKnownUriFilter.DATABASE_NAME).table(RDBKnownUriFilter.TABLE_NAME).indexCreate(RDBKnownUriFilter.COLUMN_URI).run(this.connector.connection); r.db(RDBKnownUriFilter.DATABASE_NAME).table(RDBKnownUriFilter.TABLE_NAME).indexWait(RDBKnownUriFilter.COLUMN_URI).run(this.connector.connection); CrawleableUri uri1 = new CrawleableUri(new URI("http: CrawleableUri uri2 = new CrawleableUri(new URI("http: filter.add(uri1, System.currentTimeMillis() - 10); filter.add(uri2, System.currentTimeMillis() + 50000); List<CrawleableUri> uris = filter.getOutdatedUris(); Assert.assertEquals(1, uris.size()); Assert.assertEquals(uri1, uris.get(0)); Cursor<Boolean> cursor = r.db(RDBKnownUriFilter.DATABASE_NAME).table(RDBKnownUriFilter.TABLE_NAME). filter(doc -> doc.getField(RDBKnownUriFilter.COLUMN_URI).eq(uri1.getUri().toString())). getField(RDBKnownUriFilter.COLUMN_CRAWLING_IN_PROCESS).run(connector.connection); Assert.assertTrue(cursor.next()); cursor = r.db(RDBKnownUriFilter.DATABASE_NAME).table(RDBKnownUriFilter.TABLE_NAME). filter(doc -> doc.getField(RDBKnownUriFilter.COLUMN_URI).eq(uri2.getUri().toString())). getField(RDBKnownUriFilter.COLUMN_CRAWLING_IN_PROCESS).run(connector.connection); Assert.assertFalse(cursor.next()); uris = filter.getOutdatedUris(); Assert.assertTrue(uris.isEmpty()); r.db(RDBKnownUriFilter.DATABASE_NAME).table(RDBKnownUriFilter.TABLE_NAME). filter(doc -> doc.getField(RDBKnownUriFilter.COLUMN_URI).eq(uri1.getUri().toString())). update(r.hashMap(RDBKnownUriFilter.COLUMN_TIMESTAMP_LAST_CRAWL, System.currentTimeMillis() - 10 * FrontierImpl.DEFAULT_GENERAL_RECRAWL_TIME)).run(connector.connection); uris = filter.getOutdatedUris(); Assert.assertEquals(1, uris.size()); Assert.assertEquals(uri1, uris.get(0)); cursor.close(); }