crawler: add delay function

dethi · Dec 9, 2017 · 0cd11e7 · 0cd11e7
1 parent 939b7b0
commit 0cd11e7
Show file tree

Hide file tree

Showing 3 changed files with 24 additions and 16 deletions.
diff --git a/src/main/java/com/epita/guereza/App.java b/src/main/java/com/epita/guereza/App.java
@@ -7,6 +7,9 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public abstract class App {
     private static final Logger LOGGER = LoggerFactory.getLogger(App.class);
@@ -38,4 +41,11 @@ public void sendMessage(final String channel, final Object obj) {
             LOGGER.error("Impossible to send message: {}", e.getMessage());
         }
     }
+
+    protected void retryIn(final int seconds, Runnable consumer) {
+        LOGGER.info("Retry fetching url in {}seconds", seconds);
+        final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+        executor.schedule(consumer, seconds, TimeUnit.SECONDS);
+        executor.shutdownNow();
+    }
 }
diff --git a/src/main/java/com/epita/guereza/CrawlerApp.java b/src/main/java/com/epita/guereza/CrawlerApp.java
@@ -38,12 +38,16 @@ private void storeUrls(final String[] urls) {
 
     @Override
     public void run() {
-        eventBus.subscribe(subscribeUrl, c -> {
-            System.out.println(c.getContent());
-            final String[] urls = crawlAndExtract(c.getContent());
-            storeUrls(urls);
-
-            requestNextUrl();
+        eventBus.subscribe(subscribeUrl, msg -> {
+            if (msg != null) {
+                LOGGER.info("Receive url: {}", msg.getContent());
+                final String[] urls = crawlAndExtract(msg.getContent());
+                storeUrls(urls);
+
+                requestNextUrl();
+            } else {
+                retryIn(30, this::requestNextUrl);
+            }
         });
 
         requestNextUrl();

diff --git a/src/main/java/com/epita/guereza/IndexerApp.java b/src/main/java/com/epita/guereza/IndexerApp.java
@@ -11,6 +11,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 public class IndexerApp extends App {
     private static final Logger LOGGER = LoggerFactory.getLogger(IndexerApp.class);
@@ -44,24 +45,17 @@ private void requestNextUrl() {
         sendMessage("/request/indexer/url", subscribeUrl);
     }
 
-    private void retryIn(final int seconds) {
-        LOGGER.info("Retry fetching url in {}seconds", seconds);
-        final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
-        executor.schedule(this::requestNextUrl, seconds, TimeUnit.SECONDS);
-        executor.shutdownNow();
-    }
 
     @Override
     public void run() {
         eventBus.subscribe(subscribeUrl, msg -> {
             if (msg != null) {
-                final String url = msg.getContent();
-                LOGGER.info("Receive url: {}", url);
-                indexAndPublish(url);
+                LOGGER.info("Receive url: {}", msg.getContent());
+                indexAndPublish(msg.getContent());
 
                 requestNextUrl();
             } else {
-                retryIn(30);
+                retryIn(30, this::requestNextUrl);
             }
         });