busy-waiting loops replaced with event driven programming

AnmolxSingh · AnmolxSingh · commit 0a22629d3eb7 · 2024-06-05T01:48:31.000+05:30
diff --git a/queue-based-load-leveling/src/main/java/com/iluwatar/queue/load/leveling/ServiceExecutor.java b/queue-based-load-leveling/src/main/java/com/iluwatar/queue/load/leveling/ServiceExecutor.java
@@ -26,16 +26,20 @@
 
 import lombok.extern.slf4j.Slf4j;
 
+import java.util.concurrent.BlockingQueue;
+
 /**
- * ServiceExecuotr class. This class will pick up Messages one by one from the Blocking Queue and
+ * ServiceExecuotr class. This class will pick up Messages one by one from the
+ * Blocking Queue and
  * process them.
  */
 @Slf4j
+
 public class ServiceExecutor implements Runnable {
 
-  private final MessageQueue msgQueue;
+  private final BlockingQueue<Message> msgQueue;
 
-  public ServiceExecutor(MessageQueue msgQueue) {
+  public ServiceExecutor(BlockingQueue<Message> msgQueue) {
     this.msgQueue = msgQueue;
   }
 
@@ -44,19 +48,14 @@ public ServiceExecutor(MessageQueue msgQueue) {
    */
   public void run() {
     try {
-      while (!Thread.currentThread().isInterrupted()) {
-        var msg = msgQueue.retrieveMsg();
-
-        if (null != msg) {
-          LOGGER.info(msg + " is served.");
-        } else {
-          LOGGER.info("Service Executor: Waiting for Messages to serve .. ");
-        }
+      while (true) {
+        Message msg = msgQueue.take(); // This will block until a message is available
 
-        Thread.sleep(1000);
+        LOGGER.info(msg + " is served.");
       }
-    } catch (Exception e) {
-      LOGGER.error(e.getMessage());
+    } catch (InterruptedException e) {
+      LOGGER.error("ServiceExecutor interrupted", e);
+      Thread.currentThread().interrupt();
     }
   }
 }
diff --git a/retry/src/main/java/com/iluwatar/retry/BusinessException.java b/retry/src/main/java/com/iluwatar/retry/BusinessException.java
@@ -27,9 +27,12 @@
 import java.io.Serial;
 
 /**
- * The top-most type in our exception hierarchy that signifies that an unexpected error condition
- * occurred. Its use is reserved as a "catch-all" for cases where no other subtype captures the
- * specificity of the error condition in question. Calling code is not expected to be able to handle
+ * The top-most type in our exception hierarchy that signifies that an
+ * unexpected error condition
+ * occurred. Its use is reserved as a "catch-all" for cases where no other
+ * subtype captures the
+ * specificity of the error condition in question. Calling code is not expected
+ * to be able to handle
  * this error and should be reported to the maintainers immediately.
  *
  */
@@ -38,11 +41,19 @@ public class BusinessException extends Exception {
   private static final long serialVersionUID = 6235833142062144336L;
 
   /**
-   * Ctor.
+   * Ctor .
    *
    * @param message the error message
    */
   public BusinessException(String message) {
     super(message);
   }
+
+  public BusinessException(String message, Throwable cause) {
+    super(message, cause);
+  }
+
+  public BusinessException(Throwable cause) {
+    super(cause);
+  }
 }
diff --git a/retry/src/main/java/com/iluwatar/retry/Retry.java b/retry/src/main/java/com/iluwatar/retry/Retry.java
@@ -28,11 +28,13 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
 /**
- * Decorates {@link BusinessOperation business operation} with "retry" capabilities.
+ * Decorates {@link BusinessOperation business operation} with "retry"
+ * capabilities.
  *
  * @param <T> the remote op's return type
  */
@@ -43,23 +45,24 @@ public final class Retry<T> implements BusinessOperation<T> {
   private final AtomicInteger attempts;
   private final Predicate<Exception> test;
   private final List<Exception> errors;
+  private int attemptsCount;
 
   /**
    * Ctor.
    *
    * @param op          the {@link BusinessOperation} to retry
    * @param maxAttempts number of times to retry
    * @param delay       delay (in milliseconds) between attempts
-   * @param ignoreTests tests to check whether the remote exception can be ignored. No exceptions
+   * @param ignoreTests tests to check whether the remote exception can be
+   *                    ignored. No exceptions
    *                    will be ignored if no tests are given
    */
   @SafeVarargs
   public Retry(
       BusinessOperation<T> op,
       int maxAttempts,
       long delay,
-      Predicate<Exception>... ignoreTests
-  ) {
+      Predicate<Exception>... ignoreTests) {
     this.op = op;
     this.maxAttempts = maxAttempts;
     this.delay = delay;
@@ -88,22 +91,42 @@ public int attempts() {
 
   @Override
   public T perform() throws BusinessException {
-    do {
+    CompletableFuture<T> future = new CompletableFuture<>();
+    retryAttempt(future, 1);
+    try {
+      return future.get();
+    } catch (Exception e) {
+      throw new BusinessException("Max retry attempts exceeded.", e);
+    }
+  }
+
+  private void retryAttempt(CompletableFuture<T> future, int attempt) {
+    if (attempt > maxAttempts) {
+      future.completeExceptionally(new BusinessException("Max retry attempts exceeded."));
+      return;
+    }
+
+    CompletableFuture.runAsync(() -> {
       try {
-        return this.op.perform();
+        T result = op.perform();
+        future.complete(result);
       } catch (BusinessException e) {
-        this.errors.add(e);
-
-        if (this.attempts.incrementAndGet() >= this.maxAttempts || !this.test.test(e)) {
-          throw e;
+        errors.add(e);
+        attemptsCount++;
+        if (test.test(e)) {
+          retryAttempt(future, attempt + 1);
+        } else {
+          future.completeExceptionally(e);
         }
+      }
+    });
 
-        try {
-          Thread.sleep(this.delay);
-        } catch (InterruptedException f) {
-          //ignore
-        }
+    if (attempt < maxAttempts) {
+      try {
+        Thread.sleep(delay);
+      } catch (InterruptedException ignored) {
+        Thread.currentThread().interrupt();
       }
-    } while (true);
+    }
   }
 }
diff --git a/retry/src/main/java/com/iluwatar/retry/RetryExponentialBackoff.java b/retry/src/main/java/com/iluwatar/retry/RetryExponentialBackoff.java
@@ -29,11 +29,13 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
 /**
- * Decorates {@link BusinessOperation business operation} with "retry" capabilities.
+ * Decorates {@link BusinessOperation business operation} with "retry"
+ * capabilities using exponential backoff strategy.
  *
  * @param <T> the remote op's return type
  */
@@ -42,25 +44,23 @@ public final class RetryExponentialBackoff<T> implements BusinessOperation<T> {
   private final BusinessOperation<T> op;
   private final int maxAttempts;
   private final long maxDelay;
-  private final AtomicInteger attempts;
   private final Predicate<Exception> test;
   private final List<Exception> errors;
+  private final AtomicInteger attempts;
 
   /**
-   * Ctor.
+   * Constructor.
    *
    * @param op          the {@link BusinessOperation} to retry
    * @param maxAttempts number of times to retry
-   * @param ignoreTests tests to check whether the remote exception can be ignored. No exceptions
+   * @param maxDelay    the maximum delay between retries (in milliseconds)
+   * @param ignoreTests tests to check whether the remote exception can be
+   *                    ignored. No exceptions
    *                    will be ignored if no tests are given
    */
   @SafeVarargs
-  public RetryExponentialBackoff(
-      BusinessOperation<T> op,
-      int maxAttempts,
-      long maxDelay,
-      Predicate<Exception>... ignoreTests
-  ) {
+  public RetryExponentialBackoff(BusinessOperation<T> op, int maxAttempts, long maxDelay,
+      Predicate<Exception>... ignoreTests) {
     this.op = op;
     this.maxAttempts = maxAttempts;
     this.maxDelay = maxDelay;
@@ -70,7 +70,7 @@ public RetryExponentialBackoff(
   }
 
   /**
-   * The errors encountered while retrying, in the encounter order.
+   * The errors encountered while retrying, in the order of occurrence.
    *
    * @return the errors encountered while retrying
    */
@@ -89,24 +89,59 @@ public int attempts() {
 
   @Override
   public T perform() throws BusinessException {
-    do {
-      try {
-        return this.op.perform();
-      } catch (BusinessException e) {
-        this.errors.add(e);
+    CompletableFuture<T> future = new CompletableFuture<>();
+    retryAttempt(future, 1);
+    try {
+      return future.get();
+    } catch (Exception e) {
+      throw new BusinessException("Max retry attempts exceeded.", e);
+    }
+  }
 
-        if (this.attempts.incrementAndGet() >= this.maxAttempts || !this.test.test(e)) {
-          throw e;
-        }
+  /**
+   * Recursive method to perform retry attempts asynchronously using
+   * CompletableFuture.
+   *
+   * @param future  CompletableFuture to handle the result of the retry attempts
+   * @param attempt current attempt number
+   */
+  private void retryAttempt(CompletableFuture<T> future, int attempt) {
+    if (attempt > maxAttempts) {
+      future.completeExceptionally(new BusinessException("Max retry attempts exceeded."));
+      return;
+    }
 
-        try {
-          var testDelay = (long) Math.pow(2, this.attempts()) * 1000 + RANDOM.nextInt(1000);
-          var delay = Math.min(testDelay, this.maxDelay);
-          Thread.sleep(delay);
-        } catch (InterruptedException f) {
-          //ignore
+    CompletableFuture.runAsync(() -> {
+      try {
+        T result = op.perform();
+        future.complete(result);
+      } catch (BusinessException e) {
+        errors.add(e);
+        attempts.incrementAndGet();
+        if (attempts.get() >= maxAttempts || !test.test(e)) {
+          future.completeExceptionally(e);
+        } else {
+          long delay = calculateDelay(attempts.get());
+          try {
+            Thread.sleep(delay);
+          } catch (InterruptedException ignored) {
+            Thread.currentThread().interrupt();
+          }
+          retryAttempt(future, attempt + 1);
         }
       }
-    } while (true);
+    });
+  }
+
+  /**
+   * Calculates the delay for exponential backoff based on the current attempt
+   * number.
+   *
+   * @param attempt the current attempt number
+   * @return the calculated delay (in milliseconds)
+   */
+  private long calculateDelay(int attempt) {
+    long testDelay = (long) Math.pow(2, attempt) * 1000 + RANDOM.nextInt(1000);
+    return Math.min(testDelay, maxDelay);
   }
 }
diff --git a/twin/src/main/java/com/iluwatar/twin/BallThread.java b/twin/src/main/java/com/iluwatar/twin/BallThread.java
