updated

Author: Mostafa-Hisham0

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

@@ -1,34 +1,59 @@
+/*
+ * This project is licensed under the MIT license. Module model-view-viewmodel is using ZK framework licensed under LGPL (see lgpl-3.0.txt).
+ *
+ * The MIT License
+ * Copyright © 2014-2022 Ilkka Seppälä
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
 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;
 
 /**
- * Collects and buffers logs from different services, periodically flushing them
- * to a central log store based on a time interval or buffer threshold.
+ * Responsible for collecting and buffering logs from different services.
+ * Once the logs reach a certain threshold or after a certain time interval,
+ * they are flushed to the central log store. This class ensures logs are collected
+ * and processed asynchronously and efficiently, providing both an immediate collection
+ * and periodic flushing.
  */
 @Slf4j
 public class LogAggregator {
 
   private static final int BUFFER_THRESHOLD = 3;
-  private static final int FLUSH_INTERVAL = 5; // Interval in seconds for periodic flushing
-
   private final CentralLogStore centralLogStore;
   private final ConcurrentLinkedQueue<LogEntry> buffer = new ConcurrentLinkedQueue<>();
   private final LogLevel minLogLevel;
+  private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+  private final Object bufferWait = new Object();
   private final AtomicInteger logCount = new AtomicInteger(0);
 
-  private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
-
   /**
-   * Constructor of LogAggregator.
+   * constructor of LogAggregator.
    *
-   * @param centralLogStore central log store implementation
-   * @param minLogLevel     minimum log level to store log
+   * @param centralLogStore central log store implement
+   * @param minLogLevel min log level to store log
    */
   public LogAggregator(CentralLogStore centralLogStore, LogLevel minLogLevel) {
     this.centralLogStore = centralLogStore;
@@ -37,8 +62,7 @@ public LogAggregator(CentralLogStore centralLogStore, LogLevel minLogLevel) {
   }
 
   /**
-   * Collects a log entry and buffers it for eventual flushing to the central log store.
-   * Filters logs based on the configured minimum log level.
+   * Collects a given log entry, and filters it by the defined log level.
    *
    * @param logEntry The log entry to collect.
    */
@@ -54,6 +78,7 @@ public void collectLog(LogEntry logEntry) {
     }
 
     buffer.offer(logEntry);
+    bufferWake();
 
     if (logCount.incrementAndGet() >= BUFFER_THRESHOLD) {
       flushBuffer();
@@ -64,41 +89,48 @@ public void collectLog(LogEntry logEntry) {
    * Stops the log aggregator service and flushes any remaining logs to
    * the central log store.
    *
-   * @throws InterruptedException If interrupted while shutting down.
+   * @throws InterruptedException If any thread has interrupted the current thread.
    */
   public void stop() throws InterruptedException {
-    LOGGER.info("Stopping log aggregator...");
-    scheduler.shutdown();
-    if (!scheduler.awaitTermination(10, TimeUnit.SECONDS)) {
-      LOGGER.error("Log aggregator did not terminate cleanly.");
+    executorService.shutdownNow();
+    if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
+      LOGGER.error("Log aggregator did not terminate.");
     }
     flushBuffer();
   }
 
-  /**
-   * Flushes the buffered logs to the central log store.
-   */
   private void flushBuffer() {
-    LOGGER.info("Flushing buffer...");
     LogEntry logEntry;
     while ((logEntry = buffer.poll()) != null) {
       centralLogStore.storeLog(logEntry);
       logCount.decrementAndGet();
     }
-    LOGGER.info("Buffer flushed.");
   }
 
-  /**
-   * Starts the periodic buffer flusher task using a scheduled executor service.
-   */
   private void startBufferFlusher() {
-    scheduler.scheduleAtFixedRate(() -> {
-      try {
-        LOGGER.info("Periodic buffer flush initiated...");
-        flushBuffer();
-      } catch (Exception e) {
-        LOGGER.error("Error during buffer flush", e);
+    executorService.execute(() -> {
+      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();
+        }
       }
-    }, FLUSH_INTERVAL, FLUSH_INTERVAL, TimeUnit.SECONDS);
+    });
+  }
+
+  /**
+   * Wakes up buffer.
+   */
+  public void bufferWake() {
+    synchronized (bufferWait) {
+      bufferWait.notifyAll();
+    }
   }
 }

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

@@ -36,6 +36,7 @@
 public class MessageQueue {
 
   private final BlockingQueue<Message> blkQueue;
+  public final Object serviceExecutorWait = new Object();
 
   // Default constructor when called creates Blocking Queue object. 
   public MessageQueue() {
@@ -50,6 +51,9 @@ public void submitMsg(Message msg) {
     try {
       if (null != msg) {
         blkQueue.add(msg);
+        synchronized (serviceExecutorWait) {
+          serviceExecutorWait.notifyAll();
+        }
       }
     } catch (Exception e) {
       LOGGER.error(e.getMessage());

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

@@ -24,32 +24,38 @@
  */
 package com.iluwatar.queue.load.leveling;
 
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
 
+/**
+ * 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 BlockingQueue<String> msgQueue;
-
-    public ServiceExecutor(BlockingQueue<String> msgQueue) {
-        this.msgQueue = msgQueue;
-    }
+  private final MessageQueue msgQueue;
+  public ServiceExecutor(MessageQueue msgQueue) {
+    this.msgQueue = msgQueue;
+  }
 
-    @Override
-    public void run() {
-        try {
-            while (!Thread.currentThread().isInterrupted()) {
-                String msg = msgQueue.poll(1, TimeUnit.SECONDS);  // Wait for message with timeout
+  /**
+   * The ServiceExecutor thread will retrieve each message and process it.
+   */
+  public void run() {
+    try {
+      while (!Thread.currentThread().isInterrupted()) {
+        var msg = msgQueue.retrieveMsg();
 
-                if (msg != null) {
-                    LOGGER.info(msg + " is served.");
-                } else {
-                    LOGGER.info("Service Executor: Waiting for Messages to serve...");
-                }
-            }
-        } catch (InterruptedException e) {
-            LOGGER.error("Service Executor interrupted: " + e.getMessage());
+        if (null != msg) {
+          LOGGER.info(msg + " is served.");
+        } else {
+          LOGGER.info("Service Executor: Waiting for Messages to serve .. ");
+          synchronized (msgQueue.serviceExecutorWait) {
+            msgQueue.serviceExecutorWait.wait();
+          }
         }
+      }
+    } catch (Exception e) {
+      LOGGER.error(e.getMessage());
     }
+  }
 }

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

@@ -66,9 +66,8 @@ public void run() {
     try {
       while (count > 0) {
         var statusMsg = "Message-" + count + " submitted by " + Thread.currentThread().getName();
-        this.submit(new Message(statusMsg));
-
         LOGGER.info(statusMsg);
+        this.submit(new Message(statusMsg));
 
         // reduce the message count.
         count--;

queue-based-load-leveling/src/test/java/com/iluwatar/queue/load/leveling/TaskGenSrvExeTest.java

@@ -24,14 +24,19 @@
  */
 package com.iluwatar.queue.load.leveling;
 
+import static java.util.concurrent.CompletableFuture.anyOf;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
 
 /**
  * Test case for submitting Message to Blocking Queue by TaskGenerator and retrieve the message by
  * ServiceExecutor.
  */
+@Slf4j
 class TaskGenSrvExeTest {
 
   @Test
@@ -53,4 +58,37 @@ void taskGeneratorTest() {
     assertNotNull(srvExeThr);
   }
 
+  /**
+   * Tests that service executor waits at start since no message is sent to execute upon.
+   * @throws InterruptedException
+   */
+  @Test
+  void serviceExecutorStartStateTest() throws InterruptedException {
+    var msgQueue = new MessageQueue();
+    var srvRunnable = new ServiceExecutor(msgQueue);
+    var srvExeThr = new Thread(srvRunnable);
+    srvExeThr.start();
+    Thread.sleep(200); // sleep a little until service executor thread waits
+    LOGGER.info("Current Service Executor State: " + srvExeThr.getState());
+    assertEquals(srvExeThr.getState(), Thread.State.WAITING);
+
+  }
+
+  @Test
+  void serviceExecutorWakeStateTest() throws InterruptedException {
+    var msgQueue = new MessageQueue();
+    var srvRunnable = new ServiceExecutor(msgQueue);
+    var srvExeThr = new Thread(srvRunnable);
+    srvExeThr.start();
+    Thread.sleep(200); // sleep a little until service executor thread waits
+    synchronized (msgQueue.serviceExecutorWait){
+      msgQueue.serviceExecutorWait.notifyAll();
+    }
+    var srvExeState = srvExeThr.getState();
+    LOGGER.info("Current Service Executor State: " + srvExeState);
+    // assert that state changes from waiting
+    assertTrue(srvExeState != Thread.State.WAITING);
+
+  }
+
 }