Merge branch 'master' into remove

pom.xml

@@ -10,7 +10,7 @@
 
     <groupId>ai.preferred</groupId>
     <artifactId>venom</artifactId>
-    <version>4.2.2-SNAPSHOT</version>
+    <version>4.2.3-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>${project.groupId}:${project.artifactId}</name>

src/main/java/ai/preferred/venom/Crawler.java

@@ -47,7 +47,7 @@
  * @author Truong Quoc Tuan
  * @author Ween Jiann Lee
  */
-public final class Crawler implements Interruptible {
+public final class Crawler implements Interruptible, AutoCloseable {
 
   /**
    * Logger.
@@ -383,11 +383,7 @@ public void cancelled(final @NotNull Request request) {
     }
     if (!fatalHandlerExceptions.isEmpty()) {
       LOGGER.debug("Handler exception found... Interrupting.");
-      try {
-        interrupt();
-      } catch (final Exception e) {
-        throw new RuntimeException(e);
-      }
+      interrupt();
     }
     LOGGER.debug("({}) will stop producing requests.", crawlerThread.getName());
   }
@@ -425,44 +421,33 @@ public synchronized Crawler startAndClose() throws Exception {
     return this;
   }
 
-  @Override
+  /**
+   * Interrupts then close this object.
+   *
+   * @throws Exception if exception is thrown on close.
+   */
   public void interruptAndClose() throws Exception {
     interrupt();
-
-    try {
-      crawlerThread.join();
-      threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
-    } catch (final InterruptedException e) {
-      LOGGER.warn("The joining has been interrupted!", e);
-      Thread.currentThread().interrupt();
-    }
+    close();
   }
 
   /**
    * Interrupts crawler, fetcher and worker threads.
-   *
-   * @throws Exception if any resources throws an exception on close.
    */
-  private void interrupt() throws Exception {
-    exitWhenDone.set(true);
-    crawlerThread.interrupt();
-    threadPool.shutdownNow();
+  @Override
+  public void interrupt() {
+    if (!Thread.currentThread().equals(crawlerThread) && crawlerThread.isAlive()) {
+      crawlerThread.interrupt();
+    }
 
-    Exception cachedException = null;
-    for (final Interruptible interruptible : new Interruptible[]{workerManager, fetcher}) {
-      try {
-        interruptible.interruptAndClose();
-      } catch (final Exception e) {
-        if (cachedException != null) {
-          cachedException.addSuppressed(e);
-        } else {
-          cachedException = e;
-        }
-      }
+    if (!threadPool.isTerminated()) {
+      threadPool.shutdownNow();
     }
 
-    if (cachedException != null) {
-      throw cachedException;
+    workerManager.interrupt();
+
+    if (fetcher instanceof Interruptible) {
+      ((Interruptible) fetcher).interrupt();
     }
   }
 
@@ -476,16 +461,17 @@ public void close() throws Exception {
         LOGGER.debug("{} producer thread joined.", crawlerThread.getName());
       } catch (InterruptedException e) {
         LOGGER.warn("The producer thread joining has been interrupted", e);
-        threadPool.shutdownNow();
+        interrupt();
         Thread.currentThread().interrupt();
       }
 
       threadPool.shutdown();
-
       try {
         threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+        LOGGER.debug("Thread pool has terminated gracefully.");
       } catch (InterruptedException e) {
         LOGGER.warn("The thread pool joining has been interrupted", e);
+        interrupt();
         Thread.currentThread().interrupt();
       }
 
@@ -500,6 +486,9 @@ public void close() throws Exception {
             cachedException = e;
           }
         }
+        if (Thread.currentThread().isInterrupted()) {
+          interrupt();
+        }
       }
 
       if (!fatalHandlerExceptions.isEmpty()) {
@@ -517,6 +506,10 @@ public void close() throws Exception {
         throw mainHandlerException;
       }
 
+      if (Thread.currentThread().isInterrupted()) {
+        Thread.currentThread().interrupt();
+      }
+
       if (cachedException != null) {
         throw cachedException;
       }

src/main/java/ai/preferred/venom/Interruptible.java

@@ -19,15 +19,16 @@
 /**
  * @author Ween Jiann Lee
  */
-public interface Interruptible extends AutoCloseable {
+public interface Interruptible {
 
   /**
-   * Interrupt a thread and then close it.
-   *
-   * @throws Exception Exception.
+   * Interrupt the underlying mechanisms of the class.
+   * <p>
+   * Please note that this {@code interrupt} method should be
+   * idempotent.  In other words, calling this {@code interrupt}
+   * method more than once should not have any side effect.
+   * </p>
    */
-  default void interruptAndClose() throws Exception {
-    close();
-  }
+  void interrupt();
 
 }

src/main/java/ai/preferred/venom/ThreadedWorkerManager.java

@@ -71,37 +71,29 @@ public final Worker getWorker() {
   }
 
   @Override
-  public final void interruptAndClose() {
-    if (executor == null) {
+  public final void interrupt() {
+    if (executor == null || executor.isTerminated()) {
       return;
     }
-    LOGGER.debug("Forcefully shutting down the worker manager");
+    LOGGER.debug("Forcefully shutting down the worker manager.");
     executor.shutdownNow();
-    try {
-      executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
-      LOGGER.debug("The worker manager has been terminated");
-    } catch (final InterruptedException e) {
-      LOGGER.warn("Closing has been interrupted", e);
-      Thread.currentThread().interrupt();
-    }
   }
 
   @Override
   public final void close() {
-    if (executor == null) {
+    if (executor == null || executor.isTerminated()) {
       return;
     }
-    LOGGER.debug("Shutting down the worker manager");
+    LOGGER.debug("Shutting down the worker manager.");
     executor.shutdown();
     try {
       if (executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS)) {
-        LOGGER.debug("The worker manager has been terminated");
+        LOGGER.debug("The worker manager has been terminated.");
       } else {
-        executor.shutdownNow();
+        interrupt();
       }
     } catch (final InterruptedException e) {
-      LOGGER.warn("Closing has been interrupted, forcefully shutting down", e);
-      executor.shutdownNow();
+      interrupt();
       Thread.currentThread().interrupt();
     }
   }
@@ -120,7 +112,7 @@ public final void executeBlockingIO(final @NotNull Runnable task) {
       final ManagedBlockerTask managedBlockerTask = new ManagedBlockerTask(task);
       try {
         ForkJoinPool.managedBlock(managedBlockerTask);
-      } catch (InterruptedException e) {
+      } catch (final InterruptedException e) {
         Thread.currentThread().interrupt();
         throw new AssertionError("Exception of unknown cause. Please verify codebase.", e);
       }

src/main/java/ai/preferred/venom/WorkerManager.java

@@ -21,7 +21,7 @@
 /**
  * @author Maksim Tkachenko
  */
-public interface WorkerManager extends Interruptible {
+public interface WorkerManager extends Interruptible, AutoCloseable {
 
   /**
    * Get the result collector in use.

src/main/java/ai/preferred/venom/fetcher/Fetcher.java

@@ -16,7 +16,6 @@
 
 package ai.preferred.venom.fetcher;
 
-import ai.preferred.venom.Interruptible;
 import ai.preferred.venom.request.Request;
 import ai.preferred.venom.response.Response;
 
@@ -33,7 +32,7 @@
  * @author Truong Quoc Tuan
  * @author Ween Jiann Lee
  */
-public interface Fetcher extends Interruptible {
+public interface Fetcher extends AutoCloseable {
 
   /**
    * Fetcher starter.