danvics/telegram-files
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

✅ chore: Add concurrent update file download status test

Browse source
This commit is contained in:
jarvis2f 2025-02-05 23:25:05 +08:00
parent 56997142d6
commit b4ace30cc1
4 changed files with 234 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.env.example
View file

@ -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=

14
api/src/main/java/telegram/files/MessyUtils.java
View file

@ -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> T await(Future<T> future) {
CompletableFuture<T> completableFuture = new CompletableFuture<>();
future.onComplete(result -> {
if (result.succeeded()) {
completableFuture.complete(result.result());
} else {
completableFuture.completeExceptionally(result.cause());
}
});
return completableFuture.join();
}
}

175
api/src/test/java/telegram/files/FileDownloadStatusConcurrentTest.java Normal file
View file

@ -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<String> future = vertx.deployVerticle(new DataVerticle());
MessyUtils.await(future);
Future<FileRecord> 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<JsonObject> future = DataVerticle.fileRepository.updateDownloadStatus(newFileId, fileRecord.uniqueId(), updateLocalPath, FileRecord.DownloadStatus.downloading, completionDate);
MessyUtils.await(future);
Future<FileRecord> 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<java.util.concurrent.Future<Void>> 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<Void> future : futures) {
future.get();
}
while (completedCount.get() < THREAD_COUNT) {
Thread.sleep(100);
}
Future<FileRecord> 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<Process> 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<FileRecord> 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"
);
}
}

44
api/src/test/java/telegram/files/UpdateFileDownloadProcess.java Normal file
View file

@ -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 <fileId> <uniqueId> <localPath> <downloadStatus> <completionDate>");
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<String> deployVerticle = vertx.deployVerticle(new DataVerticle());
MessyUtils.await(deployVerticle);
Future<JsonObject> 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);
}
}
}