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

✨ feat: Log adjustment, AutoDownloadVerticle runs with VIRTUAL_THREAD.

Browse source
This commit is contained in:
jarvis2f 2024-12-25 11:31:23 +08:00
parent a111795d64
commit dd19756efb
13 changed files with 99 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.env.example
View file

@ -1,4 +1,5 @@
# Telegram files log level, ALL < FINEST < FINER < FINE < CONFIG < INFO < WARNING < SEVERE < OFF
LOG_LEVEL=INFO
# The root directory of the application
APP_ENV=dev
APP_ROOT=
@ -6,3 +7,5 @@ TDLIB_PATH=
TELEGRAM_API_ID=
TELEGRAM_API_HASH=
# Value 0 corresponds to fatal errors, value 1 corresponds to errors, value 2 corresponds to warnings and debug warnings, value 3 corresponds to informational, value 4 corresponds to debug, value 5 corresponds to verbose debug, value greater than 5 and up to 1023 can be used to enable even more logging.
TELEGRAM_LOG_LEVEL=0

71
api/src/main/java/telegram/files/AutoDownloadVerticle.java
View file

@ -11,6 +11,7 @@ import io.vertx.core.Promise;
import io.vertx.core.json.Json;
import io.vertx.core.json.JsonObject;
import org.drinkless.tdlib.TdApi;
import telegram.files.repository.FileRecord;
import telegram.files.repository.SettingAutoRecords;
import telegram.files.repository.SettingKey;
@ -27,7 +28,7 @@ public class AutoDownloadVerticle extends AbstractVerticle {
private static final int HISTORY_SCAN_INTERVAL = 2 * 60 * 1000;
private static final int MAX_HISTORY_SCAN_TIME = 20 * 1000;
private static final int MAX_HISTORY_SCAN_TIME = 10 * 1000;
private static final int MAX_WAITING_LENGTH = 30;
@ -107,7 +108,7 @@ public class AutoDownloadVerticle extends AbstractVerticle {
this.limit = Convert.toInt(message.body());
});
vertx.eventBus().consumer(EventEnum.MESSAGE_RECEIVED.address(), message -> {
log.debug("Auto download message received: %s".formatted(message.body()));
log.trace("Auto download message received: %s".formatted(message.body()));
this.onNewMessage((JsonObject) message.body());
});
return Future.succeededFuture();
@ -128,7 +129,9 @@ public class AutoDownloadVerticle extends AbstractVerticle {
}
private void addHistoryMessage(SettingAutoRecords.Item auto, long currentTimeMillis) {
log.debug("Start scan history! TelegramId: %d ChatId: %d FileType: %s".formatted(auto.telegramId, auto.chatId, auto.nextFileType));
if (System.currentTimeMillis() - currentTimeMillis > MAX_HISTORY_SCAN_TIME || isExceedLimit(auto.telegramId)) {
log.debug("Scan history end! TelegramId: %d ChatId: %d".formatted(auto.telegramId, auto.chatId));
return;
}
if (StrUtil.isBlank(auto.nextFileType)) {
@ -140,36 +143,39 @@ public class AutoDownloadVerticle extends AbstractVerticle {
searchChatMessages.fromMessageId = auto.nextFromMessageId;
searchChatMessages.limit = Math.min(MAX_WAITING_LENGTH, 100);
searchChatMessages.filter = TdApiHelp.getSearchMessagesFilter(auto.nextFileType);
telegramVerticle.execute(searchChatMessages)
.onSuccess(foundChatMessages -> {
if (foundChatMessages.messages.length == 0) {
int nextTypeIndex = FILE_TYPE_ORDER.indexOf(auto.nextFileType) + 1;
if (nextTypeIndex < FILE_TYPE_ORDER.size()) {
String originalType = auto.nextFileType;
auto.nextFileType = FILE_TYPE_ORDER.get(nextTypeIndex);
auto.nextFromMessageId = 0;
log.debug("%s No more %s files found! Switch to %s".formatted(auto.uniqueKey(), originalType, auto.nextFileType));
TdApi.FoundChatMessages foundChatMessages = Future.await(telegramVerticle.execute(searchChatMessages)
.onFailure(r -> log.error("Search chat messages failed! TelegramId: %d ChatId: %d".formatted(auto.telegramId, auto.chatId), r))
);
if (foundChatMessages == null) {
return;
}
if (foundChatMessages.messages.length == 0) {
int nextTypeIndex = FILE_TYPE_ORDER.indexOf(auto.nextFileType) + 1;
if (nextTypeIndex < FILE_TYPE_ORDER.size()) {
String originalType = auto.nextFileType;
auto.nextFileType = FILE_TYPE_ORDER.get(nextTypeIndex);
auto.nextFromMessageId = 0;
log.debug("%s No more %s files found! Switch to %s".formatted(auto.uniqueKey(), originalType, auto.nextFileType));
addHistoryMessage(auto, currentTimeMillis);
} else {
log.debug("%s No more history files found! TelegramId: %d ChatId: %d".formatted(auto.uniqueKey(), auto.telegramId, auto.chatId));
}
} else {
DataVerticle.fileRepository.getFilesByUniqueId(TdApiHelp.getFileUniqueIds(Arrays.asList(foundChatMessages.messages)))
.onSuccess(existFiles -> {
List<TdApi.Message> messages = Stream.of(foundChatMessages.messages)
.filter(message -> !existFiles.containsKey(TdApiHelp.getFileUniqueId(message)))
.toList();
if (CollUtil.isEmpty(messages)) {
auto.nextFromMessageId = foundChatMessages.nextFromMessageId;
addHistoryMessage(auto, currentTimeMillis);
} else if (addWaitingDownloadMessages(auto.telegramId, messages, false)) {
auto.nextFromMessageId = foundChatMessages.nextFromMessageId;
} else {
log.debug("%s No more history files found! TelegramId: %d ChatId: %d".formatted(auto.uniqueKey(), auto.telegramId, auto.chatId));
addHistoryMessage(auto, currentTimeMillis);
}
} else {
DataVerticle.fileRepository.getFilesByUniqueId(TdApiHelp.getFileUniqueIds(Arrays.asList(foundChatMessages.messages)))
.onSuccess(existFiles -> {
List<TdApi.Message> messages = Stream.of(foundChatMessages.messages)
.filter(message -> !existFiles.containsKey(TdApiHelp.getFileUniqueId(message)))
.toList();
if (CollUtil.isEmpty(messages)) {
auto.nextFromMessageId = foundChatMessages.nextFromMessageId;
addHistoryMessage(auto, currentTimeMillis);
} else if (addWaitingDownloadMessages(auto.telegramId, messages, false)) {
auto.nextFromMessageId = foundChatMessages.nextFromMessageId;
} else {
addHistoryMessage(auto, currentTimeMillis);
}
});
}
});
});
}
}
private boolean isExceedLimit(long telegramId) {
@ -178,11 +184,8 @@ public class AutoDownloadVerticle extends AbstractVerticle {
}
private int getSurplusSize(long telegramId) {
TelegramVerticle telegramVerticle = this.getTelegramVerticle(telegramId);
Integer downloading = telegramVerticle.getDownloadStatistics()
.map(statistics -> statistics.getInteger("downloading"))
.result();
return downloading == null ? limit : Math.min(0, limit - downloading);
Integer downloading = Future.await(DataVerticle.fileRepository.countByStatus(telegramId, FileRecord.DownloadStatus.downloading));
return downloading == null ? limit : Math.max(0, limit - downloading);
}
private boolean addWaitingDownloadMessages(long telegramId, List<TdApi.Message> messages, boolean force) {

17
api/src/main/java/telegram/files/Config.java
View file

@ -5,8 +5,12 @@ import cn.hutool.core.io.FileUtil;
import cn.hutool.core.util.StrUtil;
import java.io.File;
import java.util.logging.Level;
import java.util.logging.Logger;
public class Config {
public static final String LOG_LEVEL = StrUtil.blankToDefault(System.getenv("LOG_LEVEL"), "INFO");
public static final String APP_ENV = StrUtil.blankToDefault(System.getenv("APP_ENV"), "prod");
public static final String APP_ROOT = System.getenv("APP_ROOT");
@ -17,6 +21,10 @@ public class Config {
public static final String TELEGRAM_API_HASH = System.getenv("TELEGRAM_API_HASH");
public static final int TELEGRAM_LOG_LEVEL = Convert.toInt(System.getenv("TELEGRAM_LOG_LEVEL"), 0);
private static Level logLevel;
static {
if (APP_ENV == null) {
throw new RuntimeException("APP_ENV is not set");
@ -31,6 +39,15 @@ public class Config {
throw new RuntimeException("TELEGRAM_API_HASH is not set");
}
try {
logLevel = Level.parse(Config.LOG_LEVEL);
System.out.println("Setting telegram.files log level to " + logLevel);
} catch (IllegalArgumentException e) {
System.err.println("Invalid log level [" + Config.LOG_LEVEL + "], using default INFO.");
}
Logger.getLogger("telegram.files").setLevel(logLevel);
if (!FileUtil.exist(APP_ROOT)) {
FileUtil.mkdir(APP_ROOT);
}

4
api/src/main/java/telegram/files/HttpVerticle.java
View file

@ -196,7 +196,9 @@ public class HttpVerticle extends AbstractVerticle {
}
public Future<Void> initAutoDownloadVerticle() {
return vertx.deployVerticle(new AutoDownloadVerticle(), new DeploymentOptions().setThreadingModel(ThreadingModel.WORKER))
return vertx.deployVerticle(new AutoDownloadVerticle(), new DeploymentOptions()
.setThreadingModel(ThreadingModel.VIRTUAL_THREAD)
)
.mapEmpty();
}

2
api/src/main/java/telegram/files/TelegramUpdateHandler.java
View file

@ -30,7 +30,7 @@ public class TelegramUpdateHandler implements Client.ResultHandler {
if (onFileUpdated != null)
onFileUpdated.accept((TdApi.UpdateFile) object);
case TdApi.UpdateFileDownload.CONSTRUCTOR:
log.debug("File download update: %s".formatted(object));
log.trace("File download update: %s".formatted(object));
break;
case TdApi.UpdateFileDownloads.CONSTRUCTOR:
if (onFileDownloadsUpdated != null)

10
api/src/main/java/telegram/files/TelegramVerticle.java
View file

@ -56,7 +56,7 @@ public class TelegramVerticle extends AbstractVerticle {
Client.setLogMessageHandler(0, new LogMessageHandler());
try {
Client.execute(new TdApi.SetLogVerbosityLevel(0));
Client.execute(new TdApi.SetLogVerbosityLevel(Config.TELEGRAM_LOG_LEVEL));
Client.execute(new TdApi.SetLogStream(new TdApi.LogStreamFile("tdlib.log", 1 << 27, false)));
} catch (Client.ExecutionException error) {
throw new IOError(new IOException("Write access to the current directory is required"));
@ -600,7 +600,7 @@ public class TelegramVerticle extends AbstractVerticle {
request.systemLanguageCode = "en";
request.deviceModel = "Telegram Files";
request.applicationVersion = Start.VERSION;
log.debug("[%s] Send SetTdlibParameters: %s".formatted(getRootId(), request));
log.trace("[%s] Send SetTdlibParameters: %s".formatted(getRootId(), request));
client.send(request, this::handleAuthorizationResult);
break;
@ -650,7 +650,7 @@ public class TelegramVerticle extends AbstractVerticle {
}
private void onFileUpdated(TdApi.UpdateFile updateFile) {
log.debug("[%s] Receive file update: %s".formatted(getRootId(), updateFile));
log.trace("[%s] Receive file update: %s".formatted(getRootId(), updateFile));
TdApi.File file = updateFile.file;
if (file != null) {
String localPath = null;
@ -675,12 +675,12 @@ public class TelegramVerticle extends AbstractVerticle {
}
private void onFileDownloadsUpdated(TdApi.UpdateFileDownloads updateFileDownloads) {
log.debug("[%s] Receive file downloads update: %s".formatted(getRootId(), updateFileDownloads));
log.trace("[%s] Receive file downloads update: %s".formatted(getRootId(), updateFileDownloads));
sendHttpEvent(EventPayload.build(EventPayload.TYPE_FILE_DOWNLOAD, updateFileDownloads));
}
private void onMessageReceived(TdApi.Message message) {
log.debug("[%s] Receive message: %s".formatted(getRootId(), message));
log.trace("[%s] Receive message: %s".formatted(getRootId(), message));
vertx.eventBus().publish(EventEnum.MESSAGE_RECEIVED.address(), JsonObject.of()
.put("telegramId", telegramRecord.id())
.put("chatId", message.chatId)

2
api/src/main/java/telegram/files/repository/FileRepository.java
View file

@ -25,6 +25,8 @@ public interface FileRepository {
Future<JsonObject> getDownloadStatistics(long telegramId);
Future<Integer> countByStatus(long telegramId, FileRecord.DownloadStatus downloadStatus);
Future<JsonObject> updateStatus(int fileId,
String uniqueId,
String localPath,

16
api/src/main/java/telegram/files/repository/impl/FileRepositoryImpl.java
View file

@ -42,7 +42,7 @@ public class FileRepositoryImpl implements FileRepository {
.execute()
.onComplete(r -> conn.close())
.onFailure(err -> log.error("Failed to create table file_record: %s".formatted(err.getMessage())))
.onSuccess(ps -> log.debug("Successfully created table: file_record"))
.onSuccess(ps -> log.trace("Successfully created table: file_record"))
)
.mapEmpty();
}
@ -61,7 +61,7 @@ public class FileRepositoryImpl implements FileRepository {
.mapFrom(FileRecord.PARAM_MAPPER)
.execute(fileRecord)
.map(r -> fileRecord)
.onSuccess(r -> log.debug("Successfully created file record: %s".formatted(fileRecord.id()))
.onSuccess(r -> log.trace("Successfully created file record: %s".formatted(fileRecord.id()))
)
.onFailure(
err -> log.error("Failed to create file record: %s".formatted(err.getMessage()))
@ -234,6 +234,18 @@ public class FileRepositoryImpl implements FileRepository {
.onFailure(err -> log.error("Failed to get download statistics: %s".formatted(err.getMessage())));
}
@Override
public Future<Integer> countByStatus(long telegramId, FileRecord.DownloadStatus downloadStatus) {
return SqlTemplate
.forQuery(pool, """
SELECT COUNT(*) FROM file_record WHERE telegram_id = #{telegramId} AND download_status = #{downloadStatus}
""")
.mapTo(rs -> rs.getInteger(0))
.execute(Map.of("telegramId", telegramId, "downloadStatus", downloadStatus.name()))
.map(rs -> rs.size() > 0 ? rs.iterator().next() : 0)
.onFailure(err -> log.error("Failed to count file record: %s".formatted(err.getMessage())));
}
@Override
public Future<JsonObject> updateStatus(int fileId, String uniqueId, String localPath, FileRecord.DownloadStatus downloadStatus) {
if (StrUtil.isBlank(localPath) && downloadStatus == null) {

8
api/src/main/java/telegram/files/repository/impl/SettingRepositoryImpl.java
View file

@ -35,7 +35,7 @@ public class SettingRepositoryImpl implements SettingRepository {
.execute()
.onComplete(r -> conn.close())
.onFailure(err -> log.error("Failed to create table setting_record: %s".formatted(err.getMessage())))
.onSuccess(ps -> log.debug("Successfully created table: setting_record"))
.onSuccess(ps -> log.trace("Successfully created table: setting_record"))
)
.mapEmpty();
}
@ -50,7 +50,7 @@ public class SettingRepositoryImpl implements SettingRepository {
.mapFrom(SettingRecord.PARAM_MAPPER)
.execute(new SettingRecord(key, value))
.map(r -> new SettingRecord(key, value))
.onSuccess(r -> log.debug("Successfully created or updated setting record: %s".formatted(key)))
.onSuccess(r -> log.trace("Successfully created or updated setting record: %s".formatted(key)))
.onFailure(
err -> log.error("Failed to create or update setting record: %s".formatted(err.getMessage()))
);
@ -74,7 +74,7 @@ public class SettingRepositoryImpl implements SettingRepository {
.mapTo(SettingRecord.ROW_MAPPER)
.execute(Collections.emptyMap())
.map(IterUtil::toList)
.onSuccess(r -> log.debug("Successfully fetched setting record for keys: " + keyStr))
.onSuccess(r -> log.trace("Successfully fetched setting record for keys: " + keyStr))
.onFailure(
err -> log.error("Failed to fetch setting record: %s".formatted(err.getMessage()))
);
@ -95,7 +95,7 @@ public class SettingRepositoryImpl implements SettingRepository {
}
return key.defaultValue == null ? null : (T) key.defaultValue;
})
.onSuccess(r -> log.debug("Successfully fetched setting record for key: " + key))
.onSuccess(r -> log.trace("Successfully fetched setting record for key: " + key))
.onFailure(
err -> log.error("Failed to fetch setting record: %s".formatted(err.getMessage()))
);

4
api/src/main/java/telegram/files/repository/impl/TelegramRepositoryImpl.java
View file

@ -34,7 +34,7 @@ public class TelegramRepositoryImpl implements TelegramRepository {
.execute()
.onComplete(r -> conn.close())
.onFailure(err -> log.error("Failed to create table telegram_record: %s".formatted(err.getMessage())))
.onSuccess(ps -> log.debug("Successfully created table: telegram_record"))
.onSuccess(ps -> log.trace("Successfully created table: telegram_record"))
)
.mapEmpty();
}
@ -51,7 +51,7 @@ public class TelegramRepositoryImpl implements TelegramRepository {
.mapFrom(TelegramRecord.PARAM_MAPPER)
.execute(telegramRecord)
.map(r -> telegramRecord)
.onSuccess(r -> log.debug("Successfully created telegram record: %s".formatted(telegramRecord.id())))
.onSuccess(r -> log.trace("Successfully created telegram record: %s".formatted(telegramRecord.id())))
.onFailure(
err -> log.error("Failed to create telegram record: %s".formatted(err.getMessage()))
);

2
api/src/main/resources/logging.properties
View file

@ -5,7 +5,7 @@
handlers = java.util.logging.ConsoleHandler, java.util.logging.FileHandler
# \u63A7\u5236\u53F0 Handler \u7684\u65E5\u5FD7\u7EA7\u522B
java.util.logging.ConsoleHandler.level = INFO
java.util.logging.ConsoleHandler.level = FINE
# \u63A7\u5236\u53F0 Handler \u7684\u65E5\u5FD7\u683C\u5F0F\u5316\u5668
java.util.logging.ConsoleHandler.formatter = java.util.logging.SimpleFormatter

6
api/src/test/java/telegram/files/DataVerticleTest.java
View file

@ -40,7 +40,7 @@ public class DataVerticleTest {
@Test
@DisplayName("Test Create telegram record")
void createTelegramRecordTest(Vertx vertx, VertxTestContext testContext) {
TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test");
TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test", null);
DataVerticle.telegramRepository.create(telegramRecord)
.compose(r -> DataVerticle.telegramRepository.getById(r.id()))
.onComplete(testContext.succeeding(r -> testContext.verify(() -> {
@ -52,8 +52,8 @@ public class DataVerticleTest {
@Test
@DisplayName("Test Get all telegram record")
void getAllTelegramRecordTest(Vertx vertx, VertxTestContext testContext) {
TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test");
TelegramRecord telegramRecord2 = new TelegramRecord(2, "test2", "test2");
TelegramRecord telegramRecord = new TelegramRecord(1, "test", "test", null);
TelegramRecord telegramRecord2 = new TelegramRecord(2, "test2", "test2", null);
DataVerticle.telegramRepository.create(telegramRecord)
.compose(r -> DataVerticle.telegramRepository.create(telegramRecord2))
.compose(r -> DataVerticle.telegramRepository.getAll())

8
web/src/components/file-statistics.tsx
View file

@ -8,6 +8,7 @@ import {
File,
FileText,
Image,
LoaderPinwheel,
Music,
Network,
PauseCircle,
@ -76,7 +77,10 @@ const FileStatistics: React.FC<FileStatisticsProps> = ({ telegramId }) => {
if (!data) {
return (
<div className="flex items-center space-x-2 rounded-lg bg-white p-4 text-gray-600 shadow-md">
<Download className="h-5 w-5 animate-spin" />
<LoaderPinwheel
className="h-5 w-5 animate-spin"
style={{ strokeWidth: "0.8px" }}
/>
<span>Loading...</span>
</div>
);
@ -121,7 +125,7 @@ const FileStatistics: React.FC<FileStatisticsProps> = ({ telegramId }) => {
];
return (
<div className="space-y-6 rounded-lg bg-gray-50 p-6">
<div className="space-y-6 rounded-lg bg-gray-50 p-2 md:p-6">
<div className="flex-1 rounded-lg bg-white p-4 shadow-md">
<div className="flex items-center space-x-3 border-gray-200">
<h3 className="text-md flex items-center space-x-2 font-semibold text-gray-700">