Add code

Author: anomaly2104

.gitignore

@@ -0,0 +1,32 @@
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**
+!**/src/test/**
+
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
+
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
+
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+
+### VS Code ###
+.vscode/
+.DS_Store

pom.xml

@@ -0,0 +1,24 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>org.example</groupId>
+    <artifactId>message-queue-pub-sub</artifactId>
+    <version>1.0-SNAPSHOT</version>
+    <dependencies>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>RELEASE</version>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>15</maven.compiler.source>
+        <maven.compiler.target>15</maven.compiler.target>
+    </properties>
+
+</project>

problem-statement.md

@@ -0,0 +1,9 @@
+## Message Queue
+We have to design a message queue supporting publisher-subscriber model. It should support following operations:
+
+* It should support multiple topics where messages can be published.
+* Publisher should be able to publish a message to a particular topic.
+* Subscribers should be able to subscribe to a topic.
+* Whenever a message is published to a topic, all the subscribers, who are subscribed to that topic, should receive the message.
+* Subscribers should be able to run in parallel
+

src/main/java/com/uditagarwal/Main.java

@@ -0,0 +1,31 @@
+package com.uditagarwal;
+
+import com.uditagarwal.pub_sub_queue.public_interface.Queue;
+import com.uditagarwal.pub_sub_queue.model.Message;
+import com.uditagarwal.pub_sub_queue.model.Topic;
+
+public class Main {
+    public static void main(String[] args) throws InterruptedException {
+        final Queue queue = new Queue();
+        final Topic topic1 = queue.createTopic("t1");
+        final Topic topic2 = queue.createTopic("t2");
+        final SleepingSubscriber sub1 = new SleepingSubscriber("sub1", 10000);
+        final SleepingSubscriber sub2 = new SleepingSubscriber("sub2", 10000);
+        queue.subscribe(sub1, topic1);
+        queue.subscribe(sub2, topic1);
+
+        final SleepingSubscriber sub3 = new SleepingSubscriber("sub3", 5000);
+        queue.subscribe(sub3, topic2);
+
+        queue.publish(topic1, new Message("m1"));
+        queue.publish(topic1, new Message("m2"));
+
+        queue.publish(topic2, new Message("m3"));
+
+        Thread.sleep(15000);
+        queue.publish(topic2, new Message("m4"));
+        queue.publish(topic1, new Message("m5"));
+
+        queue.resetOffset(topic1, sub1, 0);
+    }
+}

src/main/java/com/uditagarwal/SleepingSubscriber.java

@@ -0,0 +1,26 @@
+package com.uditagarwal;
+
+import com.uditagarwal.pub_sub_queue.public_interface.ISubscriber;
+import com.uditagarwal.pub_sub_queue.model.Message;
+
+public class SleepingSubscriber implements ISubscriber {
+    private final String id;
+    private final int sleepTimeInMillis;
+
+    public SleepingSubscriber(String id, int sleepTimeInMillis) {
+        this.id = id;
+        this.sleepTimeInMillis = sleepTimeInMillis;
+    }
+
+    @Override
+    public String getId() {
+        return id;
+    }
+
+    @Override
+    public void consume(Message message) throws InterruptedException {
+        System.out.println("Subscriber: " + id + " started consuming: " + message.getMsg());
+        Thread.sleep(sleepTimeInMillis);
+        System.out.println("Subscriber: " + id + " done consuming: " + message.getMsg());
+    }
+}

src/main/java/com/uditagarwal/pub_sub_queue/handler/SubscriberWorker.java

@@ -0,0 +1,45 @@
+package com.uditagarwal.pub_sub_queue.handler;
+
+import com.uditagarwal.pub_sub_queue.model.Message;
+import com.uditagarwal.pub_sub_queue.model.Topic;
+import com.uditagarwal.pub_sub_queue.model.TopicSubscriber;
+import lombok.Getter;
+import lombok.NonNull;
+import lombok.SneakyThrows;
+
+@Getter
+public class SubscriberWorker implements Runnable {
+
+    private final Topic topic;
+    private final TopicSubscriber topicSubscriber;
+
+    public SubscriberWorker(@NonNull final Topic topic, @NonNull final TopicSubscriber topicSubscriber) {
+        this.topic = topic;
+        this.topicSubscriber = topicSubscriber;
+    }
+
+    @SneakyThrows
+    @Override
+    public void run() {
+        synchronized (topicSubscriber) {
+            do {
+                int curOffset = topicSubscriber.getOffset().get();
+                while (curOffset >= topic.getMessages().size()) {
+                    topicSubscriber.wait();
+                }
+                Message message = topic.getMessages().get(curOffset);
+                topicSubscriber.getSubscriber().consume(message);
+
+                // We cannot just increment here since subscriber offset can be reset while it is consuming. So, after
+                // consuming we need to increase only if it was previous one.
+                topicSubscriber.getOffset().compareAndSet(curOffset, curOffset + 1);
+            } while (true);
+        }
+    }
+
+    synchronized public void wakeUpIfNeeded() {
+        synchronized (topicSubscriber) {
+            topicSubscriber.notify();
+        }
+    }
+}

src/main/java/com/uditagarwal/pub_sub_queue/handler/TopicHandler.java

@@ -0,0 +1,35 @@
+package com.uditagarwal.pub_sub_queue.handler;
+
+import com.uditagarwal.pub_sub_queue.model.Topic;
+import com.uditagarwal.pub_sub_queue.model.TopicSubscriber;
+import lombok.NonNull;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class TopicHandler {
+    private final Topic topic;
+    private final Map<String, SubscriberWorker> subscriberWorkers;
+
+    public TopicHandler(@NonNull final Topic topic) {
+        this.topic = topic;
+        subscriberWorkers = new HashMap<>();
+    }
+
+    public void publish() {
+        for (TopicSubscriber topicSubscriber:topic.getSubscribers()) {
+            startSubsriberWorker(topicSubscriber);
+        }
+    }
+
+    public void startSubsriberWorker(@NonNull final TopicSubscriber topicSubscriber) {
+        final String subscriberId = topicSubscriber.getSubscriber().getId();
+        if (!subscriberWorkers.containsKey(subscriberId)) {
+            final SubscriberWorker subscriberWorker = new SubscriberWorker(topic, topicSubscriber);
+            subscriberWorkers.put(subscriberId, subscriberWorker);
+            new Thread(subscriberWorker).start();
+        }
+        final SubscriberWorker subscriberWorker = subscriberWorkers.get(subscriberId);
+        subscriberWorker.wakeUpIfNeeded();
+    }
+}

src/main/java/com/uditagarwal/pub_sub_queue/model/Message.java

@@ -0,0 +1,11 @@
+package com.uditagarwal.pub_sub_queue.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+@AllArgsConstructor
+@Getter
+public class Message {
+    private final String msg;
+
+}

src/main/java/com/uditagarwal/pub_sub_queue/model/Topic.java

@@ -0,0 +1,30 @@
+package com.uditagarwal.pub_sub_queue.model;
+
+import lombok.Getter;
+import lombok.NonNull;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Getter
+public class Topic {
+    private final String topicName;
+    private final String topicId;
+    private final List<Message> messages; // TODO: Change getter this to send only immutable list outside.
+    private final List<TopicSubscriber> subscribers; // TODO: Change getter this to send only immutable list outside.
+
+    public Topic(@NonNull final String topicName, @NonNull final String topicId) {
+        this.topicName = topicName;
+        this.topicId = topicId;
+        this.messages = new ArrayList<>();
+        this.subscribers = new ArrayList<>();
+    }
+
+    public synchronized void addMessage(@NonNull final Message message) {
+        messages.add(message);
+    }
+
+    public void addSubscriber(@NonNull final TopicSubscriber subscriber) {
+        subscribers.add(subscriber);
+    }
+}

src/main/java/com/uditagarwal/pub_sub_queue/model/TopicSubscriber.java

@@ -0,0 +1,19 @@
+package com.uditagarwal.pub_sub_queue.model;
+
+import com.uditagarwal.pub_sub_queue.public_interface.ISubscriber;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+@Getter
+@AllArgsConstructor
+public class TopicSubscriber {
+    AtomicInteger offset;
+    ISubscriber subscriber;
+
+    public TopicSubscriber(ISubscriber subscriber) {
+        this.subscriber = subscriber;
+        this.offset = new AtomicInteger(0);
+    }
+}

src/main/java/com/uditagarwal/pub_sub_queue/public_interface/ISubscriber.java

@@ -0,0 +1,9 @@
+package com.uditagarwal.pub_sub_queue.public_interface;
+
+import com.uditagarwal.pub_sub_queue.model.Message;
+
+public interface ISubscriber {
+
+    String getId();
+    void consume(Message message) throws InterruptedException;
+}

src/main/java/com/uditagarwal/pub_sub_queue/public_interface/Queue.java

@@ -0,0 +1,49 @@
+package com.uditagarwal.pub_sub_queue.public_interface;
+
+import com.uditagarwal.pub_sub_queue.handler.TopicHandler;
+import com.uditagarwal.pub_sub_queue.model.Message;
+import com.uditagarwal.pub_sub_queue.model.Topic;
+import com.uditagarwal.pub_sub_queue.model.TopicSubscriber;
+import lombok.NonNull;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+public class Queue {
+    Map<String, TopicHandler> topicProcessors;
+
+    public Queue() {
+        this.topicProcessors = new HashMap<>();
+    }
+
+    public Topic createTopic(@NonNull final String topicName) {
+        final Topic topic = new Topic(topicName, UUID.randomUUID().toString());
+        TopicHandler topicHandler = new TopicHandler(topic);
+        topicProcessors.put(topic.getTopicId(), topicHandler);
+        System.out.println("Created topic: " + topic.getTopicName());
+        return topic;
+    }
+
+    public void subscribe(@NonNull final ISubscriber subscriber, @NonNull final Topic topic) {
+        topic.addSubscriber(new TopicSubscriber(subscriber));
+        System.out.println(subscriber.getId() + " subscribed to topic: " + topic.getTopicName());
+    }
+
+    public void publish(@NonNull final Topic topic, @NonNull final Message message) {
+        topic.addMessage(message);
+        System.out.println(message.getMsg() + " published to topic: " + topic.getTopicName());
+        new Thread(() -> topicProcessors.get(topic.getTopicId()).publish()).start();
+    }
+
+    public void resetOffset(@NonNull final Topic topic, @NonNull final ISubscriber subscriber, @NonNull final Integer newOffset) {
+        for (TopicSubscriber topicSubscriber : topic.getSubscribers()) {
+            if (topicSubscriber.getSubscriber().equals(subscriber)) {
+                topicSubscriber.getOffset().set(newOffset);
+                System.out.println(topicSubscriber.getSubscriber().getId() + " offset reset to: " + newOffset);
+                new Thread(() -> topicProcessors.get(topic.getTopicId()).startSubsriberWorker(topicSubscriber)).start();
+                break;
+            }
+        }
+    }
+}