multithreaded logger

kousiknath · kousiknath · commit df79aee3e1ba · 2023-09-07T00:01:10.000+05:30
diff --git a/src/com/lld/kafka/README.md b/src/com/lld/kafka/README.md
@@ -1,4 +1,4 @@
-<b> Design a message broker which has the following properties: </b>
+<b> Design a applicationAwareMessage broker which has the following properties: </b>
 
 1. Producers should be able to produce messages.
 2. Producers can be divided into multiple partitions.
diff --git a/src/com/lld/multithreadedlogging/MainApplication.java b/src/com/lld/multithreadedlogging/MainApplication.java
@@ -0,0 +1,24 @@
+package com.lld.multithreadedlogging;
+
+import com.lld.multithreadedlogging.service.QueueService;
+import com.lld.multithreadedlogging.service.impl.QueueServiceImpl;
+import com.lld.multithreadedlogging.worker.Consumer;
+import com.lld.multithreadedlogging.worker.Producer;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class MainApplication {
+    private static void test1() {
+        QueueService queueService = new QueueServiceImpl(100);
+        ExecutorService executorService = Executors.newFixedThreadPool(3);
+        executorService.execute(new Producer(queueService));
+        executorService.execute(new Producer(queueService));
+        executorService.execute(new Consumer(queueService));
+        executorService.shutdown();
+    }
+
+    public static void main(String[] args) {
+        test1();
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/README.md b/src/com/lld/multithreadedlogging/README.md
@@ -0,0 +1,9 @@
+Design a logging system:
+
+1. The system supports multiple log level like error, debug, warning, info etc.
+2. Multiple log mediums are supported like file based logs, network logs, db logs etc.
+3. Multiple producers can produce to the queue simultaneously.
+4. Irrespective of the rate of production, the system should consume messages as fast
+as possible from the queue and flush them to the log medium.
+5. A single applicationAwareMessage can get published to multiple medium.
+6. 
diff --git a/src/com/lld/multithreadedlogging/configuration/Config.java b/src/com/lld/multithreadedlogging/configuration/Config.java
@@ -0,0 +1,7 @@
+package com.lld.multithreadedlogging.configuration;
+
+public class Config {
+    public final static int DEFAULT_PUBLISH_TIMEOUT_MS = 100;
+    public final static int DEFAULT_CONSUMPTION_TIMEOUT_MS = 100;
+    public final static int DEFAULT_CAPACITY = 100;
+}
diff --git a/src/com/lld/multithreadedlogging/constant/LogLevel.java b/src/com/lld/multithreadedlogging/constant/LogLevel.java
@@ -0,0 +1,8 @@
+package com.lld.multithreadedlogging.constant;
+
+public enum LogLevel {
+    INFO,
+    DEBUG,
+    WARN,
+    ERROR
+}
diff --git a/src/com/lld/multithreadedlogging/constant/SupportedLogMedium.java b/src/com/lld/multithreadedlogging/constant/SupportedLogMedium.java
@@ -0,0 +1,7 @@
+package com.lld.multithreadedlogging.constant;
+
+public enum SupportedLogMedium {
+    FILE,
+    NETWORK,
+    DB
+}
diff --git a/src/com/lld/multithreadedlogging/logger/ApplicationAwareLogger.java b/src/com/lld/multithreadedlogging/logger/ApplicationAwareLogger.java
@@ -0,0 +1,10 @@
+package com.lld.multithreadedlogging.logger;
+
+import com.lld.multithreadedlogging.model.message.ApplicationAwareMessage;
+
+public interface ApplicationAwareLogger {
+    void info(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException;
+    void debug(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException;
+    void warn(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException;
+    void error(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException;
+}
diff --git a/src/com/lld/multithreadedlogging/logger/InternalLogger.java b/src/com/lld/multithreadedlogging/logger/InternalLogger.java
@@ -0,0 +1,8 @@
+package com.lld.multithreadedlogging.logger;
+
+import com.lld.multithreadedlogging.model.message.ApplicationAwareMessage;
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+
+public interface InternalLogger {
+    void log(WritableMessage message);
+}
diff --git a/src/com/lld/multithreadedlogging/logger/factory/InternalLoggerSimpleFactory.java b/src/com/lld/multithreadedlogging/logger/factory/InternalLoggerSimpleFactory.java
@@ -0,0 +1,22 @@
+package com.lld.multithreadedlogging.logger.factory;
+
+import com.lld.multithreadedlogging.constant.SupportedLogMedium;
+import com.lld.multithreadedlogging.logger.InternalLogger;
+import com.lld.multithreadedlogging.logger.impl.DBInternalLogger;
+import com.lld.multithreadedlogging.logger.impl.FileInternalLogger;
+import com.lld.multithreadedlogging.logger.impl.NetworkInternalLogger;
+
+public class InternalLoggerSimpleFactory {
+    public static InternalLogger getLoggerInstance(SupportedLogMedium medium) {
+        switch (medium) {
+            case DB:
+                return DBInternalLogger.getInstance();
+            case FILE:
+                return FileInternalLogger.getInstance();
+            case NETWORK:
+                return NetworkInternalLogger.getInstance();
+            default:
+                return null;
+        }
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/logger/impl/ApplicationAwareLoggerImpl.java b/src/com/lld/multithreadedlogging/logger/impl/ApplicationAwareLoggerImpl.java
@@ -0,0 +1,50 @@
+package com.lld.multithreadedlogging.logger.impl;
+
+import com.lld.multithreadedlogging.constant.LogLevel;
+import com.lld.multithreadedlogging.constant.SupportedLogMedium;
+import com.lld.multithreadedlogging.model.message.ApplicationAwareMessage;
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+import com.lld.multithreadedlogging.logger.ApplicationAwareLogger;
+import com.lld.multithreadedlogging.service.QueueService;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ *  Applications will talk to this logger instance to publish message to the system
+ */
+public class ApplicationAwareLoggerImpl implements ApplicationAwareLogger {
+    QueueService queueService;
+
+    public ApplicationAwareLoggerImpl(QueueService queueService) {
+        this.queueService = queueService;
+    }
+    @Override
+    public void info(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException {
+        this.queueService.publish(getWritableMessages(applicationAwareMessage, LogLevel.INFO));
+    }
+
+    @Override
+    public void debug(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException {
+        this.queueService.publish(getWritableMessages(applicationAwareMessage, LogLevel.DEBUG));
+    }
+
+    @Override
+    public void warn(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException {
+        this.queueService.publish(getWritableMessages(applicationAwareMessage, LogLevel.WARN));
+    }
+
+    @Override
+    public void error(ApplicationAwareMessage applicationAwareMessage) throws InterruptedException {
+        this.queueService.publish(getWritableMessages(applicationAwareMessage, LogLevel.ERROR));
+    }
+
+    private List<WritableMessage> getWritableMessages(ApplicationAwareMessage applicationAwareMessage, LogLevel logLevel) {
+        List<WritableMessage> writableMessages = new ArrayList<>();
+        for (SupportedLogMedium supportedLogMedium: applicationAwareMessage.getMediums()) {
+            writableMessages.add(new WritableMessage(applicationAwareMessage, logLevel, supportedLogMedium));
+        }
+
+        return writableMessages;
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/logger/impl/DBInternalLogger.java b/src/com/lld/multithreadedlogging/logger/impl/DBInternalLogger.java
@@ -0,0 +1,28 @@
+package com.lld.multithreadedlogging.logger.impl;
+
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+import com.lld.multithreadedlogging.logger.InternalLogger;
+
+public class DBInternalLogger implements InternalLogger {
+    private DBInternalLogger() {
+        // Initiate connection to DB driver here
+    }
+
+    private static DBInternalLogger instance;
+
+    public static DBInternalLogger getInstance() {
+        if (instance == null) {
+            synchronized (DBInternalLogger.class) {
+                if (instance == null) {
+                    instance = new DBInternalLogger();
+                }
+            }
+        }
+
+        return instance;
+    }
+    @Override
+    public void log(WritableMessage message) {
+        System.out.println(message);
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/logger/impl/FileInternalLogger.java b/src/com/lld/multithreadedlogging/logger/impl/FileInternalLogger.java
@@ -0,0 +1,29 @@
+package com.lld.multithreadedlogging.logger.impl;
+
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+import com.lld.multithreadedlogging.logger.InternalLogger;
+
+public class FileInternalLogger implements InternalLogger {
+    private FileInternalLogger() {
+
+    }
+
+    private static FileInternalLogger instance;
+
+    public static FileInternalLogger getInstance() {
+        if (instance == null) {
+            synchronized (FileInternalLogger.class) {
+                if (instance == null) {
+                    instance = new FileInternalLogger();
+                }
+            }
+        }
+
+        return instance;
+    }
+
+    @Override
+    public void log(WritableMessage message) {
+        System.out.println(message);
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/logger/impl/NetworkInternalLogger.java b/src/com/lld/multithreadedlogging/logger/impl/NetworkInternalLogger.java
@@ -0,0 +1,30 @@
+package com.lld.multithreadedlogging.logger.impl;
+
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+import com.lld.multithreadedlogging.logger.InternalLogger;
+
+public class NetworkInternalLogger implements InternalLogger {
+    private NetworkInternalLogger() {
+        // Initiate connection to network driver / http connection which will
+        // write to that particular logger
+    }
+
+    private static NetworkInternalLogger instance;
+
+    public static NetworkInternalLogger getInstance() {
+        if (instance == null) {
+            synchronized (NetworkInternalLogger.class) {
+                if (instance == null) {
+                    instance = new NetworkInternalLogger();
+                }
+            }
+        }
+
+        return instance;
+    }
+
+    @Override
+    public void log(WritableMessage message) {
+        System.out.println(message);
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/model/message/ApplicationAwareMessage.java b/src/com/lld/multithreadedlogging/model/message/ApplicationAwareMessage.java
@@ -0,0 +1,41 @@
+package com.lld.multithreadedlogging.model.message;
+
+import com.lld.multithreadedlogging.constant.LogLevel;
+import com.lld.multithreadedlogging.constant.SupportedLogMedium;
+
+import java.time.Instant;
+import java.util.List;
+
+/**
+ * ApplicationAwareMessage is the message produced by the application.
+ */
+
+public class ApplicationAwareMessage {
+    private String msg;
+    private Instant time;
+    private List<SupportedLogMedium> mediums;
+    private Thread thread;
+
+    public ApplicationAwareMessage(String msg, List<SupportedLogMedium> mediums, Thread thread) {
+        this.msg = msg;
+        this.mediums = mediums;
+        this.time = Instant.now();
+        this.thread = thread;
+    }
+
+    public String getMsg() {
+        return msg;
+    }
+
+    public Instant getTime() {
+        return time;
+    }
+
+    public List<SupportedLogMedium> getMediums() {
+        return mediums;
+    }
+
+    public Thread getThread() {
+        return thread;
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/model/message/WritableMessage.java b/src/com/lld/multithreadedlogging/model/message/WritableMessage.java
@@ -0,0 +1,34 @@
+package com.lld.multithreadedlogging.model.message;
+
+import com.lld.multithreadedlogging.constant.LogLevel;
+import com.lld.multithreadedlogging.constant.SupportedLogMedium;
+
+public class WritableMessage {
+    /*
+        WritableMessage wraps an instance of ApplicationAwareMessage so that we can log extra
+        metadata if required while actually logging the application message.
+     */
+    private ApplicationAwareMessage applicationAwareMessage;
+    private LogLevel logLevel;
+    private SupportedLogMedium medium;
+    private Thread sourceThread;
+
+    public WritableMessage(ApplicationAwareMessage applicationAwareMessage, LogLevel logLevel, SupportedLogMedium medium) {
+        this.applicationAwareMessage = applicationAwareMessage;
+        this.logLevel = logLevel;
+        this.medium = medium;
+        this.sourceThread = applicationAwareMessage.getThread();
+    }
+
+    public SupportedLogMedium getMedium() {
+        return this.medium;
+    }
+
+    @Override
+    public String toString() {
+        return "Log Level = " + logLevel
+                + " Thread - " + sourceThread.getName()
+                + " timestamp = " + applicationAwareMessage.getTime().toString()
+                + " message = " + applicationAwareMessage.getMsg();
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/service/QueueService.java b/src/com/lld/multithreadedlogging/service/QueueService.java
@@ -0,0 +1,10 @@
+package com.lld.multithreadedlogging.service;
+
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+
+import java.util.List;
+
+public interface QueueService {
+    void publish(List<WritableMessage> message) throws InterruptedException;
+    WritableMessage consume() throws InterruptedException;
+}
diff --git a/src/com/lld/multithreadedlogging/service/impl/QueueServiceImpl.java b/src/com/lld/multithreadedlogging/service/impl/QueueServiceImpl.java
@@ -0,0 +1,34 @@
+package com.lld.multithreadedlogging.service.impl;
+
+import com.lld.multithreadedlogging.configuration.Config;
+import com.lld.multithreadedlogging.model.message.WritableMessage;
+import com.lld.multithreadedlogging.service.QueueService;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+public class QueueServiceImpl implements QueueService {
+    private ArrayBlockingQueue<WritableMessage> blockingQueue;
+    private int capacity;
+
+    public QueueServiceImpl(int cap) {
+        this.capacity = Math.max(cap, Config.DEFAULT_CAPACITY);
+        this.blockingQueue = new ArrayBlockingQueue<>(this.capacity);
+    }
+
+    @Override
+    public void publish(List<WritableMessage> messages) throws InterruptedException {
+        Objects.requireNonNull(messages);
+
+        for (WritableMessage message: messages) {
+            this.blockingQueue.offer(message, Config.DEFAULT_PUBLISH_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+        }
+    }
+
+    @Override
+    public WritableMessage consume() throws InterruptedException {
+        return this.blockingQueue.poll(Config.DEFAULT_CONSUMPTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
+    }
+}
diff --git a/src/com/lld/multithreadedlogging/worker/Consumer.java b/src/com/lld/multithreadedlogging/worker/Consumer.java
diff --git a/src/com/lld/multithreadedlogging/worker/Producer.java b/src/com/lld/multithreadedlogging/worker/Producer.java

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-<b> Design a message broker which has the following properties: </b>`
	`1`	`+<b> Design a applicationAwareMessage broker which has the following properties: </b>`
`2`	`2`
`3`	`3`	`1. Producers should be able to produce messages.`
`4`	`4`	`2. Producers can be divided into multiple partitions.`
-Original file line number
+Diff line change
@@ @@ -0,0 +1,8 @@ @@
 +package com.lld.multithreadedlogging.constant;
++
 +public enum LogLevel {
 +    INFO,
 +    DEBUG,
 +    WARN,
 +    ERROR
 +}