fixed code

This project contributed to the Java Design Patterns repository by addressing an issue related to eliminating busy-waiting loops. The code was refactored to improve performance, adhere to Object-Oriented (OO) principles, and enhance maintainability. Key fixes included replacing busy-waiting with efficient mechanisms like wait/notify and ScheduledExecutorService, improving thread safety and code clarity. Comprehensive unit tests were added to validate functionality under various scenarios, including concurrency and interruption handling. These changes ensure robust, modern implementations aligned with OOAD principles.

Author: s-pakinamkhaled

commander/src/main/java/com/iluwatar/commander/Retry.java renamed to commander/src/main/java/com/iluwatar/commander/HandleErrorIssue.java

@@ -23,50 +23,81 @@
  * THE SOFTWARE.
  */
 package com.iluwatar.commander;
-
 import java.security.SecureRandom;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Predicate;
 
+
+    
+          
+            
+    
+
+          
+          Expand Down
+          
+            
+    
+
+          
+          Expand Up
+    
+    @@ -59,6 +62,7 @@ public interface HandleErrorIssue<T> {
+  
 /**
  * Retry pattern.
  *
  * @param <T> is the type of object passed into HandleErrorIssue as a parameter.
  */
-
 public class Retry<T> {
-
   /**
    * Operation Interface will define method to be implemented.
    */
-
   public interface Operation {
     void operation(List<Exception> list) throws Exception;
   }
-
   /**
    * HandleErrorIssue defines how to handle errors.
    *
    * @param <T> is the type of object to be passed into the method as parameter.
    */
-
   public interface HandleErrorIssue<T> {
     void handleIssue(T obj, Exception e);
   }
 
   private static final SecureRandom RANDOM = new SecureRandom();
 
+  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
   private final Operation op;
   private final HandleErrorIssue<T> handleError;
   private final int maxAttempts;
+
+    
+          
+            
+    
+
+          
+          Expand Down
+          
+            
+    
+
+          
+          Expand Up
+    
+    @@ -86,26 +90,25 @@ public interface HandleErrorIssue<T> {
+  
   private final long maxDelay;
   private final AtomicInteger attempts;
   private final Predicate<Exception> test;
   private final List<Exception> errors;
-
   Retry(Operation op, HandleErrorIssue<T> handleError, int maxAttempts,
         long maxDelay, Predicate<Exception>... ignoreTests) {
     this.op = op;
@@ -77,7 +108,6 @@ public interface HandleErrorIssue<T> {
     this.test = Arrays.stream(ignoreTests).reduce(Predicate::or).orElse(e -> false);
     this.errors = new ArrayList<>();
   }
-
   /**
    * Performing the operation with retries.
    *
@@ -86,26 +116,25 @@ public interface HandleErrorIssue<T> {
    */
 
   public void perform(List<Exception> list, T obj) {
-    do {
+    scheduler.schedule(() -> {
       try {
         op.operation(list);
-        return;
-      } catch (Exception e) {
+      }catch (Exception e){
         this.errors.add(e);
         if (this.attempts.incrementAndGet() >= this.maxAttempts || !this.test.test(e)) {
           this.handleError.handleIssue(obj, e);
+          scheduler.shutdown();
           return; //return here... don't go further
         }
-        try {
-          long testDelay =
-              (long) Math.pow(2, this.attempts.intValue()) * 1000 + RANDOM.nextInt(1000);
-          long delay = Math.min(testDelay, this.maxDelay);
-          Thread.sleep(delay);
-        } catch (InterruptedException f) {
-          //ignore
-        }
+        perform(list, obj);
       }
-    } while (true);
+    }, calculateDelay(), TimeUnit.MILLISECONDS);
+  }
+
+  private long calculateDelay(){
+    long testDelay =
+        (long) Math.pow(2, this.attempts.intValue()) * 1000 + RANDOM.nextInt(1000);
+    return Math.min(testDelay, this.maxDelay);
   }
 
-}
+}

microservices-log-aggregation/src/main/java/com/iluwatar/logaggregation/LogAggregator.java

@@ -25,12 +25,22 @@
 package com.iluwatar.logaggregation;
 
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import lombok.extern.slf4j.Slf4j;
 
+    
+        
+          
+    
+
+        
+        Expand All
+    
+    @@ -45,7 +45,7 @@ public class LogAggregator {
+  
 /**
  * Responsible for collecting and buffering logs from different services.
  * Once the logs reach a certain threshold or after a certain time interval,
@@ -40,15 +50,31 @@
  */
 @Slf4j
 public class LogAggregator {
-
   private static final int BUFFER_THRESHOLD = 3;
   private final CentralLogStore centralLogStore;
   private final ConcurrentLinkedQueue<LogEntry> buffer = new ConcurrentLinkedQueue<>();
   private final LogLevel minLogLevel;
-  private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
   private final AtomicInteger logCount = new AtomicInteger(0);
 
   /**
+
+    
+          
+            
+    
+
+          
+          Expand Down
+          
+            
+    
+
+          
+          Expand Up
+    
+    @@ -90,8 +90,8 @@ public void collectLog(LogEntry logEntry) {
+  
    * constructor of LogAggregator.
    *
    * @param centralLogStore central log store implement
@@ -59,7 +85,6 @@ public LogAggregator(CentralLogStore centralLogStore, LogLevel minLogLevel) {
     this.minLogLevel = minLogLevel;
     startBufferFlusher();
   }
-
   /**
    * Collects a given log entry, and filters it by the defined log level.
    *
@@ -70,33 +95,39 @@ public void collectLog(LogEntry logEntry) {
       LOGGER.warn("Log level or threshold level is null. Skipping.");
       return;
     }
-
     if (logEntry.getLevel().compareTo(minLogLevel) < 0) {
       LOGGER.debug("Log level below threshold. Skipping.");
       return;
     }
-
     buffer.offer(logEntry);
-
     if (logCount.incrementAndGet() >= BUFFER_THRESHOLD) {
       flushBuffer();
     }
   }
-
   /**
    * Stops the log aggregator service and flushes any remaining logs to
    * the central log store.
    *
    * @throws InterruptedException If any thread has interrupted the current thread.
    */
   public void stop() throws InterruptedException {
-    executorService.shutdownNow();
-    if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
+    scheduler.shutdownNow();
+    if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
       LOGGER.error("Log aggregator did not terminate.");
     }
     flushBuffer();
-  }
 
+    
+        
+          
+    
+
+        
+        Expand All
+    
+    @@ -106,15 +106,7 @@ private void flushBuffer() {
+  
+  }
   private void flushBuffer() {
     LogEntry logEntry;
     while ((logEntry = buffer.poll()) != null) {
@@ -106,15 +137,7 @@ private void flushBuffer() {
   }
 
   private void startBufferFlusher() {
-    executorService.execute(() -> {
-      while (!Thread.currentThread().isInterrupted()) {
-        try {
-          Thread.sleep(5000); // Flush every 5 seconds.
-          flushBuffer();
-        } catch (InterruptedException e) {
-          Thread.currentThread().interrupt();
-        }
-      }
-    });
+    //flush every 5 seconds
+    scheduler.scheduleWithFixedDelay(this::flushBuffer, 0, 5000, TimeUnit.MILLISECONDS);
   }
-}
+}

queue-based-load-leveling/src/main/java/com/iluwatar/queue/load/leveling/App.java

@@ -26,10 +26,26 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 
 /**
+
+    
+          
+            
+    
+
+          
+          Expand Down
+          
+            
+    
+
+          
+          Expand Up
+    
+    @@ -104,12 +103,7 @@ public static void main(String[] args) {
+  
  * Many solutions in the cloud involve running tasks that invoke services. In this environment, if a
  * service is subjected to intermittent heavy loads, it can cause performance or reliability
  * issues.
@@ -60,58 +76,54 @@
  */
 @Slf4j
 public class App {
-
   //Executor shut down time limit.
   private static final int SHUTDOWN_TIME = 15;
-
   /**
    * Program entry point.
    *
    * @param args command line args
    */
   public static void main(String[] args) {
-
     // An Executor that provides methods to manage termination and methods that can 
     // produce a Future for tracking progress of one or more asynchronous tasks.
     ExecutorService executor = null;
-
     try {
       // Create a MessageQueue object.
       var msgQueue = new MessageQueue();
-
       LOGGER.info("Submitting TaskGenerators and ServiceExecutor threads.");
-
       // Create three TaskGenerator threads. Each of them will submit different number of jobs.
       final var taskRunnable1 = new TaskGenerator(msgQueue, 5);
       final var taskRunnable2 = new TaskGenerator(msgQueue, 1);
       final var taskRunnable3 = new TaskGenerator(msgQueue, 2);
-
       // Create e service which should process the submitted jobs.
       final var srvRunnable = new ServiceExecutor(msgQueue);
-
       // Create a ThreadPool of 2 threads and
       // submit all Runnable task for execution to executor
       executor = Executors.newFixedThreadPool(2);
       executor.submit(taskRunnable1);
       executor.submit(taskRunnable2);
       executor.submit(taskRunnable3);
-
       // submitting serviceExecutor thread to the Executor service.
       executor.submit(srvRunnable);
-
       // Initiates an orderly shutdown.
       LOGGER.info("Initiating shutdown."
           + " Executor will shutdown only after all the Threads are completed.");
       executor.shutdown();
 
-      // Wait for SHUTDOWN_TIME seconds for all the threads to complete 
-      // their tasks and then shut down the executor and then exit. 
-      if (!executor.awaitTermination(SHUTDOWN_TIME, TimeUnit.SECONDS)) {
-        LOGGER.info("Executor was shut down and Exiting.");
-        executor.shutdownNow();
-      }
+      srvRunnable.shutdown(SHUTDOWN_TIME);
     } catch (Exception e) {
       LOGGER.error(e.getMessage());
     }
+
+    
+          
+            
+    
+
+          
+          Expand Down
+    
+    
+  
   }
 }