refactor: added concurrency using completable future (#124)

* refactor: added concurrency using completable future

* refactor: added completablefuture implementation

* refactor: props has async-type

* refactor: fixedThreadPool added

* chore: added comment

* refactor: injection using factory instead of lookup

* test: added test

* fix: conditional expression

* test: fix tests

* chore: updated test props

* fix: conflict beans

* style: style

* fix: removed cached database between tests

* test: updated tests

* test: added tests

* refactor: removed unnecessary code

* style: style

src/main/java/ru/dankoy/korvotoanki/config/IoServiceConfig.java

@@ -0,0 +1,34 @@
+package ru.dankoy.korvotoanki.config;
+
+import java.util.function.Function;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
+import ru.dankoy.korvotoanki.core.service.filenameformatter.FileNameFormatterService;
+import ru.dankoy.korvotoanki.core.service.fileprovider.FileProviderService;
+import ru.dankoy.korvotoanki.core.service.io.IOService;
+import ru.dankoy.korvotoanki.core.service.io.IOServiceFile;
+
+@RequiredArgsConstructor
+@Configuration
+public class IoServiceConfig {
+
+  private final FileProviderService fileProviderService;
+  private final FileNameFormatterService fileNameFormatterService;
+
+  @Bean
+  public Function<String, IOService> ioServiceFileFactory() {
+    return name -> ioServiceFile(fileProviderService, fileNameFormatterService, name);
+  }
+
+  /** This bean is created as prototype and here only to inject it by function bean. */
+  @Bean
+  @Scope("prototype")
+  public IOService ioServiceFile(
+      FileProviderService fileProviderService,
+      FileNameFormatterService fileNameFormatterService,
+      String name) {
+    return new IOServiceFile(fileProviderService, fileNameFormatterService, name);
+  }
+}

src/main/java/ru/dankoy/korvotoanki/core/service/converter/AnkiConverterServiceCompletableFuture.java

@@ -0,0 +1,93 @@
+package ru.dankoy.korvotoanki.core.service.converter;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.stereotype.Service;
+import ru.dankoy.korvotoanki.config.Languages;
+import ru.dankoy.korvotoanki.config.appprops.ExternalApiProperties;
+import ru.dankoy.korvotoanki.core.domain.Vocabulary;
+import ru.dankoy.korvotoanki.core.domain.anki.AnkiData;
+import ru.dankoy.korvotoanki.core.domain.dictionaryapi.Word;
+import ru.dankoy.korvotoanki.core.exceptions.DictionaryApiException;
+import ru.dankoy.korvotoanki.core.exceptions.GoogleTranslatorException;
+import ru.dankoy.korvotoanki.core.fabric.anki.AnkiDataFabric;
+import ru.dankoy.korvotoanki.core.service.dictionaryapi.DictionaryService;
+import ru.dankoy.korvotoanki.core.service.googletrans.GoogleTranslator;
+
+@ConditionalOnExpression(
+    "${korvo-to-anki.async} and '${korvo-to-anki.async-type}'.equals('completable_future')")
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class AnkiConverterServiceCompletableFuture implements AnkiConverterService {
+
+  private final DictionaryService dictionaryService;
+  private final GoogleTranslator googleTranslator;
+  private final AnkiDataFabric ankiDataFabric;
+  private final ExternalApiProperties externalApiProperties;
+
+  public AnkiData convert(
+      Vocabulary vocabulary, String sourceLanguage, String targetLanguage, List<String> options) {
+
+    var isDictionaryApiEnabled = externalApiProperties.isDictionaryApiEnabled();
+
+    var fixedThreadPool = Executors.newFixedThreadPool(2);
+
+    log.debug(String.format("Working with word: '%s'", vocabulary.word()));
+
+    // ignore auto source language because won't know the defined language. Look up
+    // for Tika Language Detection
+    var cf1 =
+        CompletableFuture.supplyAsync(
+                () -> {
+                  if (isDictionaryApiEnabled
+                      && sourceLanguage.equals(Languages.EN.name().toLowerCase()))
+                    return dictionaryService.define(vocabulary.word());
+                  else return Collections.singletonList(Word.emptyWord());
+                },
+                fixedThreadPool)
+            .handle(
+                (result, ex) -> {
+                  if (ex != null && ex.getCause() instanceof DictionaryApiException) {
+                    log.warn(
+                        String.format(
+                            "Couldn't get definition from dictionaryapi.dev for '%s' - %s",
+                            vocabulary.word(), ex.getMessage()));
+                    return Collections.singletonList(Word.emptyWord());
+                  }
+                  return result;
+                });
+
+    var cf2 =
+        CompletableFuture.supplyAsync(
+                () ->
+                    googleTranslator.translate(
+                        vocabulary.word(), targetLanguage, sourceLanguage, options),
+                fixedThreadPool)
+            .handle(
+                (result, ex) -> {
+                  // it wraps exceptions in CompletionException
+                  if (ex != null && ex.getCause() instanceof GoogleTranslatorException gte) {
+                    log.error(
+                        String.format(
+                            "Couldn't translate '%s' from %s to %s - %s",
+                            vocabulary.word(), sourceLanguage, targetLanguage, ex.getMessage()));
+                    throw gte;
+                  }
+                  return result;
+                });
+
+    return CompletableFuture.allOf(cf1, cf2)
+        .thenApply(
+            ignored -> {
+              return ankiDataFabric.createAnkiData(vocabulary, cf2.join(), cf1.join());
+            })
+        .whenComplete((e, ex) -> fixedThreadPool.shutdownNow())
+        .join();
+  }
+}

src/main/java/ru/dankoy/korvotoanki/core/service/converter/AnkiConverterServiceImpl.java

@@ -4,6 +4,7 @@
 import java.util.List;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Service;
 import ru.dankoy.korvotoanki.config.Languages;
 import ru.dankoy.korvotoanki.config.appprops.ExternalApiProperties;
@@ -15,6 +16,7 @@
 import ru.dankoy.korvotoanki.core.service.dictionaryapi.DictionaryService;
 import ru.dankoy.korvotoanki.core.service.googletrans.GoogleTranslator;
 
+@ConditionalOnProperty(prefix = "korvo-to-anki", value = "async", havingValue = "false")
 @Slf4j
 @Service
 @RequiredArgsConstructor

src/main/java/ru/dankoy/korvotoanki/core/service/exporter/ExporterServiceAnkiAsync.java

@@ -9,7 +9,7 @@
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Lookup;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import ru.dankoy.korvotoanki.config.appprops.FilesProperties;
 import ru.dankoy.korvotoanki.core.domain.Vocabulary;
@@ -23,7 +23,8 @@
 import ru.dankoy.korvotoanki.core.service.templatecreator.TemplateCreatorService;
 import ru.dankoy.korvotoanki.core.service.vocabulary.VocabularyService;
 
-@ConditionalOnProperty(prefix = "korvo-to-anki", value = "async", havingValue = "true")
+@ConditionalOnExpression(
+    "${korvo-to-anki.async} and '${korvo-to-anki.async-type}' == 'countdownlatch'")
 @Slf4j
 @Service
 @RequiredArgsConstructor

src/main/java/ru/dankoy/korvotoanki/core/service/exporter/ExporterServiceAnkiCompletableFuture.java

@@ -0,0 +1,124 @@
+package ru.dankoy.korvotoanki.core.service.exporter;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.stereotype.Service;
+import ru.dankoy.korvotoanki.config.appprops.FilesProperties;
+import ru.dankoy.korvotoanki.core.domain.Vocabulary;
+import ru.dankoy.korvotoanki.core.domain.anki.AnkiData;
+import ru.dankoy.korvotoanki.core.service.converter.AnkiConverterService;
+import ru.dankoy.korvotoanki.core.service.io.IOService;
+import ru.dankoy.korvotoanki.core.service.state.StateService;
+import ru.dankoy.korvotoanki.core.service.templatecreator.TemplateCreatorService;
+import ru.dankoy.korvotoanki.core.service.vocabulary.VocabularyService;
+
+@ConditionalOnExpression(
+    "${korvo-to-anki.async} and '${korvo-to-anki.async-type}'.equals('completable_future')")
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class ExporterServiceAnkiCompletableFuture implements ExporterService {
+
+  private static final int STEP_SIZE = 30;
+  private static final int THREADS = 2;
+  private static final AtomicInteger atomicInteger = new AtomicInteger(0);
+  private final VocabularyService vocabularyService;
+  private final AnkiConverterService ankiConverterService;
+  private final TemplateCreatorService templateCreatorService;
+  private final FilesProperties filesProperties;
+  private final StateService sqliteStateService;
+  private final Function<String, IOService> ioServiceFileFactory;
+
+  @Override
+  public void export(String sourceLanguage, String targetLanguage, List<String> options) {
+
+    List<AnkiData> ankiDataList = new CopyOnWriteArrayList<>();
+
+    List<Vocabulary> vocabulariesFull = vocabularyService.getAll();
+    List<Vocabulary> filtered = sqliteStateService.filterState(vocabulariesFull);
+
+    var fixedThreadPool = Executors.newFixedThreadPool(THREADS);
+
+    if (!filtered.isEmpty()) {
+
+      CompletableFuture<Void> concurrentExportAllOf = null;
+
+      if (filtered.size() < THREADS) {
+
+        CompletableFuture<Void> future1 =
+            CompletableFuture.runAsync(
+                () -> asyncFunc(ankiDataList, filtered, sourceLanguage, targetLanguage, options),
+                fixedThreadPool);
+
+        concurrentExportAllOf = CompletableFuture.allOf(future1);
+
+      } else {
+
+        List<Vocabulary> oneV = filtered.subList(0, filtered.size() / 2);
+        List<Vocabulary> twoV = filtered.subList((filtered.size() / 2), filtered.size());
+
+        CompletableFuture<Void> future1 =
+            CompletableFuture.runAsync(
+                () -> asyncFunc(ankiDataList, oneV, sourceLanguage, targetLanguage, options),
+                fixedThreadPool);
+        CompletableFuture<Void> future2 =
+            CompletableFuture.runAsync(
+                () -> asyncFunc(ankiDataList, twoV, sourceLanguage, targetLanguage, options),
+                fixedThreadPool);
+        concurrentExportAllOf = CompletableFuture.allOf(future1, future2);
+      }
+
+      // wait till export is finished
+      concurrentExportAllOf.join();
+
+      var ioService = ioServiceFileFactory.apply(filesProperties.getExportFileName());
+
+      // prepare cf for printing the template
+      CompletableFuture<String> template =
+          CompletableFuture.supplyAsync(
+              () -> templateCreatorService.create(ankiDataList), fixedThreadPool);
+
+      // run all of the futures in parallel
+      CompletableFuture<Void> printExportAndSaveState =
+          CompletableFuture.allOf(
+              CompletableFuture.runAsync(() -> ioService.print(template.join()), fixedThreadPool),
+              CompletableFuture.runAsync(
+                  () -> sqliteStateService.saveState(filtered), fixedThreadPool));
+
+      // wait till done
+      printExportAndSaveState.whenComplete((e, ex) -> fixedThreadPool.shutdownNow()).join();
+
+    } else {
+      log.info("State is the same as database. Export is not necessary.");
+    }
+  }
+
+  private void asyncFunc(
+      List<AnkiData> ankiDataList,
+      List<Vocabulary> vocabularies,
+      String sourceLanguage,
+      String targetLanguage,
+      List<String> options) {
+    for (Vocabulary v : vocabularies) {
+      var i = atomicInteger.getAndIncrement();
+      // sleep is not necessary anymore since rate limiter for dictionary api is
+      // implemented
+      if (i != 0 && i % STEP_SIZE == 0) {
+        log.info("processed - {}", i);
+      }
+      var ankiData = ankiConverterService.convert(v, sourceLanguage, targetLanguage, options);
+      log.info(
+          "Thread {} obtained new anki for word {}",
+          Thread.currentThread().getName(),
+          ankiData.getWord());
+      ankiDataList.add(ankiData);
+    }
+  }
+}

src/main/java/ru/dankoy/korvotoanki/core/service/io/IOServiceFile.java

@@ -7,12 +7,15 @@
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.Primary;
 import org.springframework.context.annotation.Scope;
 import org.springframework.stereotype.Service;
 import ru.dankoy.korvotoanki.core.exceptions.IoException;
 import ru.dankoy.korvotoanki.core.service.filenameformatter.FileNameFormatterService;
 import ru.dankoy.korvotoanki.core.service.fileprovider.FileProviderService;
 
+// Injection this bean from configuration class bean #{link IoServiceConfig}
+@Primary
 @Service
 @Scope(value = BeanDefinition.SCOPE_PROTOTYPE)
 @Slf4j

src/main/resources/application.yml

@@ -37,6 +37,7 @@ korvo-to-anki:
     export-file-name: export-${spring.application.name}
     state-file-name: ${spring.application.name}.state
   async: true
+  async-type: "completable_future"
   http-client: "web-client"
   api:
     dictionaryApiEnabled: true