create method in MongoDBConfig for Listens events from mongodb and lo…

…ok all threads when added new topic to cache
XD-cods · May 16, 2024 · 9941a02 · 9941a02
1 parent 5540fa8
commit 9941a02
Show file tree

Hide file tree

Showing 5 changed files with 69 additions and 15 deletions.
diff --git a/src/main/java/org/example/Main.java b/src/main/java/org/example/Main.java
@@ -3,10 +3,12 @@
 import com.pengrad.telegrambot.TelegramBot;
 import org.example.configs.MongoDBConfig;
 import org.example.services.QuizService;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.PropertySource;
+import org.springframework.stereotype.Component;
 
 @PropertySource("classpath:application.properties")
 public class Main {

diff --git a/src/main/java/org/example/QuizBotListener.java b/src/main/java/org/example/QuizBotListener.java
@@ -26,10 +26,7 @@
 
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Random;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 public class QuizBotListener implements UpdatesListener {
@@ -69,9 +66,7 @@ public int process(List<Update> updates) throws NullPointerException {
       Long userId = message.chat().id();
       String messageText = message.text();
       QuizBotSession quizBotSession = null;
-      if (messageText.equals(UserBotConstants.START_BOT_COMMAND)) {
-        quizBotSession = sessionCache.computeIfAbsent(userId, () -> new QuizBotSession(QuizBotSessionMode.SESSION_CREATED));
-      }
+      quizBotSession = sessionCache.computeIfAbsent(userId, () -> new QuizBotSession(QuizBotSessionMode.SESSION_CREATED));
       //todo start here
       switch (messageText) {
         case UserBotConstants.START_BOT_COMMAND -> {

diff --git a/src/main/java/org/example/configs/MongoDBConfig.java b/src/main/java/org/example/configs/MongoDBConfig.java
@@ -3,13 +3,25 @@
 import com.mongodb.ConnectionString;
 import com.mongodb.client.MongoClient;
 import com.mongodb.client.MongoClients;
+import com.mongodb.client.model.changestream.FullDocument;
+import com.mongodb.client.model.changestream.OperationType;
+import org.example.model.QuizQuestions;
+import org.example.services.QuizService;
 import org.jetbrains.annotations.NotNull;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.PropertySource;
 import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration;
+import org.springframework.data.mongodb.core.ChangeStreamOptions;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.aggregation.Aggregation;
+import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
+import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
+import org.springframework.data.mongodb.core.messaging.Subscription;
+import org.springframework.data.mongodb.core.query.Criteria;
 import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
 
 @Configuration
@@ -18,12 +30,13 @@
 @EnableMongoRepositories("org.example.repositories")
 public class MongoDBConfig extends AbstractMongoClientConfiguration {
 
-
   @Value("${spring.data.mongodb.database}")
   private String databaseName;
+
   @Value("${spring.data.mongodb.uri}")
   private String connectionUri;
 
+
   @NotNull
   @Override
   protected String getDatabaseName() {
@@ -42,4 +55,24 @@ public MongoClient reconnectToDB() {
     currentClient.close();
     return MongoClients.create(new ConnectionString(connectionUri));
   }
+
+  @Bean
+  public DefaultMessageListenerContainer messageListenerContainer(MongoTemplate mongoTemplate, QuizService quizService) {
+    DefaultMessageListenerContainer container = new DefaultMessageListenerContainer(mongoTemplate);
+    container.start();
+    ChangeStreamOptions changeStreamOptions = ChangeStreamOptions.builder()
+            .filter(Aggregation.newAggregation(Aggregation.match(Criteria.where("operationType").is(OperationType.INSERT.getValue()))))
+            .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
+            .build();
+    ChangeStreamRequest.ChangeStreamRequestOptions options = new ChangeStreamRequest.ChangeStreamRequestOptions(databaseName, "quizQuestions", changeStreamOptions);
+
+    Subscription subscription = container.register(new ChangeStreamRequest<>(message -> {
+      QuizQuestions quizQuestions = message.getBody();
+      if (quizQuestions != null) {
+        String topic = quizQuestions.getTopicName();
+        quizService.addTopic(topic);
+      }
+    }, options), QuizQuestions.class);
+    return container;
+  }
 }
diff --git a/src/main/java/org/example/repositories/QuizRepo.java b/src/main/java/org/example/repositories/QuizRepo.java
@@ -1,14 +1,11 @@
 package org.example.repositories;
 
-import org.example.model.Question;
 import org.example.model.QuizQuestions;
 import org.springframework.data.mongodb.repository.Aggregation;
 import org.springframework.data.mongodb.repository.MongoRepository;
 import org.springframework.data.mongodb.repository.Query;
 import org.springframework.stereotype.Repository;
 
-import java.util.List;
-
 @Repository
 public interface QuizRepo extends MongoRepository<QuizQuestions, String> {
 
@@ -22,4 +19,11 @@ public interface QuizRepo extends MongoRepository<QuizQuestions, String> {
           "{ $group: {_id: \"$_id\", topicName:{$first:\"$topicName\"},questionList: { $push: \"$questionList\" },_class:{$first:\"$_class\"}} }",
   })
   QuizQuestions findRandomQuestionsByTopicName(String topicName, int n);
+
+  @Aggregation({
+          "{$project: {_id: 0,topicName: 1}}",
+          "{$group: {_id: 0,topicName: { $push: \"$topicName\" }}}",
+          "{$project: {topicName: 1,_id: 0}}"
+  })
+  String findAllTopic();
 }
diff --git a/src/main/java/org/example/services/QuizService.java b/src/main/java/org/example/services/QuizService.java
@@ -1,31 +1,51 @@
 package org.example.services;
 
-import org.example.model.Question;
+import org.cache2k.Cache;
+import org.cache2k.Cache2kBuilder;
 import org.example.model.QuizQuestions;
 import org.example.repositories.QuizRepo;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 @Service
 public class QuizService {
-  private static List<String> topics = new ArrayList<>();
+  private final Lock lock = new ReentrantLock();
+  private Cache<String, List<String>> topicsCache;
   private final QuizRepo quizRepo;
 
   @Autowired
   public QuizService(QuizRepo quizRepo) {
     this.quizRepo = quizRepo;
-    topics = quizRepo.findAll().stream().map(QuizQuestions::getTopicName).toList();
+    topicsCache = new Cache2kBuilder<String, List<String>>() {}
+                          .eternal(true)
+                          .build();
+    List<String> topics = Arrays.stream(quizRepo.findAllTopic().split(",")).toList();
+    topicsCache.put("topics", topics);
   }
 
   public List<String> getTopics() {
-    return topics;
+    return topicsCache.get("topics");
+  }
+
+  public synchronized void addTopic(String topicName){
+    lock.lock();
+    try {
+      List<String> topics = new ArrayList<>(topicsCache.get("topics"));
+      topics.add(topicName);
+      topicsCache.put("topics", topics);
+    } finally {
+      lock.unlock();
+    }
   }
 
   public void deleteAllQuiz() {
-    topics = new ArrayList<>();
+    topicsCache.remove("topics");
     quizRepo.deleteAll();
   }