From b4ace30cc195c16babce4ea5288c236478d672b6 Mon Sep 17 00:00:00 2001 From: jarvis2f <137974272+jarvis2f@users.noreply.github.com> Date: Wed, 5 Feb 2025 23:25:05 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=85=20chore:=20Add=20concurrent=20update?= =?UTF-8?q?=20file=20download=20status=20test?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 1 + .../main/java/telegram/files/MessyUtils.java | 14 ++ .../FileDownloadStatusConcurrentTest.java | 175 ++++++++++++++++++ .../files/UpdateFileDownloadProcess.java | 44 +++++ 4 files changed, 234 insertions(+) create mode 100644 api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java create mode 100644 api/src/test/java/telegram/files/UpdateFileDownloadProcess.java diff --git a/.env.example b/.env.example index 64158c9..d5aa4ec 100644 --- a/.env.example +++ b/.env.example @@ -5,6 +5,7 @@ APP_ENV=prod APP_ROOT=/app/data # only use in test #TDLIB_PATH= +#JAVA_HOME= TELEGRAM_API_ID= TELEGRAM_API_HASH= diff --git a/api/src/main/java/telegram/files/MessyUtils.java b/api/src/main/java/telegram/files/MessyUtils.java index 839ebd7..3a7eb83 100644 --- a/api/src/main/java/telegram/files/MessyUtils.java +++ b/api/src/main/java/telegram/files/MessyUtils.java @@ -1,5 +1,7 @@ package telegram.files; +import io.vertx.core.Future; + import java.io.File; import java.io.FileInputStream; import java.nio.MappedByteBuffer; @@ -61,4 +63,16 @@ public class MessyUtils { return md5File1.equals(md5File2); } + + public static T await(Future future) { + CompletableFuture completableFuture = new CompletableFuture<>(); + future.onComplete(result -> { + if (result.succeeded()) { + completableFuture.complete(result.result()); + } else { + completableFuture.completeExceptionally(result.cause()); + } + }); + return completableFuture.join(); + } } diff --git a/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java b/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java new file mode 100644 index 0000000..4996f21 --- /dev/null +++ b/api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java @@ -0,0 +1,175 @@ +package telegram.files; + +import cn.hutool.core.io.FileUtil; +import io.vertx.core.Future; +import io.vertx.core.Vertx; +import io.vertx.core.json.JsonObject; +import org.junit.jupiter.api.*; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import telegram.files.repository.FileRecord; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; + +@Execution(ExecutionMode.CONCURRENT) +public class FileDownloadStatusConcurrentTest { + + private static final int THREAD_COUNT = 10; + + private static ExecutorService executor; + + static FileRecord fileRecord = new FileRecord( + 1, "unique_id", 1, 1, 1, 1, false, 1, 0, "type", "mime_type", "file_name", "thumbnail", "caption", null, + FileRecord.DownloadStatus.idle.name(), FileRecord.TransferStatus.idle.name(), 0, null + ); + + @BeforeAll + static void setUpAll() { + executor = Executors.newFixedThreadPool(THREAD_COUNT); + + Vertx vertx = Vertx.vertx(); + Future future = vertx.deployVerticle(new DataVerticle()); + MessyUtils.await(future); + + Future fileRecordFuture = DataVerticle.fileRepository.create(fileRecord); + MessyUtils.await(fileRecordFuture); + } + + @AfterAll + static void tearDownAll() { + executor.shutdown(); + String dataPath = DataVerticle.getDataPath(); + if (FileUtil.file(dataPath).exists()) { + FileUtil.del(dataPath); + } + } + + @RepeatedTest(THREAD_COUNT) + @DisplayName("Concurrent update file download status") + void concurrentUpdateFileDownloadStatusTest() { + int newFileId = 2; + + String updateLocalPath = "local_path_" + Thread.currentThread().threadId(); + Long completionDate = System.currentTimeMillis(); + + Future future = DataVerticle.fileRepository.updateDownloadStatus(newFileId, fileRecord.uniqueId(), updateLocalPath, FileRecord.DownloadStatus.downloading, completionDate); + MessyUtils.await(future); + + Future finalRecordFuture = DataVerticle.fileRepository.getByPrimaryKey(newFileId, fileRecord.uniqueId()); + FileRecord finalRecord = MessyUtils.await(finalRecordFuture); + + Assertions.assertEquals(FileRecord.DownloadStatus.downloading.name(), finalRecord.downloadStatus()); + Assertions.assertNotNull(finalRecord.localPath()); + Assertions.assertNotNull(finalRecord.completionDate()); + } + + @Test + @DisplayName("Multi-threaded concurrent update file download status using ExecutorService") + void multiThreadedConcurrentUpdateFileDownloadStatusTest() throws Exception { + int newFileId = 2; + + List> futures = new ArrayList<>(); + AtomicInteger completedCount = new AtomicInteger(0); + + for (int i = 0; i < THREAD_COUNT; i++) { + int threadIndex = i; + futures.add(executor.submit(() -> { + String updateLocalPath = "local_path_" + threadIndex; + Long completionDate = System.currentTimeMillis(); + System.out.println("Thread start update: " + Thread.currentThread().getName() + ", localPath: " + updateLocalPath + ", completionDate: " + completionDate); + + try { + DataVerticle.fileRepository.updateDownloadStatus(newFileId, + fileRecord.uniqueId(), + updateLocalPath, + FileRecord.DownloadStatus.downloading, + completionDate) + .onComplete(result -> completedCount.incrementAndGet()); + } catch (Exception e) { + throw new RuntimeException(e); + } + return null; + })); + } + + for (int i = 0; i < THREAD_COUNT; i++) { + futures.add(executor.submit(() -> { + try { + DataVerticle.fileRepository.getByUniqueId(fileRecord.uniqueId()); + } catch (Exception e) { + throw new RuntimeException(e); + } + return null; + })); + } + + for (java.util.concurrent.Future future : futures) { + future.get(); + } + while (completedCount.get() < THREAD_COUNT) { + Thread.sleep(100); + } + + Future finalRecordFuture = DataVerticle.fileRepository.getByPrimaryKey(newFileId, fileRecord.uniqueId()); + FileRecord finalRecord = MessyUtils.await(finalRecordFuture); + + Assertions.assertEquals(FileRecord.DownloadStatus.downloading.name(), finalRecord.downloadStatus()); + Assertions.assertNotNull(finalRecord.localPath()); + Assertions.assertNotNull(finalRecord.completionDate()); + } + + @Test + @DisplayName("Multi-process concurrent update file download status") + void multiProcessConcurrentUpdateFileDownloadStatusTest() throws Exception { + int newFileId = 2; + + int processCount = 5; + List processes = new ArrayList<>(); + String classpath = System.getProperty("java.class.path"); + String javaPath = System.getProperty("java.home") + "/bin/java"; + + for (int i = 0; i < processCount; i++) { + String localPath = "local_path_" + i; + ProcessBuilder pb = new ProcessBuilder(javaPath, + "-cp", + classpath, + "-Djava.library.path=" + System.getenv("TDLIB_PATH"), + "telegram.files.UpdateFileDownloadProcess", + String.valueOf(newFileId), + "unique_id", + localPath, + "downloading", + "1"); + pb.environment().putAll(System.getenv()); + pb.inheritIO(); + processes.add(pb.start()); + } + + for (Process process : processes) { + process.waitFor(); + } + + Future future = DataVerticle.fileRepository.getByPrimaryKey(newFileId, fileRecord.uniqueId()); + FileRecord finalRecord = MessyUtils.await(future); + + Assertions.assertNotNull(finalRecord, "Final record should not be null"); + Assertions.assertTrue( + finalRecord.downloadStatus().equals(FileRecord.DownloadStatus.downloading.name()) || + finalRecord.downloadStatus().equals(FileRecord.DownloadStatus.completed.name()), + "Download status should be one of the concurrent updates" + ); + Assertions.assertTrue( + IntStream.range(0, processCount).anyMatch(i -> finalRecord.localPath().equals("local_path_" + i)), + "Local path should match one of the concurrent updates" + ); + Assertions.assertTrue( + finalRecord.completionDate().equals(1L) || finalRecord.completionDate().equals(2L), + "Completion date should match one of the concurrent updates" + ); + } +} diff --git a/api/src/test/java/telegram/files/UpdateFileDownloadProcess.java b/api/src/test/java/telegram/files/UpdateFileDownloadProcess.java new file mode 100644 index 0000000..8a74f6c --- /dev/null +++ b/api/src/test/java/telegram/files/UpdateFileDownloadProcess.java @@ -0,0 +1,44 @@ +package telegram.files; + + +import io.vertx.core.Future; +import io.vertx.core.Vertx; +import io.vertx.core.json.JsonObject; +import telegram.files.repository.FileRecord; + +public class UpdateFileDownloadProcess { + + public static void main(String[] args) { + if (args.length < 5) { + System.out.println("Usage: UpdateFileDownloadProcess "); + System.exit(1); + } + System.out.println("Process: " + ProcessHandle.current().pid() + " started"); + + int fileId = Integer.parseInt(args[0]); + String uniqueId = args[1]; + String localPath = args[2]; + String downloadStatus = args[3]; + Long completionDate = Long.parseLong(args[4]); + + try { + Vertx vertx = Vertx.vertx(); + Future deployVerticle = vertx.deployVerticle(new DataVerticle()); + MessyUtils.await(deployVerticle); + + Future future = DataVerticle.fileRepository.updateDownloadStatus(fileId, + uniqueId, + localPath, + FileRecord.DownloadStatus.valueOf(downloadStatus), + completionDate); + MessyUtils.await(future); + System.out.println("Process: " + ProcessHandle.current().pid() + " finished"); + System.exit(0); + } catch (Exception e) { + System.out.println("Process: " + ProcessHandle.current().pid() + " failed"); + e.printStackTrace(); + System.exit(1); + } + } + +}