optimized delay management for sending posts

yvasyliev · Nov 8, 2023 · 9629779 · 9629779
1 parent ba0bea8
commit 9629779
Show file tree

Hide file tree

Showing 18 changed files with 188 additions and 180 deletions.
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
@@ -31,10 +31,13 @@ jobs:
       - name: Run test
         run: mvn test
         env:
+          spring.jpa.hibernate.ddl-auto: create-drop
           telegram.admin.id: ${{ secrets.telegram.admin.id }}
           telegram.bot.token: ${{ secrets.test.telegram.bot.token }}
           telegram.channel.id: ${{ secrets.test.telegram.channel.id }}
           telegram.chat.id: ${{ secrets.test.telegram.chat.id }}
+          telegram.schedule.posting.enabled: false
+
 
   release:
     runs-on: ubuntu-latest

diff --git a/src/main/java/com/github/yvasyliev/Application.java b/src/main/java/com/github/yvasyliev/Application.java
@@ -13,28 +13,25 @@
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.context.ApplicationPidFileWriter;
-import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Scope;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.telegram.telegrambots.meta.TelegramBotsApi;
 import org.telegram.telegrambots.meta.exceptions.TelegramApiException;
 import org.telegram.telegrambots.starter.TelegramBotStarterConfiguration;
 import org.telegram.telegrambots.updatesreceivers.DefaultBotSession;
 
-import java.io.File;
 import java.net.http.HttpClient;
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 @SpringBootApplication
+@EnableScheduling
 public class Application extends TelegramBotStarterConfiguration {
-    private static ConfigurableApplicationContext context;
+    private static ApplicationContext context;
 
     public static void main(String[] args) {
         context = new SpringApplicationBuilder(Application.class)
@@ -43,8 +40,10 @@ public static void main(String[] args) {
                 .run(args);
     }
 
-    public static ConfigurableApplicationContext getContext() {
-        return context;
+    public static void withContext(Consumer<ApplicationContext> contextConsumer) {
+        if (context != null) {
+            contextConsumer.accept(context);
+        }
     }
 
     @Bean("/help")
@@ -71,11 +70,6 @@ public Module module(JsonDeserializer<Post> postJsonDeserializer) {
         return module;
     }
 
-    @Bean
-    public ScheduledExecutorService scheduledExecutorService() {
-        return Executors.newSingleThreadScheduledExecutor();
-    }
-
     @Bean
     public String userAgent(@Value("${REDDIT_USERNAME}") String redditUsername) {
         return "java:reddit-telegram-repeater:2.0.0  (by /u/%s)".formatted(redditUsername);
@@ -86,11 +80,6 @@ public HttpClient httpClient() {
         return HttpClient.newHttpClient();
     }
 
-    @Bean
-    public File stateSrc() {
-        return new File("state.json");
-    }
-
     @Bean
     @Scope(BeanDefinition.SCOPE_PROTOTYPE)
     public <K, V> Map<K, V> synchronizedFixedSizeMap(@Value("16") int maxSize) {
@@ -126,13 +115,4 @@ public Map<Long, ExternalMessageData> longExternalMessageDataMap(@Value("16") in
         return synchronizedFixedSizeMap(maxSize);
     }
 
-    @Bean
-    public Executor delayedExecutor() {
-        return CompletableFuture.delayedExecutor(10, TimeUnit.SECONDS, singleThreadExecutor());
-    }
-
-    @Bean
-    public Executor singleThreadExecutor() {
-        return Executors.newSingleThreadExecutor();
-    }
 }
diff --git a/src/main/java/com/github/yvasyliev/appenders/TelegramBotAppender.java b/src/main/java/com/github/yvasyliev/appenders/TelegramBotAppender.java
@@ -19,9 +19,13 @@
 import java.io.PrintWriter;
 import java.io.Serializable;
 import java.io.StringWriter;
+import java.util.Optional;
 
 @Plugin(name = "TelegramBotAppender", category = Core.CATEGORY_NAME, elementType = Appender.ELEMENT_TYPE)
 public class TelegramBotAppender extends AbstractAppender {
+    private static final String MESSAGE_TEMPLATE = """
+            %s
+            %s""";
 
     protected TelegramBotAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions, Property[] properties) {
         super(name, filter, layout, ignoreExceptions, properties);
@@ -34,32 +38,33 @@ public static TelegramBotAppender createAppender(@PluginAttribute("name") String
 
     @Override
     public void append(LogEvent event) {
-        var formattedMessage = event.getMessage().getFormattedMessage();
-
-        var stackTrace = getStackTrace(event.getThrown());
-        if (stackTrace != null) {
-            formattedMessage = """
-                    %s
-                    %s""".formatted(formattedMessage, stackTrace);
-        }
-
-        try {
-            Application.getContext().getBean(TelegramNotifier.class).applyWithException(formattedMessage);
-        } catch (Exception e) {
-            e.printStackTrace(System.err);
-        }
+        Application.withContext(applicationContext -> {
+            try {
+                applicationContext
+                        .getBean(TelegramNotifier.class)
+                        .applyWithException(buildMessage(event));
+            } catch (Exception e) {
+                e.printStackTrace(System.err);
+            }
+        });
     }
 
-    private String getStackTrace(Throwable throwable) {
-        if (throwable == null) {
-            return null;
-        }
+    private String buildMessage(LogEvent event) {
+        var formattedMessage = event.getMessage().getFormattedMessage();
+        return getStackTrace(event.getThrown())
+                .map(stackTrace -> MESSAGE_TEMPLATE.formatted(formattedMessage, stackTrace))
+                .orElse(formattedMessage);
+    }
 
-        try (var stringWriter = new StringWriter(); var printWriter = new PrintWriter(stringWriter)) {
-            throwable.printStackTrace(printWriter);
-            return stringWriter.toString();
-        } catch (IOException e) {
-            return "Failed to read stack trace: " + e;
-        }
+    private Optional<String> getStackTrace(Throwable throwable) {
+        return Optional.ofNullable(throwable).map(t -> {
+            try (var stringWriter = new StringWriter(); var printWriter = new PrintWriter(stringWriter)) {
+                t.printStackTrace(printWriter);
+                return stringWriter.toString();
+            } catch (IOException e) {
+                e.printStackTrace(System.err);
+                return null;
+            }
+        });
     }
 }
diff --git a/src/main/java/com/github/yvasyliev/bots/telegram/RedTelBot.java b/src/main/java/com/github/yvasyliev/bots/telegram/RedTelBot.java
@@ -6,6 +6,7 @@
 import com.github.yvasyliev.model.dto.ChannelPost;
 import com.github.yvasyliev.model.dto.ExternalMessageData;
 import com.github.yvasyliev.model.events.NewChannelPostEvent;
+import com.github.yvasyliev.service.async.DelayedBlockingExecutor;
 import com.github.yvasyliev.service.telegram.callbacks.Callback;
 import com.github.yvasyliev.service.telegram.commands.Command;
 import jakarta.annotation.PreDestroy;
@@ -17,6 +18,8 @@
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Component;
 import org.telegram.telegrambots.meta.TelegramBotsApi;
+import org.telegram.telegrambots.meta.api.methods.send.SendMediaGroup;
+import org.telegram.telegrambots.meta.api.methods.send.SendPhoto;
 import org.telegram.telegrambots.meta.api.objects.CallbackQuery;
 import org.telegram.telegrambots.meta.api.objects.Message;
 import org.telegram.telegrambots.meta.api.objects.Update;
@@ -25,8 +28,10 @@
 import org.telegram.telegrambots.meta.generics.BotSession;
 import org.telegram.telegrambots.starter.AfterBotRegistration;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 
 @Component
 public class RedTelBot extends AbstractRedTelBot {
@@ -49,6 +54,9 @@ public class RedTelBot extends AbstractRedTelBot {
     @Autowired
     private ApplicationEventPublisher eventPublisher;
 
+    @Autowired
+    private DelayedBlockingExecutor delayedBlockingExecutor;
+
     public RedTelBot(@Value("${telegram.bot.token}") String botToken) {
         super(botToken);
     }
@@ -140,6 +148,14 @@ public boolean isAdmin(User user) {
         return user.getId().toString().equals(getAdminId());
     }
 
+    public CompletableFuture<List<Message>> executeDelayed(SendMediaGroup sendMediaGroup) {
+        return delayedBlockingExecutor.submit(() -> execute(sendMediaGroup));
+    }
+
+    public CompletableFuture<Message> executeDelayed(SendPhoto sendPhoto) {
+        return delayedBlockingExecutor.submit(() -> execute(sendPhoto));
+    }
+
     private Optional<String> getCommand(Message message) {
         long userId = message.getFrom().getId();
         if (message.hasText()) {

diff --git a/src/main/java/com/github/yvasyliev/service/async/DelayedBlockingExecutor.java b/src/main/java/com/github/yvasyliev/service/async/DelayedBlockingExecutor.java
@@ -0,0 +1,35 @@
+package com.github.yvasyliev.service.async;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.util.Queue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class DelayedBlockingExecutor {
+    @Value("#{new java.util.concurrent.LinkedBlockingQueue()}")
+    private Queue<Runnable> delayedRunnables;
+
+    public <T> CompletableFuture<T> submit(Callable<T> callable) {
+        var completableFuture = new CompletableFuture<T>();
+        delayedRunnables.add(() -> {
+            try {
+                completableFuture.complete(callable.call());
+            } catch (Exception e) {
+                completableFuture.completeExceptionally(e);
+            }
+        });
+        return completableFuture;
+    }
+
+    @Scheduled(fixedDelayString = "${delayed.blocking.executor.delay.in.seconds:20}", timeUnit = TimeUnit.SECONDS)
+    public void runNextRunnable() {
+        if (!delayedRunnables.isEmpty()) {
+            delayedRunnables.poll().run();
+        }
+    }
+}
diff --git a/src/main/java/com/github/yvasyliev/service/data/BlockedAuthorService.java b/src/main/java/com/github/yvasyliev/service/data/BlockedAuthorService.java
@@ -6,15 +6,13 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.util.List;
-
 @Service
 public class BlockedAuthorService {
     @Autowired
     private BlockedAuthorRepository blockedAuthorRepository;
 
-    public List<BlockedAuthor> findAll() {
-        return blockedAuthorRepository.findAll();
+    public boolean isBlocked(String username) {
+        return blockedAuthorRepository.existsById(username);
     }
 
     @Transactional

diff --git a/src/main/java/com/github/yvasyliev/service/deserializers/PostDeserializer.java b/src/main/java/com/github/yvasyliev/service/deserializers/PostDeserializer.java
@@ -7,7 +7,6 @@
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.github.yvasyliev.model.dto.post.Post;
-import com.github.yvasyliev.model.entities.BlockedAuthor;
 import com.github.yvasyliev.service.data.BlockedAuthorService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -30,20 +29,15 @@ public Post deserialize(JsonParser jsonParser, DeserializationContext deserializ
         var jsonPost = jsonParser.readValueAs(JsonNode.class);
         var author = jsonPost.get("author").textValue();
         var created = jsonPost.get("created").intValue();
-        var postUrl = jsonPost.has("url_overridden_by_dest") ? jsonPost.get("url_overridden_by_dest").textValue() : jsonPost.get("url").textValue();
-        var blockedAuthors = blockedAuthorService
-                .findAll()
-                .stream()
-                .map(BlockedAuthor::getUsername)
-                .toList();
+        var postUrl = jsonPost.has("url_overridden_by_dest") ? jsonPost.get("url_overridden_by_dest").textValue() : jsonPost.get("url").textValue(); // TODO: 11/8/2023 simplify
 
         jsonPost = extractRootPost(jsonPost);
         for (var postMapper : postMappers) {
             try {
                 var optionalPost = postMapper.applyWithException(jsonPost).map(post -> {
                     post.setAuthor(author);
                     post.setCreated(created);
-                    post.setApproved(!blockedAuthors.contains(author));
+                    post.setApproved(!blockedAuthorService.isBlocked(author));
                     post.setPostUrl(postUrl);
                     return post;
                 });

diff --git a/src/main/java/com/github/yvasyliev/service/reddit/RedditVideoDownloader.java b/src/main/java/com/github/yvasyliev/service/reddit/RedditVideoDownloader.java
@@ -7,6 +7,7 @@
 
 import java.io.IOException;
 
+// TODO: 11/7/2023 remove Jsoup
 @Service
 public class RedditVideoDownloader implements ThrowingFunction<String, String> {
     @Override