implemented review changes

Author: alhusseain

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

@@ -27,6 +27,7 @@
 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;
@@ -108,21 +109,21 @@ private void flushBuffer() {
   }
 
   private void startBufferFlusher() {
-    executorService.execute(() -> {
+    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    scheduler.scheduleAtFixedRate(() -> {
       while (!Thread.currentThread().isInterrupted()) {
         try {
           synchronized (bufferWait) {
             if (buffer.isEmpty()) {
               bufferWait.wait();
             }
           }
-          Thread.sleep(5000); // Flush every 5 seconds.
           flushBuffer();
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
         }
       }
-    });
+    }, 5, 5, TimeUnit.SECONDS);
   }
 
   /**
@@ -133,4 +134,4 @@ public void bufferWake() {
       bufferWait.notifyAll();
     }
   }
-}
+}

microservices-log-aggregation/src/test/java/com/iluwatar/logaggregation/LogAggregatorTest.java

@@ -24,13 +24,16 @@
  */
 package com.iluwatar.logaggregation;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 
 import java.time.LocalDateTime;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Order;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;

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

@@ -26,6 +26,7 @@
 
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 
@@ -88,7 +89,8 @@ public static void main(String[] args) {
 
       // Create e service which should process the submitted jobs.
       final var srvRunnable = new ServiceExecutor(msgQueue);
-
+      ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+      scheduler.scheduleAtFixedRate(srvRunnable, 1, 1, TimeUnit.SECONDS);
       // Create a ThreadPool of 2 threads and
       // submit all Runnable task for execution to executor
       executor = Executors.newFixedThreadPool(2);

server-session/src/main/java/com/iluwatar/sessionserver/App.java

@@ -31,6 +31,9 @@
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 
 
@@ -59,8 +62,7 @@ public class App {
   private static Map<String, Integer> sessions = new HashMap<>();
   private static Map<String, Instant> sessionCreationTimes = new HashMap<>();
   private static final long SESSION_EXPIRATION_TIME = 10000;
-  private static Object sessionExpirationWait = new Object(); // used to make expiration task wait or work based on event (login request sent or not)
-  private static Thread sessionExpirationThread;
+
 
   /**
    * Main entry point.
@@ -79,60 +81,31 @@ public static void main(String[] args) throws IOException {
     server.start();
 
     // Start background task to check for expired sessions
-    sessionExpirationTask();
 
     LOGGER.info("Server started. Listening on port 8080...");
+
+    ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+    scheduler.scheduleAtFixedRate(App::sessionExpirationTask, SESSION_EXPIRATION_TIME, SESSION_EXPIRATION_TIME, TimeUnit.MILLISECONDS);
   }
 
   private static void sessionExpirationTask() {
-    sessionExpirationThread = new Thread(() -> {
-      while (true) {
-        try {
-          synchronized (sessions) {
-            if (sessions.isEmpty()) {
-              synchronized (sessionExpirationWait) {
-                sessionExpirationWait.wait();
-              }
-            }
-          }
-          LOGGER.info("Session expiration checker started...");
-          Thread.sleep(SESSION_EXPIRATION_TIME); // Sleep for expiration time
-          Instant currentTime = Instant.now();
-          synchronized (sessions) {
-            synchronized (sessionCreationTimes) {
-              Iterator<Map.Entry<String, Instant>> iterator =
-                  sessionCreationTimes.entrySet().iterator();
-              while (iterator.hasNext()) {
-                Map.Entry<String, Instant> entry = iterator.next();
-                if (entry.getValue().plusMillis(SESSION_EXPIRATION_TIME).isBefore(currentTime)) {
-                  LOGGER.info("User " + entry.getValue() + " removed");
-                  sessions.remove(entry.getKey());
-                  iterator.remove();
-                }
-              }
-            }
+    LOGGER.info("Session expiration checker started...");
+    Instant currentTime = Instant.now();
+    synchronized (sessions) {
+      synchronized (sessionCreationTimes) {
+        Iterator<Map.Entry<String, Instant>> iterator =
+            sessionCreationTimes.entrySet().iterator();
+        while (iterator.hasNext()) {
+          Map.Entry<String, Instant> entry = iterator.next();
+          if (entry.getValue().plusMillis(SESSION_EXPIRATION_TIME).isBefore(currentTime)) {
+            LOGGER.info("User " + entry.getValue() + " removed");
+            sessions.remove(entry.getKey());
+            iterator.remove();
           }
-          LOGGER.info("Session expiration checker finished!");
-        } catch (InterruptedException e) {
-          LOGGER.error("An error occurred: ", e);
-          Thread.currentThread().interrupt();
         }
       }
-    });
-    sessionExpirationThread.start();
-  }
-
-  /**
-   * allows sessionExpirationTask to run again, called when a login request is sent.
-   */
-  public static void expirationTaskWake() {
-    synchronized (sessionExpirationWait) {
-      sessionExpirationWait.notifyAll();
     }
-  }
-
-  public static Thread.State getExpirationTaskState() {
-    return sessionExpirationThread.getState();
+    LOGGER.info("Session expiration checker finished!");
   }
 
 }

server-session/src/main/java/com/iluwatar/sessionserver/LoginHandler.java

@@ -63,7 +63,6 @@ public void handle(HttpExchange exchange) {
 
     // Set session ID as cookie
     exchange.getResponseHeaders().add("Set-Cookie", "sessionID=" + sessionId);
-    App.expirationTaskWake(); // Wake up expiration task
     // Send response
     String response = "Login successful!\n" + "Session ID: " + sessionId;
     try {

server-session/src/test/java/com.iluwatar.sessionserver/AppTest.java

@@ -57,7 +57,6 @@ public void run() {
         } else {
           twin.draw();
           twin.move();
-          Thread.sleep(250);
         }
       }
     } catch (InterruptedException e) {