Compare commits
10
Commits
fb327ba00c
...
f5ef8ca487
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f5ef8ca487 | ||
|
|
50e513b8cc | ||
|
|
d3b18f90fa | ||
|
|
a9ba631847 | ||
|
|
60facae9b5 | ||
|
|
b15eeaf45e | ||
|
|
1e6e3a1557 | ||
|
|
7f823b6150 | ||
|
|
f65c5ad860 | ||
|
|
3c8be3e7c9 |
@@ -0,0 +1,64 @@
|
|||||||
|
# 2026-09-08 审查修复与编译交接
|
||||||
|
|
||||||
|
本批源码已在本机按问题分别提交,尚未推送、后端编译或部署。编译机恢复后,需要先取得下列仓库的提交,再按现有发布手册构建、验证与上线。此次没有调整鉴权或数据库表结构。
|
||||||
|
|
||||||
|
## 提交清单
|
||||||
|
|
||||||
|
| 仓库 | 提交 | 内容 |
|
||||||
|
| --- | --- | --- |
|
||||||
|
| storageNode | cbfd634 | 压缩临时文件、ZIP 完整性校验、失败保留源目录并恢复可重试状态 |
|
||||||
|
| lionwebsite-backend | f65c5ad | 下发节点前持久化任务,超时保留记录,原子累加用量 |
|
||||||
|
| lionwebsite-backend | 7f823b6 | 修复 Range 解析、读取长度、空文件和越界处理 |
|
||||||
|
| lionwebsite-backend | 1e6e3a1 | 上游订阅下载移出全局锁,设置超时,拒绝旧配置/旧请求的刷新结果 |
|
||||||
|
| lionwebsite-backend | b15eeaf | 复用连接线程组,登记等待对象后再发送,统一超时/失败/关闭清理 |
|
||||||
|
| lionwebsite-backend | 60facae | 关闭图片和 TAR 文件流,合并同图请求,完成后发布缓存,检查转换结果 |
|
||||||
|
| lionwebsite-frontend-desktop | 287dc09 | 图片失败继续加载,单图重试,忽略旧页面图片事件 |
|
||||||
|
| lionwebsite-frontend-desktop | b0f832c | 退避重连、连接状态提示、唤醒后重连、重连与提交后的全量刷新 |
|
||||||
|
| lionwebsite-frontend-desktop | d0eb2f0 | 链接 GID 类型匹配,补充任务提交回归测试 |
|
||||||
|
| lionwebsite-backend | a9ba631 | 连接反复断开时重新安排监听,验证缓存失败后可重试 |
|
||||||
|
| lionwebsite-backend | d3b18f9 | 修复图片 Key 缓存递归更新与 null 写入,异常时关闭 HTTP 响应 |
|
||||||
|
|
||||||
|
各仓库分别保留了原有历史。前端原先未提交的 `index.html`、`vite.config.js`、`llm_readme.md` 以及未跟踪的 `dist/`、`node_modules/` 未纳入本批提交。
|
||||||
|
|
||||||
|
## 已验证与待验证
|
||||||
|
|
||||||
|
已完成:
|
||||||
|
|
||||||
|
- 前端 `node --test tests/*.test.mjs`:9 项通过。
|
||||||
|
- 前端生产构建:成功,产物输出到 `/tmp/lionwebsite-review-build`,没有覆盖现有 `dist/` 或线上资源。构建提示主包体积超过 500 kB,本批未做依赖拆包。
|
||||||
|
- 各次提交的 `git diff --check` 与最终源码调用链检查。
|
||||||
|
|
||||||
|
尚未执行:
|
||||||
|
|
||||||
|
- 主站、存储节点的 Java 编译及 JUnit 测试。
|
||||||
|
- GraalVM 原生构建、真实节点断连、真实上游下载和生产端到端验证。
|
||||||
|
|
||||||
|
**以下 Java 测试仅在编译机的独立工作副本执行,不在生产主站或存储机执行。两个仓库依次运行 `mvn test`,通过后再按现有部署手册构建原生程序。** 新增测试均使用临时目录、Mock 或内存通道,不需要生产数据库和真实上游。
|
||||||
|
|
||||||
|
主站新增测试:`GallerySubmissionTest`、`FileDownloadTest`、`SubscriptionRefreshServiceTest`、`RemoteServiceTest`、`PersonalArchiveTest`、`ImageFileCacheTest`、`GalleryKeyCacheTest`。
|
||||||
|
|
||||||
|
存储节点新增测试:`DownloadCheckServiceTest`。两个仓库已有的订阅快照测试也应一起运行。
|
||||||
|
|
||||||
|
## 行为与接口说明
|
||||||
|
|
||||||
|
HTTP 路由、参数名和节点消息格式保持兼容,任务状态继续使用 `已提交`、`下载中`、`等待压缩`、`压缩中`、`下载完成`。
|
||||||
|
|
||||||
|
- `POST /GalleryManage`:先保存任务再下发。节点未确认时返回业务 failure,但任务记录仍在,前端刷新后可以重试;不会把已经收到的完成状态写回已提交。新增任务时即计入本周用量,重试不重复计量。
|
||||||
|
- 存储节点压缩失败:保留源文件,回到等待压缩状态,30 秒后可自动重试;手动重试会取消等待。仅完整校验通过的 ZIP 才视为完成。
|
||||||
|
- 使用 `FileDownload.export` 的文件接口:支持单个普通范围、后缀范围和开放结尾范围;非法或不可满足范围返回 416,`Content-Range: bytes */<size>`;多个范围回退为完整 200 响应;HEAD 不输出正文;普通完整响应不再带 Content-Range。
|
||||||
|
- 图片缓存:同一图片的并发请求共享下载,临时文件关闭且转换成功后才发布。下载连接超时 5 秒、读取超时 15 秒,转换超时 60 秒。图片缺失或下载失败仍返回 404,中断可返回 503。
|
||||||
|
- 订阅刷新:连接和连接池等待超时 5 秒,读取超时 15 秒;网络操作不占用订阅状态锁。上游 Key、过滤选项或启用状态改变后会使旧缓存失效,旧刷新结果不能重新发布。仅名称修改且刷新失败时保留原有缓存。
|
||||||
|
- WebSocket:连接失败按 1/2/4/8/16/30 秒退避,建立连接超时 10 秒;重连后重新读取任务列表和用量。页面恢复可见或网络恢复时重新连接。
|
||||||
|
|
||||||
|
## 上线验证重点
|
||||||
|
|
||||||
|
按现有发布手册安排存储节点、主站和前端上线,避开正在运行的生产任务。源码修复本身不代表生产已经更新。
|
||||||
|
|
||||||
|
1. 新建任务,模拟节点已有归档或立即返回状态,确认数据库与页面均保留完成状态且没有重复行。
|
||||||
|
2. 使用临时测试目录制造一次压缩失败,确认源文件保留,恢复目录可写后能重试,损坏 ZIP 不会标记完成。
|
||||||
|
3. 下载完整文件和小范围/尾部范围,确认响应字节数;验证空文件和越界请求。
|
||||||
|
4. 在测试上游阻塞刷新时修改其他账号/读取快照,确认操作不被网络等待阻塞;旧请求结束后不能覆盖新配置。
|
||||||
|
5. 阅读器人为令一张图片失败,确认后续图片继续加载、单图重试有效。
|
||||||
|
6. 浏览器断网、恢复网络、切后台再返回,确认连接提示和进度能够恢复。
|
||||||
|
|
||||||
|
根目录 `/home/lionwebsite/API_DOCUMENTATION.md` 已追加待发布说明;本文件是这部分接口行为变更的仓库内交接记录。
|
||||||
@@ -88,7 +88,7 @@
|
|||||||
<dependency>
|
<dependency>
|
||||||
<groupId>commons-io</groupId>
|
<groupId>commons-io</groupId>
|
||||||
<artifactId>commons-io</artifactId>
|
<artifactId>commons-io</artifactId>
|
||||||
<version>2.11.0</version>
|
<version>2.15.1</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
|
|||||||
@@ -16,6 +16,9 @@ public interface CustomConfigurationMapper {
|
|||||||
// @Delete("delete from customConfiguration where parameter=#{parameter}")
|
// @Delete("delete from customConfiguration where parameter=#{parameter}")
|
||||||
// void deleteConfiguration(CustomConfiguration configuration);
|
// void deleteConfiguration(CustomConfiguration configuration);
|
||||||
|
|
||||||
|
@Update("update customConfiguration set value=cast(value as integer)+#{amount} where parameter=#{parameter}")
|
||||||
|
void incrementConfiguration(@Param("parameter") String parameter, @Param("amount") long amount);
|
||||||
|
|
||||||
@Select("select * from customConfiguration where parameter=#{parameter}")
|
@Select("select * from customConfiguration where parameter=#{parameter}")
|
||||||
CustomConfiguration selectConfiguration(String parameter);
|
CustomConfiguration selectConfiguration(String parameter);
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,6 +7,8 @@ import com.lion.lionwebsite.Exception.ResolutionNotMatchException;
|
|||||||
import com.lion.lionwebsite.Error.ErrorCode;
|
import com.lion.lionwebsite.Error.ErrorCode;
|
||||||
import com.lion.lionwebsite.Util.CustomUtil;
|
import com.lion.lionwebsite.Util.CustomUtil;
|
||||||
import com.lion.lionwebsite.Util.FileDownload;
|
import com.lion.lionwebsite.Util.FileDownload;
|
||||||
|
import com.lion.lionwebsite.Util.ImageFileCache;
|
||||||
|
import java.nio.file.Path;
|
||||||
import com.lion.lionwebsite.Util.GalleryUtil;
|
import com.lion.lionwebsite.Util.GalleryUtil;
|
||||||
import com.lion.lionwebsite.Util.Response;
|
import com.lion.lionwebsite.Util.Response;
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
@@ -93,9 +95,14 @@ public class GalleryManageService {
|
|||||||
} else {
|
} else {
|
||||||
taskName = gallery.getName();
|
taskName = gallery.getName();
|
||||||
log.info("创建任务: {} 目标分辨率:{}", link, targetResolution);
|
log.info("创建任务: {} 目标分辨率:{}", link, targetResolution);
|
||||||
|
// Persist before dispatch: the node sends its current status before its ACK.
|
||||||
|
gallery.setDownloader(user.getId());
|
||||||
|
gallery.set_download(true);
|
||||||
|
galleryMapper.insertGallery(gallery);
|
||||||
|
configurationMapper.incrementConfiguration(CustomConfiguration.WEEK_USED_AMOUNT, gallery.getFileSize());
|
||||||
if (remoteService.addGalleryToQueue(gallery) != 0) {
|
if (remoteService.addGalleryToQueue(gallery) != 0) {
|
||||||
log.error("传送任务{}失败, 未知原因", gallery.getName());
|
log.error("传送任务{}失败, 未知原因", gallery.getName());
|
||||||
response.failure("任务传送失败,未知原因,尝试点击重连按钮看看");
|
response.failure("任务已保存,但节点未确认接收;请刷新任务列表后重试");
|
||||||
pushService.taskCreateReport(user.getUsername(), taskName, response);
|
pushService.taskCreateReport(user.getUsername(), taskName, response);
|
||||||
return response.toJSONString();
|
return response.toJSONString();
|
||||||
}
|
}
|
||||||
@@ -111,20 +118,9 @@ public class GalleryManageService {
|
|||||||
return response.toJSONString();
|
return response.toJSONString();
|
||||||
}
|
}
|
||||||
|
|
||||||
//处理下载结果,将任务插入数据库并且更新每周用量
|
// Do not overwrite an immediate node status with the original submitted state.
|
||||||
if (gallery.getStatus().equals("已提交")) {
|
Gallery current = galleryMapper.selectGalleryByGid(gallery.getGid());
|
||||||
response.success(gallery.toString());
|
response.success((current == null ? gallery : current).toString());
|
||||||
gallery.setDownloader(user.getId());
|
|
||||||
gallery.set_download(true);
|
|
||||||
galleryMapper.insertGallery(gallery);
|
|
||||||
|
|
||||||
long usedAmount = Long.parseLong(configurationMapper.selectConfiguration(CustomConfiguration.WEEK_USED_AMOUNT).getValue());
|
|
||||||
usedAmount += gallery.getFileSize();
|
|
||||||
configurationMapper.updateConfiguration(CustomConfiguration.WEEK_USED_AMOUNT, String.valueOf(usedAmount));
|
|
||||||
} else {
|
|
||||||
response.failure("提交失败,未知原因");
|
|
||||||
galleryMapper.deleteGalleryByGid(gallery.getGid());
|
|
||||||
}
|
|
||||||
pushService.taskCreateReport(user.getUsername(), taskName, response);
|
pushService.taskCreateReport(user.getUsername(), taskName, response);
|
||||||
|
|
||||||
return response.toJSONString();
|
return response.toJSONString();
|
||||||
@@ -352,75 +348,38 @@ public class GalleryManageService {
|
|||||||
return response.toJSONString();
|
return response.toJSONString();
|
||||||
}
|
}
|
||||||
|
|
||||||
String[] suffixes = {".avif", ".gif"};
|
|
||||||
public Callable<?> getCachedImage(String gid, Integer page, HttpServletRequest request, HttpServletResponse response) {
|
public Callable<?> getCachedImage(String gid, Integer page, HttpServletRequest request, HttpServletResponse response) {
|
||||||
//检查文件夹是否存在
|
Path directory = Path.of(cachePath, gid);
|
||||||
File folder = new File(cachePath + gid);
|
String name = String.valueOf(page);
|
||||||
if(!folder.isDirectory())
|
Path cached = ImageFileCache.find(directory, name);
|
||||||
folder.mkdirs();
|
if (cached != null) {
|
||||||
|
FileDownload.export(request, response, cached.toString());
|
||||||
//检查对应图片是否存在,存在则直接返回
|
return null;
|
||||||
for (String suffix : suffixes) {
|
|
||||||
if(new File(cachePath + gid + "/" + page + suffix).exists()){
|
|
||||||
FileDownload.export(request, response, cachePath + gid + "/" + page + suffix);
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
//检查该图片缓存是否存在
|
|
||||||
GidToKey gidToKey = imageCacheMapper.selectKeyByGid(gid);
|
|
||||||
if(gidToKey == null)
|
|
||||||
try {
|
|
||||||
log.error("未缓存gid:{}", gid);
|
|
||||||
response.sendError(404);
|
|
||||||
return null;
|
|
||||||
}catch (IOException e){
|
|
||||||
log.warn("sendError 404 failed", e);
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
return () -> {
|
return () -> {
|
||||||
if(response.isCommitted()) {
|
if (response.isCommitted()) return null;
|
||||||
log.info("连接已关闭: gid={} page={}", gid, page);
|
try {
|
||||||
return null;
|
Path image = ImageFileCache.get(directory, name, () -> {
|
||||||
}
|
GidToKey gidToKey = imageCacheMapper.selectKeyByGid(gid);
|
||||||
|
ImageKeyCache imageKey = imageCacheMapper.selectImageKeyCacheByGidAndPage(gid, page);
|
||||||
String imageUrl = null;
|
if (gidToKey == null || imageKey == null)
|
||||||
//获取该图片key
|
throw new IOException("图片索引不存在");
|
||||||
ImageKeyCache imageKeyCache = imageCacheMapper.selectImageKeyCacheByGidAndPage(gid, page);
|
for (int attempt = 0; attempt < 2; attempt++) {
|
||||||
if (imageKeyCache == null) {
|
String url = GalleryUtil.getImageUrl(getMpvKey(gidToKey.toUrl()), imageKey);
|
||||||
CustomUtil.fourZeroFour(response);
|
if (url != null) return url;
|
||||||
return null;
|
GalleryUtil.refreshMpvKey(gidToKey.toUrl());
|
||||||
}
|
}
|
||||||
|
throw new IOException("无法获取图片地址");
|
||||||
//获取图片地址
|
});
|
||||||
for (int i = 0; i < 2; i++) {
|
FileDownload.export(request, response, image.toString());
|
||||||
imageUrl = GalleryUtil.getImageUrl(getMpvKey(gidToKey.toUrl()), imageKeyCache);
|
} catch (InterruptedException e) {
|
||||||
if (imageUrl != null)
|
Thread.currentThread().interrupt();
|
||||||
break;
|
if (!response.isCommitted()) response.sendError(503);
|
||||||
GalleryUtil.refreshMpvKey(gidToKey.toUrl());
|
} catch (Exception e) {
|
||||||
}
|
log.warn("获取在线图片失败: gid={} page={} errorType={}", gid, page, e.getClass().getSimpleName());
|
||||||
|
if (!response.isCommitted()) response.sendError(404);
|
||||||
if (imageUrl == null) {
|
}
|
||||||
CustomUtil.fourZeroFour(response);
|
return null;
|
||||||
log.error("获取图片url失败:gid={} page={} imageKey={}", gid, page, imageKeyCache.getImgkey());
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
|
|
||||||
//下载图片,转格式并返回
|
|
||||||
String suffix = imageUrl.substring(imageUrl.lastIndexOf("."));
|
|
||||||
String imagePath = cachePath + gid + "/" + page + suffix;
|
|
||||||
try {
|
|
||||||
new URI(imageUrl).toURL().openConnection().getInputStream().transferTo(new FileOutputStream(imagePath));
|
|
||||||
}catch (Exception e){
|
|
||||||
log.error("下载图片失败:url{}", imageUrl, e);
|
|
||||||
CustomUtil.fourZeroFour(response);
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
if (!suffix.equals(".gif"))
|
|
||||||
imagePath = GalleryUtil.convertImg(imagePath, suffix);
|
|
||||||
FileDownload.export(request, response, imagePath);
|
|
||||||
return null;
|
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -27,6 +27,7 @@ import org.springframework.web.multipart.MultipartFile;
|
|||||||
import java.io.BufferedOutputStream;
|
import java.io.BufferedOutputStream;
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
import java.io.IOException;
|
import java.io.IOException;
|
||||||
|
import java.io.InputStream;
|
||||||
import java.io.OutputStream;
|
import java.io.OutputStream;
|
||||||
import java.net.URLDecoder;
|
import java.net.URLDecoder;
|
||||||
import java.nio.charset.StandardCharsets;
|
import java.nio.charset.StandardCharsets;
|
||||||
@@ -303,39 +304,51 @@ public class PersonalService{
|
|||||||
}
|
}
|
||||||
|
|
||||||
compressThreadPool.submit(() -> {
|
compressThreadPool.submit(() -> {
|
||||||
try(OutputStream bos = new BufferedOutputStream(Files.newOutputStream(Paths.get(finalPath + ".tar***undone")));
|
Path temporary = Paths.get(finalPath + ".tar***undone");
|
||||||
TarArchiveOutputStream aos = new TarArchiveOutputStream(bos)) {
|
try {
|
||||||
aos.setLongFileMode(TarArchiveOutputStream.LONGFILE_GNU); //解除文件名长度限制
|
writeTar(Paths.get(finalPath), temporary);
|
||||||
Path dirPath = Paths.get(finalPath);
|
// writeTar closes the archive (including its trailer) before publication.
|
||||||
Files.walkFileTree(dirPath, new SimpleFileVisitor<>() {
|
try {
|
||||||
@Override
|
Files.move(temporary, Paths.get(finalPath + ".tar"), StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
|
||||||
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
|
} catch (AtomicMoveNotSupportedException e) {
|
||||||
TarArchiveEntry entry = new TarArchiveEntry(dir.toFile(), dirPath.relativize(dir).toString());
|
Files.move(temporary, Paths.get(finalPath + ".tar"), StandardCopyOption.REPLACE_EXISTING);
|
||||||
aos.putArchiveEntry(entry);
|
}
|
||||||
aos.closeArchiveEntry();
|
log.info("打包成功: {}.tar", finalPath);
|
||||||
return super.preVisitDirectory(dir, attrs);
|
} catch (IOException e) {
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
|
|
||||||
TarArchiveEntry entry = new TarArchiveEntry(file.toFile(), dirPath.relativize(file).toString());
|
|
||||||
aos.putArchiveEntry(entry);
|
|
||||||
IOUtils.copy(Files.newInputStream(file.toFile().toPath()), aos);
|
|
||||||
aos.closeArchiveEntry();
|
|
||||||
return super.visitFile(file, attrs);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
File targetFile = new File(finalPath + ".tar***undone");
|
|
||||||
log.info("打包成功,重命名:" + targetFile.renameTo(new File(finalPath + ".tar")));
|
|
||||||
}catch (IOException e){
|
|
||||||
log.error("打包失败", e);
|
log.error("打包失败", e);
|
||||||
log.info("打包失败,删除文件结果:" + new File(finalPath + ".tar***undone").delete());
|
} finally {
|
||||||
|
try { Files.deleteIfExists(temporary); }
|
||||||
|
catch (IOException e) { log.warn("清理打包临时文件失败", e); }
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
response.success("加入队列成功");
|
response.success("加入队列成功");
|
||||||
return response.toJSONString();
|
return response.toJSONString();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void writeTar(Path directory, Path output) throws IOException {
|
||||||
|
try (OutputStream stream = new BufferedOutputStream(Files.newOutputStream(output));
|
||||||
|
TarArchiveOutputStream archive = new TarArchiveOutputStream(stream)) {
|
||||||
|
archive.setLongFileMode(TarArchiveOutputStream.LONGFILE_GNU);
|
||||||
|
Files.walkFileTree(directory, new SimpleFileVisitor<>() {
|
||||||
|
@Override public FileVisitResult preVisitDirectory(Path path, BasicFileAttributes attrs) throws IOException {
|
||||||
|
if (!path.equals(directory)) {
|
||||||
|
archive.putArchiveEntry(new TarArchiveEntry(path.toFile(), directory.relativize(path).toString()));
|
||||||
|
archive.closeArchiveEntry();
|
||||||
|
}
|
||||||
|
return FileVisitResult.CONTINUE;
|
||||||
|
}
|
||||||
|
@Override public FileVisitResult visitFile(Path path, BasicFileAttributes attrs) throws IOException {
|
||||||
|
archive.putArchiveEntry(new TarArchiveEntry(path.toFile(), directory.relativize(path).toString()));
|
||||||
|
try (InputStream input = Files.newInputStream(path)) {
|
||||||
|
IOUtils.copy(input, archive);
|
||||||
|
}
|
||||||
|
archive.closeArchiveEntry();
|
||||||
|
return FileVisitResult.CONTINUE;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 删除文件
|
* 删除文件
|
||||||
* @param path 目标路径
|
* @param path 目标路径
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ package com.lion.lionwebsite.Service;
|
|||||||
import com.lion.lionwebsite.Domain.GalleryForQuery;
|
import com.lion.lionwebsite.Domain.GalleryForQuery;
|
||||||
import com.lion.lionwebsite.Util.FileDownload;
|
import com.lion.lionwebsite.Util.FileDownload;
|
||||||
import com.lion.lionwebsite.Util.GalleryUtil;
|
import com.lion.lionwebsite.Util.GalleryUtil;
|
||||||
|
import com.lion.lionwebsite.Util.ImageFileCache;
|
||||||
|
import java.nio.file.Path;
|
||||||
import com.lion.lionwebsite.Util.Response;
|
import com.lion.lionwebsite.Util.Response;
|
||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
import jakarta.servlet.ServletOutputStream;
|
import jakarta.servlet.ServletOutputStream;
|
||||||
@@ -103,22 +105,16 @@ public class QueryService {
|
|||||||
String fileName = path.substring(path.lastIndexOf("/") + 1);
|
String fileName = path.substring(path.lastIndexOf("/") + 1);
|
||||||
String suffix = fileName.substring(fileName.lastIndexOf("."));
|
String suffix = fileName.substring(fileName.lastIndexOf("."));
|
||||||
fileName = fileName.substring(0, fileName.lastIndexOf("."));
|
fileName = fileName.substring(0, fileName.lastIndexOf("."));
|
||||||
File image = new File(CachePath, fileName + ".avif");
|
String sourceUrl = "https://ehgt.org/" + path;
|
||||||
|
try {
|
||||||
if(image.isFile()){
|
Path image = ImageFileCache.get(Path.of(CachePath), fileName, () -> sourceUrl);
|
||||||
FileDownload.export(request, response, image.getAbsolutePath());
|
FileDownload.export(request, response, image.toString());
|
||||||
return;
|
} catch (InterruptedException e) {
|
||||||
}
|
Thread.currentThread().interrupt();
|
||||||
|
response.setStatus(503);
|
||||||
path = "https://ehgt.org/" + path;
|
} catch (Exception e) {
|
||||||
try(ServletOutputStream outputStream = response.getOutputStream()){
|
log.warn("获取缩略图失败: errorType={}", e.getClass().getSimpleName());
|
||||||
new URI(path).toURL().openConnection().getInputStream().transferTo(new FileOutputStream(CachePath + fileName + suffix));
|
if (!response.isCommitted()) response.setStatus(404);
|
||||||
GalleryUtil.convertImg(CachePath + fileName + suffix, suffix);
|
|
||||||
FileInputStream inputStream = new FileInputStream(image.getAbsoluteFile()); //如果放到括号里,会导致图片未创建时创建文件流失败报错
|
|
||||||
outputStream.write(inputStream.readAllBytes());
|
|
||||||
inputStream.close();
|
|
||||||
}catch (IOException | URISyntaxException e){
|
|
||||||
log.error("获取缩略图失败", e);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -44,9 +44,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
|||||||
@Slf4j
|
@Slf4j
|
||||||
public class RemoteService {
|
public class RemoteService {
|
||||||
|
|
||||||
ChannelFuture channelFuture;
|
volatile ChannelFuture channelFuture;
|
||||||
|
|
||||||
Channel channel;
|
volatile Channel channel;
|
||||||
|
|
||||||
@Value("${remote.ip:5.255.110.45}")
|
@Value("${remote.ip:5.255.110.45}")
|
||||||
String ip;
|
String ip;
|
||||||
@@ -62,7 +62,12 @@ public class RemoteService {
|
|||||||
ConcurrentHashMap<Integer, CopyOnWriteArrayList<CompletableFuture<String>>> retryStatusWaiters =
|
ConcurrentHashMap<Integer, CopyOnWriteArrayList<CompletableFuture<String>>> retryStatusWaiters =
|
||||||
new ConcurrentHashMap<>();
|
new ConcurrentHashMap<>();
|
||||||
|
|
||||||
EventLoop eventLoopGroup = new DefaultEventLoop();
|
final EventLoop eventLoopGroup = new DefaultEventLoop();
|
||||||
|
final EventLoopGroup networkGroup = new NioEventLoopGroup(2);
|
||||||
|
final AtomicBoolean connecting = new AtomicBoolean();
|
||||||
|
final AtomicBoolean monitoring = new AtomicBoolean();
|
||||||
|
volatile boolean stopping;
|
||||||
|
volatile ServerSocket monitorSocket;
|
||||||
|
|
||||||
ExecutorService downloadThread = Executors.newCachedThreadPool();
|
ExecutorService downloadThread = Executors.newCachedThreadPool();
|
||||||
|
|
||||||
@@ -89,20 +94,22 @@ public class RemoteService {
|
|||||||
|
|
||||||
@PostConstruct
|
@PostConstruct
|
||||||
void init() {
|
void init() {
|
||||||
if(!initChannel()){ //如果远程服务器连接失败,则开启本地监听
|
initChannel();
|
||||||
monitor = new Thread(this::monitorFunc);
|
|
||||||
monitor.start();
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean initChannel(){
|
public boolean initChannel(){
|
||||||
|
if (stopping || !connecting.compareAndSet(false, true))
|
||||||
|
return !isDead();
|
||||||
try {
|
try {
|
||||||
|
if (!isDead())
|
||||||
|
return true;
|
||||||
int i;
|
int i;
|
||||||
for(i=0; i<20; i++) {
|
for(i=0; i<20; i++) {
|
||||||
try {
|
try {
|
||||||
channelFuture = new Bootstrap()
|
channelFuture = new Bootstrap()
|
||||||
.channel(NioSocketChannel.class)
|
.channel(NioSocketChannel.class)
|
||||||
.group(new NioEventLoopGroup())
|
.group(networkGroup)
|
||||||
|
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3_000)
|
||||||
.handler(new ChannelInitializer<NioSocketChannel>() {
|
.handler(new ChannelInitializer<NioSocketChannel>() {
|
||||||
@Override
|
@Override
|
||||||
protected void initChannel(NioSocketChannel channel) {
|
protected void initChannel(NioSocketChannel channel) {
|
||||||
@@ -114,9 +121,14 @@ public class RemoteService {
|
|||||||
}
|
}
|
||||||
}).connect(new InetSocketAddress(ip, port + i)).sync();
|
}).connect(new InetSocketAddress(ip, port + i)).sync();
|
||||||
break;
|
break;
|
||||||
}catch (Exception e){
|
} catch (InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
|
return false;
|
||||||
|
} catch (Exception e) {
|
||||||
log.error("连接storageNode失败,端口偏移量(重试次数):{}", i);
|
log.error("连接storageNode失败,端口偏移量(重试次数):{}", i);
|
||||||
}
|
}
|
||||||
|
if (stopping)
|
||||||
|
return false;
|
||||||
}
|
}
|
||||||
|
|
||||||
//超过二十次连不上,主动抛出错误,由下方catch
|
//超过二十次连不上,主动抛出错误,由下方catch
|
||||||
@@ -124,11 +136,16 @@ public class RemoteService {
|
|||||||
throw new Exception();
|
throw new Exception();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (stopping) {
|
||||||
|
channelFuture.channel().close();
|
||||||
|
return false;
|
||||||
|
}
|
||||||
log.info("connect success");
|
log.info("connect success");
|
||||||
if(pushService != null)
|
if(pushService != null)
|
||||||
pushService.storageNodeOnline();
|
pushService.storageNodeOnline();
|
||||||
|
|
||||||
channel = channelFuture.channel();
|
channel = channelFuture.channel();
|
||||||
|
closeMonitorSocket();
|
||||||
channel.writeAndFlush(new IdentityMessage("lionwebsite"));
|
channel.writeAndFlush(new IdentityMessage("lionwebsite"));
|
||||||
|
|
||||||
//子节点上线时,发送未完成的任务
|
//子节点上线时,发送未完成的任务
|
||||||
@@ -138,6 +155,10 @@ public class RemoteService {
|
|||||||
}catch (Exception e){
|
}catch (Exception e){
|
||||||
log.error("connect node failed, wait for node back online", e);
|
log.error("connect node failed, wait for node back online", e);
|
||||||
return false;
|
return false;
|
||||||
|
} finally {
|
||||||
|
connecting.set(false);
|
||||||
|
if (isDead())
|
||||||
|
startMonitor();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -147,7 +168,7 @@ public class RemoteService {
|
|||||||
return -2;
|
return -2;
|
||||||
}
|
}
|
||||||
|
|
||||||
channelFuture.channel().close();
|
channelFuture.channel().close().awaitUninterruptibly();
|
||||||
|
|
||||||
if(initChannel()){
|
if(initChannel()){
|
||||||
return 0;
|
return 0;
|
||||||
@@ -157,34 +178,49 @@ public class RemoteService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public byte checkAvailability(){
|
public byte checkAvailability(){
|
||||||
AvailableCheckMessage acm = new AvailableCheckMessage();
|
return sendRequest(new AvailableCheckMessage(), 10, TimeUnit.SECONDS);
|
||||||
acm.setMessageId(atomicInteger.getAndIncrement());
|
}
|
||||||
|
|
||||||
channel.writeAndFlush(acm);
|
byte sendRequest(AbstractMessage message, long timeout, TimeUnit unit) {
|
||||||
|
Channel target = channel;
|
||||||
|
if (stopping || target == null || !target.isActive())
|
||||||
|
return -1;
|
||||||
|
message.setMessageId(atomicInteger.getAndIncrement());
|
||||||
DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
|
DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
|
||||||
promiseHashMap.put(acm.messageId, promise);
|
promiseHashMap.put(message.messageId, promise);
|
||||||
try {
|
try {
|
||||||
boolean result = promise.await(10, TimeUnit.SECONDS);
|
target.writeAndFlush(message).addListener(future -> {
|
||||||
if(result){
|
if (!future.isSuccess())
|
||||||
ResponseMessage rsm = (ResponseMessage)promise.getNow();
|
promise.tryFailure(future.cause() == null ? new IOException("节点发送失败") : future.cause());
|
||||||
return rsm.getResult();
|
});
|
||||||
}
|
if (promise.await(timeout, unit) && promise.isSuccess()
|
||||||
else return -1;
|
&& promise.getNow() instanceof ResponseMessage response)
|
||||||
}catch (InterruptedException e){
|
return response.getResult();
|
||||||
log.warn("checkAvailability interrupted", e);
|
return -1;
|
||||||
|
} catch (InterruptedException e) {
|
||||||
Thread.currentThread().interrupt();
|
Thread.currentThread().interrupt();
|
||||||
return -1;
|
return -1;
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.warn("节点请求失败: messageId={}", message.messageId, e);
|
||||||
|
return -1;
|
||||||
|
} finally {
|
||||||
|
promiseHashMap.remove(message.messageId, promise);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 请求将当前全部订阅状态异步同步到存储节点,短时间内的多次请求会合并。 */
|
/** 请求将当前全部订阅状态异步同步到存储节点,短时间内的多次请求会合并。 */
|
||||||
public void requestSubscriptionSync() {
|
public void requestSubscriptionSync() {
|
||||||
if (!subscriptionSyncEnabled)
|
if (stopping || !subscriptionSyncEnabled)
|
||||||
return;
|
return;
|
||||||
subscriptionSyncQueued.set(true);
|
subscriptionSyncQueued.set(true);
|
||||||
if (!subscriptionSyncRunning.compareAndSet(false, true))
|
if (!subscriptionSyncRunning.compareAndSet(false, true))
|
||||||
return;
|
return;
|
||||||
subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue);
|
try {
|
||||||
|
subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue);
|
||||||
|
} catch (java.util.concurrent.RejectedExecutionException e) {
|
||||||
|
subscriptionSyncRunning.set(false);
|
||||||
|
if (!stopping) throw e;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@Scheduled(fixedDelayString = "${subscription.standby.retry-interval-ms:60000}")
|
@Scheduled(fixedDelayString = "${subscription.standby.retry-interval-ms:60000}")
|
||||||
@@ -207,32 +243,15 @@ public class RemoteService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private void syncSubscriptionSnapshotOnce() {
|
private void syncSubscriptionSnapshotOnce() {
|
||||||
SubscriptionSnapshotMessage message = null;
|
|
||||||
DefaultPromise<AbstractMessage> promise = null;
|
|
||||||
try {
|
try {
|
||||||
message = subscriptionStandbySnapshotService.build();
|
SubscriptionSnapshotMessage message = subscriptionStandbySnapshotService.build();
|
||||||
message.setMessageId(atomicInteger.getAndIncrement());
|
byte result = sendRequest(message, 30, TimeUnit.SECONDS);
|
||||||
promise = new DefaultPromise<>(eventLoopGroup);
|
if (result == 0 || result == 3)
|
||||||
promiseHashMap.put(message.messageId, promise);
|
log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), result);
|
||||||
channel.writeAndFlush(message);
|
else
|
||||||
if (promise.await(30, TimeUnit.SECONDS)) {
|
log.warn("订阅快照同步失败或超时 revision={} result={}", shortRevision(message.getRevision()), result);
|
||||||
AbstractMessage reply = promise.getNow();
|
|
||||||
if (reply instanceof ResponseMessage response && (response.getResult() == 0 || response.getResult() == 3))
|
|
||||||
log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), response.getResult());
|
|
||||||
else
|
|
||||||
log.warn("订阅快照同步失败 revision={} result={}", shortRevision(message.getRevision()),
|
|
||||||
reply instanceof ResponseMessage response ? response.getResult() : "invalid-response");
|
|
||||||
} else {
|
|
||||||
log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision()));
|
|
||||||
}
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
log.warn("订阅快照同步线程被中断");
|
|
||||||
} catch (Exception e) {
|
} catch (Exception e) {
|
||||||
log.warn("生成或发送订阅快照失败: {}", e.getMessage());
|
log.warn("生成或发送订阅快照失败: {}", e.getMessage());
|
||||||
} finally {
|
|
||||||
if (message != null && promise != null)
|
|
||||||
promiseHashMap.remove(message.messageId, promise);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -241,8 +260,24 @@ public class RemoteService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@PreDestroy
|
@PreDestroy
|
||||||
void shutdownSubscriptionSync() {
|
void shutdownResources() {
|
||||||
|
stopping = true;
|
||||||
|
closeMonitorSocket();
|
||||||
|
if (monitor != null) monitor.interrupt();
|
||||||
|
failPendingRequests();
|
||||||
|
if (channel != null) channel.close();
|
||||||
subscriptionSyncExecutor.shutdownNow();
|
subscriptionSyncExecutor.shutdownNow();
|
||||||
|
downloadThread.shutdownNow();
|
||||||
|
networkGroup.shutdownGracefully();
|
||||||
|
eventLoopGroup.shutdownGracefully();
|
||||||
|
}
|
||||||
|
|
||||||
|
private void failPendingRequests() {
|
||||||
|
promiseHashMap.forEach((id, promise) -> promise.tryFailure(new IOException("节点连接已关闭")));
|
||||||
|
promiseHashMap.clear();
|
||||||
|
retryStatusWaiters.forEach((gid, waiters) ->
|
||||||
|
waiters.forEach(waiter -> waiter.completeExceptionally(new IOException("节点连接已关闭"))));
|
||||||
|
retryStatusWaiters.clear();
|
||||||
}
|
}
|
||||||
|
|
||||||
public boolean isDead(){
|
public boolean isDead(){
|
||||||
@@ -266,28 +301,9 @@ public class RemoteService {
|
|||||||
GalleryTask galleryTask = new GalleryTask();
|
GalleryTask galleryTask = new GalleryTask();
|
||||||
galleryTask.setGid(gallery.getGid());
|
galleryTask.setGid(gallery.getGid());
|
||||||
galleryTask.setName(gallery.getName());
|
galleryTask.setName(gallery.getName());
|
||||||
|
DownloadPostMessage message = new DownloadPostMessage();
|
||||||
DownloadPostMessage dpm = new DownloadPostMessage();
|
message.setGalleryTask(galleryTask);
|
||||||
dpm.messageId = atomicInteger.getAndIncrement();
|
return sendRequest(message, 10, TimeUnit.SECONDS);
|
||||||
dpm.setGalleryTask(galleryTask);
|
|
||||||
|
|
||||||
DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
|
|
||||||
promiseHashMap.put(dpm.messageId, promise);
|
|
||||||
channel.writeAndFlush(dpm);
|
|
||||||
try {
|
|
||||||
boolean result = promise.await(10, TimeUnit.SECONDS);
|
|
||||||
if(result){
|
|
||||||
ResponseMessage rsm = (ResponseMessage)promise.getNow();
|
|
||||||
return rsm.getResult();
|
|
||||||
}
|
|
||||||
else return -1;
|
|
||||||
}catch (InterruptedException e){
|
|
||||||
log.warn("addGalleryToQueue interrupted", e);
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
return -1;
|
|
||||||
}finally {
|
|
||||||
promiseHashMap.remove(dpm.messageId, promise);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public RetryResult retryGallery(Gallery gallery){
|
public RetryResult retryGallery(Gallery gallery){
|
||||||
@@ -325,49 +341,58 @@ public class RemoteService {
|
|||||||
public record RetryResult(boolean success, String message) {}
|
public record RetryResult(boolean success, String message) {}
|
||||||
|
|
||||||
public byte deleteGallery(Gallery gallery){
|
public byte deleteGallery(Gallery gallery){
|
||||||
DeleteGalleryMessage dgm = new DeleteGalleryMessage();
|
DeleteGalleryMessage message = new DeleteGalleryMessage();
|
||||||
dgm.setGalleryName(gallery.getName());
|
message.setGalleryName(gallery.getName());
|
||||||
dgm.messageId = atomicInteger.getAndIncrement();
|
return sendRequest(message, 10, TimeUnit.SECONDS);
|
||||||
|
}
|
||||||
|
|
||||||
channel.writeAndFlush(dgm);
|
private void startMonitor() {
|
||||||
DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
|
if (stopping || !monitoring.compareAndSet(false, true))
|
||||||
promiseHashMap.put(dgm.messageId, promise);
|
return;
|
||||||
try{
|
monitor = new Thread(this::monitorFunc, "storage-node-monitor");
|
||||||
boolean result = promise.await(10, TimeUnit.SECONDS);
|
monitor.setDaemon(true);
|
||||||
if(result){
|
monitor.start();
|
||||||
ResponseMessage rsm = (ResponseMessage) promise.getNow();
|
}
|
||||||
return rsm.getResult();
|
|
||||||
}else return -1;
|
private void closeMonitorSocket() {
|
||||||
}catch (InterruptedException e){
|
ServerSocket socket = monitorSocket;
|
||||||
log.warn("deleteGallery interrupted", e);
|
if (socket != null) {
|
||||||
Thread.currentThread().interrupt();
|
try { socket.close(); }
|
||||||
return -1;
|
catch (IOException e) { log.debug("关闭节点监听失败", e); }
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
public void monitorFunc(){
|
public void monitorFunc(){
|
||||||
int real_port = CustomUtil._findIdlePort(port + 1);
|
try (ServerSocket socket = new ServerSocket(CustomUtil._findIdlePort(port + 1))) {
|
||||||
log.info("监听端口: {}等待节点上线", real_port);
|
monitorSocket = socket;
|
||||||
try(ServerSocket socket = new ServerSocket(real_port)) {
|
if (stopping || !isDead())
|
||||||
Socket client;
|
return;
|
||||||
while(true){
|
log.info("监听端口: {}等待节点上线", socket.getLocalPort());
|
||||||
client = socket.accept();
|
while (!stopping) {
|
||||||
|
try (Socket client = socket.accept()) {
|
||||||
if(client.getInetAddress().getHostAddress().equals(ip)){
|
if (!client.getInetAddress().getHostAddress().equals(ip))
|
||||||
//连接之后发送lionwebsite,否则存储节点不能确认这个端口是否有效
|
continue;
|
||||||
OutputStream outputStream = client.getOutputStream();
|
OutputStream output = client.getOutputStream();
|
||||||
outputStream.write("lionwebsite".getBytes());
|
output.write("lionwebsite".getBytes(java.nio.charset.StandardCharsets.UTF_8));
|
||||||
outputStream.flush();
|
output.flush();
|
||||||
outputStream.close();
|
client.shutdownOutput();
|
||||||
log.info("尝试连接");
|
if (initChannel())
|
||||||
initChannel();
|
break;
|
||||||
client.close();
|
|
||||||
socket.close();
|
|
||||||
break;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
throw new RuntimeException(e);
|
if (!stopping && isDead())
|
||||||
|
log.warn("等待节点上线失败", e);
|
||||||
|
} finally {
|
||||||
|
monitorSocket = null;
|
||||||
|
monitoring.set(false);
|
||||||
|
// A connection may flap while the previous monitor is still exiting.
|
||||||
|
if (!stopping && isDead()) {
|
||||||
|
try { networkGroup.next().schedule(this::startMonitor, 5, TimeUnit.SECONDS); }
|
||||||
|
catch (java.util.concurrent.RejectedExecutionException e) {
|
||||||
|
if (!stopping) log.warn("安排节点监听重试失败", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -416,7 +441,7 @@ public class RemoteService {
|
|||||||
else if(msg instanceof ResponseMessage rsm) {
|
else if(msg instanceof ResponseMessage rsm) {
|
||||||
Promise<AbstractMessage> promise = promiseHashMap.remove(rsm.messageId);
|
Promise<AbstractMessage> promise = promiseHashMap.remove(rsm.messageId);
|
||||||
if(promise != null)
|
if(promise != null)
|
||||||
promise.setSuccess(rsm);
|
promise.trySuccess(rsm);
|
||||||
else
|
else
|
||||||
log.warn("收到无等待者的响应消息: messageId={}", rsm.messageId);
|
log.warn("收到无等待者的响应消息: messageId={}", rsm.messageId);
|
||||||
}
|
}
|
||||||
@@ -424,13 +449,12 @@ public class RemoteService {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void channelUnregistered(ChannelHandlerContext ctx) {
|
public void channelUnregistered(ChannelHandlerContext ctx) {
|
||||||
log.info("{}", ctx.channel());
|
if (ctx.channel() == channel) {
|
||||||
log.info("{}", channel);
|
failPendingRequests();
|
||||||
if(ctx.channel() != null && ctx.channel().remoteAddress().toString().equals(channel.remoteAddress().toString())){
|
if (!stopping) {
|
||||||
log.info("activate monitor thread, waiting for node back online");
|
pushService.storageNodeOffline();
|
||||||
pushService.storageNodeOffline();
|
startMonitor();
|
||||||
monitor = new Thread(RemoteService.this::monitorFunc);
|
}
|
||||||
monitor.start();
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -37,17 +37,20 @@ public class SubService {
|
|||||||
final SubscriptionStateCoordinator stateCoordinator;
|
final SubscriptionStateCoordinator stateCoordinator;
|
||||||
|
|
||||||
public String insertSubscriptionAccount(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
public String insertSubscriptionAccount(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
||||||
return withWriteLock(() -> insertSubscriptionAccountUnlocked(name, upstreamKey, filterHighMultiplier, enabled));
|
|
||||||
}
|
|
||||||
|
|
||||||
private String insertSubscriptionAccountUnlocked(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
|
||||||
Response response = Response.generateResponse();
|
Response response = Response.generateResponse();
|
||||||
if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
|
SubscriptionAccount account;
|
||||||
return response.failure("名称和上游 key 不能为空").toJSONString();
|
Lock lock = stateCoordinator.writeLock();
|
||||||
if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0)
|
lock.lock();
|
||||||
return response.failure("名称或上游 key 已存在").toJSONString();
|
try {
|
||||||
SubscriptionAccount account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null);
|
if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
|
||||||
subMapper.insertSubscriptionAccount(account);
|
return response.failure("名称和上游 key 不能为空").toJSONString();
|
||||||
|
if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0)
|
||||||
|
return response.failure("名称或上游 key 已存在").toJSONString();
|
||||||
|
account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null);
|
||||||
|
subMapper.insertSubscriptionAccount(account);
|
||||||
|
} finally {
|
||||||
|
lock.unlock();
|
||||||
|
}
|
||||||
if (enabled)
|
if (enabled)
|
||||||
refreshService.refresh(account.getId());
|
refreshService.refresh(account.getId());
|
||||||
remoteService.requestSubscriptionSync();
|
remoteService.requestSubscriptionSync();
|
||||||
@@ -64,27 +67,34 @@ public class SubService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public String updateSubscriptionAccount(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
public String updateSubscriptionAccount(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
||||||
return withWriteLock(() -> updateSubscriptionAccountUnlocked(id, name, upstreamKey, filterHighMultiplier, enabled));
|
|
||||||
}
|
|
||||||
|
|
||||||
private String updateSubscriptionAccountUnlocked(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
|
|
||||||
Response response = Response.generateResponse();
|
Response response = Response.generateResponse();
|
||||||
SubscriptionAccount account = subMapper.selectSubscriptionAccount(id);
|
SubscriptionAccount account;
|
||||||
if (account == null)
|
Lock lock = stateCoordinator.writeLock();
|
||||||
return response.failure("子账号不存在").toJSONString();
|
lock.lock();
|
||||||
if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
|
try {
|
||||||
return response.failure("名称和上游 key 不能为空").toJSONString();
|
account = subMapper.selectSubscriptionAccount(id);
|
||||||
for (SubscriptionAccount existing : subMapper.selectAllSubscriptionAccounts()) {
|
if (account == null)
|
||||||
if (!existing.getId().equals(id) && (existing.getName().equals(name.trim()) || existing.getUpstreamKey().equals(upstreamKey.trim())))
|
return response.failure("子账号不存在").toJSONString();
|
||||||
return response.failure("名称或上游 key 已存在").toJSONString();
|
if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
|
||||||
|
return response.failure("名称和上游 key 不能为空").toJSONString();
|
||||||
|
for (SubscriptionAccount existing : subMapper.selectAllSubscriptionAccounts()) {
|
||||||
|
if (!existing.getId().equals(id) && (existing.getName().equals(name.trim()) || existing.getUpstreamKey().equals(upstreamKey.trim())))
|
||||||
|
return response.failure("名称或上游 key 已存在").toJSONString();
|
||||||
|
}
|
||||||
|
boolean changed = !account.getUpstreamKey().equals(upstreamKey.trim())
|
||||||
|
|| account.isFilterHighMultiplier() != filterHighMultiplier || account.isEnabled() != enabled;
|
||||||
|
account.setName(name.trim());
|
||||||
|
account.setUpstreamKey(upstreamKey.trim());
|
||||||
|
account.setFilterHighMultiplier(filterHighMultiplier);
|
||||||
|
account.setEnabled(enabled);
|
||||||
|
subMapper.updateSubscriptionAccount(account);
|
||||||
|
if (changed)
|
||||||
|
refreshService.invalidateCache(id);
|
||||||
|
} finally {
|
||||||
|
lock.unlock();
|
||||||
}
|
}
|
||||||
account.setName(name.trim());
|
if (enabled)
|
||||||
account.setUpstreamKey(upstreamKey.trim());
|
refreshService.refresh(id);
|
||||||
account.setFilterHighMultiplier(filterHighMultiplier);
|
|
||||||
account.setEnabled(enabled);
|
|
||||||
subMapper.updateSubscriptionAccount(account);
|
|
||||||
if (enabled && !refreshService.refresh(id))
|
|
||||||
refreshService.invalidateCache(id);
|
|
||||||
remoteService.requestSubscriptionSync();
|
remoteService.requestSubscriptionSync();
|
||||||
return response.success(accountJson(account)).toJSONString();
|
return response.success(accountJson(account)).toJSONString();
|
||||||
}
|
}
|
||||||
@@ -101,15 +111,12 @@ public class SubService {
|
|||||||
if (account.getBoundUserCount() != null && account.getBoundUserCount() > 0)
|
if (account.getBoundUserCount() != null && account.getBoundUserCount() > 0)
|
||||||
return response.failure("子账号仍绑定用户,请先改绑").toJSONString();
|
return response.failure("子账号仍绑定用户,请先改绑").toJSONString();
|
||||||
subMapper.deleteSubscriptionAccount(id);
|
subMapper.deleteSubscriptionAccount(id);
|
||||||
|
refreshService.invalidateCache(id);
|
||||||
remoteService.requestSubscriptionSync();
|
remoteService.requestSubscriptionSync();
|
||||||
return response.success("删除成功").toJSONString();
|
return response.success("删除成功").toJSONString();
|
||||||
}
|
}
|
||||||
|
|
||||||
public String refreshSubscriptionAccount(Integer id) {
|
public String refreshSubscriptionAccount(Integer id) {
|
||||||
return withWriteLock(() -> refreshSubscriptionAccountUnlocked(id));
|
|
||||||
}
|
|
||||||
|
|
||||||
private String refreshSubscriptionAccountUnlocked(Integer id) {
|
|
||||||
boolean success = refreshService.refresh(id);
|
boolean success = refreshService.refresh(id);
|
||||||
remoteService.requestSubscriptionSync();
|
remoteService.requestSubscriptionSync();
|
||||||
return success ? Response._success("刷新成功") : Response._failure("刷新失败,请查看子账号错误状态");
|
return success ? Response._success("刷新成功") : Response._failure("刷新失败,请查看子账号错误状态");
|
||||||
@@ -188,12 +195,7 @@ public class SubService {
|
|||||||
sendStatus(response, HttpServletResponse.SC_NOT_FOUND, "subscription not found");
|
sendStatus(response, HttpServletResponse.SC_NOT_FOUND, "subscription not found");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
String ip = request.getRemoteAddr();
|
String ip = resolveClientIp(request);
|
||||||
if ("127.0.0.1".equals(ip)) {
|
|
||||||
ip = request.getHeader("X-Forwarded-For");
|
|
||||||
if (ip != null && ip.contains(",")) ip = ip.split(",")[0].trim();
|
|
||||||
if (ip != null && ip.contains(":")) ip = ip.split(":")[0].trim();
|
|
||||||
}
|
|
||||||
String ua = request.getHeader("User-Agent");
|
String ua = request.getHeader("User-Agent");
|
||||||
if (ua == null) return;
|
if (ua == null) return;
|
||||||
recordUpdate(subBind.getUser(), ip, ua);
|
recordUpdate(subBind.getUser(), ip, ua);
|
||||||
@@ -209,6 +211,31 @@ public class SubService {
|
|||||||
FileDownload.export(request, response, path.toString());
|
FileDownload.export(request, response, path.toString());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static String resolveClientIp(HttpServletRequest request) {
|
||||||
|
String remoteAddr = headerAddress(request.getRemoteAddr());
|
||||||
|
if (isLoopback(remoteAddr)) {
|
||||||
|
String forwardedFor = headerAddress(request.getHeader("X-Forwarded-For"));
|
||||||
|
if (forwardedFor != null)
|
||||||
|
return forwardedFor;
|
||||||
|
String realIp = headerAddress(request.getHeader("X-Real-IP"));
|
||||||
|
if (realIp != null)
|
||||||
|
return realIp;
|
||||||
|
}
|
||||||
|
return remoteAddr == null ? "unknown" : remoteAddr;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static String headerAddress(String value) {
|
||||||
|
if (value == null || value.isBlank())
|
||||||
|
return null;
|
||||||
|
String address = value.split(",", 2)[0].trim();
|
||||||
|
return address.isBlank() || "unknown".equalsIgnoreCase(address) ? null : address;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static boolean isLoopback(String address) {
|
||||||
|
return "127.0.0.1".equals(address) || "::1".equals(address)
|
||||||
|
|| "0:0:0:0:0:0:0:1".equals(address);
|
||||||
|
}
|
||||||
|
|
||||||
private void recordUpdate(String user, String ip, String ua) {
|
private void recordUpdate(String user, String ip, String ua) {
|
||||||
String location;
|
String location;
|
||||||
try {
|
try {
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import com.lion.lionwebsite.Domain.SubscriptionAccount;
|
|||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.apache.http.HttpEntity;
|
import org.apache.http.HttpEntity;
|
||||||
|
import org.apache.http.client.config.RequestConfig;
|
||||||
import org.apache.http.client.methods.CloseableHttpResponse;
|
import org.apache.http.client.methods.CloseableHttpResponse;
|
||||||
import org.apache.http.client.methods.HttpGet;
|
import org.apache.http.client.methods.HttpGet;
|
||||||
import org.apache.http.impl.client.CloseableHttpClient;
|
import org.apache.http.impl.client.CloseableHttpClient;
|
||||||
@@ -25,7 +26,10 @@ import java.util.concurrent.locks.Lock;
|
|||||||
@Slf4j
|
@Slf4j
|
||||||
@RequiredArgsConstructor
|
@RequiredArgsConstructor
|
||||||
public class SubscriptionRefreshService {
|
public class SubscriptionRefreshService {
|
||||||
private static final CloseableHttpClient HTTP_CLIENT = HttpClients.createDefault();
|
private static final CloseableHttpClient HTTP_CLIENT = HttpClients.custom()
|
||||||
|
.setDefaultRequestConfig(RequestConfig.custom().setConnectTimeout(5_000)
|
||||||
|
.setConnectionRequestTimeout(5_000).setSocketTimeout(15_000).build())
|
||||||
|
.build();
|
||||||
private static final Pattern MULTIPLIER = Pattern.compile("(\\d+(?:\\.\\d+)?)x\\s*$", Pattern.CASE_INSENSITIVE);
|
private static final Pattern MULTIPLIER = Pattern.compile("(\\d+(?:\\.\\d+)?)x\\s*$", Pattern.CASE_INSENSITIVE);
|
||||||
|
|
||||||
final SubMapper subMapper;
|
final SubMapper subMapper;
|
||||||
@@ -53,31 +57,63 @@ public class SubscriptionRefreshService {
|
|||||||
return success;
|
return success;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private long refreshSequence;
|
||||||
|
// Accessed only while holding the coordinator write lock.
|
||||||
|
private final Map<Integer, Long> latestRefresh = new HashMap<>();
|
||||||
|
|
||||||
public boolean refresh(Integer accountId) {
|
public boolean refresh(Integer accountId) {
|
||||||
Lock stateLock = stateCoordinator.writeLock();
|
Lock stateLock = stateCoordinator.writeLock();
|
||||||
|
SubscriptionAccount account;
|
||||||
|
long version;
|
||||||
stateLock.lock();
|
stateLock.lock();
|
||||||
try {
|
try {
|
||||||
SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId);
|
account = subMapper.selectSubscriptionAccount(accountId);
|
||||||
if (account == null || !account.isEnabled())
|
if (account == null || !account.isEnabled())
|
||||||
return false;
|
return false;
|
||||||
try {
|
version = ++refreshSequence;
|
||||||
String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold);
|
latestRefresh.put(accountId, version);
|
||||||
List<String> clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold);
|
|
||||||
Path dir = Paths.get(cacheRoot, String.valueOf(accountId));
|
|
||||||
Files.createDirectories(dir);
|
|
||||||
atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8));
|
|
||||||
atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8));
|
|
||||||
subMapper.markSubscriptionRefreshSuccess(accountId);
|
|
||||||
return true;
|
|
||||||
} catch (Exception e) {
|
|
||||||
String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
|
|
||||||
subMapper.markSubscriptionRefreshFailure(accountId, message.length() > 500 ? message.substring(0, 500) : message);
|
|
||||||
log.error("刷新子账号订阅失败 accountId={}: {}", accountId, message);
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
} finally {
|
} finally {
|
||||||
stateLock.unlock();
|
stateLock.unlock();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Network access and parsing never hold the shared subscription lock.
|
||||||
|
try {
|
||||||
|
String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold);
|
||||||
|
List<String> clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold);
|
||||||
|
stateLock.lock();
|
||||||
|
try {
|
||||||
|
if (!isCurrent(account, version))
|
||||||
|
return false;
|
||||||
|
Path dir = Paths.get(cacheRoot, String.valueOf(accountId));
|
||||||
|
Files.createDirectories(dir);
|
||||||
|
atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8));
|
||||||
|
atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8));
|
||||||
|
subMapper.markSubscriptionRefreshSuccess(accountId);
|
||||||
|
return true;
|
||||||
|
} finally {
|
||||||
|
stateLock.unlock();
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
stateLock.lock();
|
||||||
|
try {
|
||||||
|
if (isCurrent(account, version)) {
|
||||||
|
String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
|
||||||
|
subMapper.markSubscriptionRefreshFailure(accountId, message.length() > 500 ? message.substring(0, 500) : message);
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
stateLock.unlock();
|
||||||
|
}
|
||||||
|
log.warn("刷新子账号订阅失败 accountId={} errorType={}", accountId, e.getClass().getSimpleName());
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private boolean isCurrent(SubscriptionAccount expected, long version) {
|
||||||
|
SubscriptionAccount current = subMapper.selectSubscriptionAccount(expected.getId());
|
||||||
|
return Objects.equals(latestRefresh.get(expected.getId()), version)
|
||||||
|
&& current != null && current.isEnabled()
|
||||||
|
&& Objects.equals(current.getUpstreamKey(), expected.getUpstreamKey())
|
||||||
|
&& current.isFilterHighMultiplier() == expected.isFilterHighMultiplier();
|
||||||
}
|
}
|
||||||
|
|
||||||
public String v2Url(SubscriptionAccount account) {
|
public String v2Url(SubscriptionAccount account) {
|
||||||
@@ -179,7 +215,7 @@ public class SubscriptionRefreshService {
|
|||||||
return matcher.find() && Double.parseDouble(matcher.group(1)) > threshold;
|
return matcher.find() && Double.parseDouble(matcher.group(1)) > threshold;
|
||||||
}
|
}
|
||||||
|
|
||||||
private static List<String> download(String url) throws IOException {
|
List<String> download(String url) throws IOException {
|
||||||
HttpGet get = new HttpGet(url);
|
HttpGet get = new HttpGet(url);
|
||||||
try (CloseableHttpResponse response = HTTP_CLIENT.execute(get)) {
|
try (CloseableHttpResponse response = HTTP_CLIENT.execute(get)) {
|
||||||
if (response.getStatusLine().getStatusCode() != 200)
|
if (response.getStatusLine().getStatusCode() != 200)
|
||||||
@@ -219,6 +255,7 @@ public class SubscriptionRefreshService {
|
|||||||
Lock stateLock = stateCoordinator.writeLock();
|
Lock stateLock = stateCoordinator.writeLock();
|
||||||
stateLock.lock();
|
stateLock.lock();
|
||||||
try {
|
try {
|
||||||
|
latestRefresh.remove(accountId); // In-flight responses must not restore invalidated content.
|
||||||
Files.deleteIfExists(cachedPath(accountId, "v2"));
|
Files.deleteIfExists(cachedPath(accountId, "v2"));
|
||||||
Files.deleteIfExists(cachedPath(accountId, "cat"));
|
Files.deleteIfExists(cachedPath(accountId, "cat"));
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
|
|||||||
@@ -1,119 +1,83 @@
|
|||||||
package com.lion.lionwebsite.Util;
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
import cn.hutool.core.util.StrUtil;
|
|
||||||
import cn.hutool.core.util.URLUtil;
|
|
||||||
import jakarta.servlet.http.HttpServletRequest;
|
import jakarta.servlet.http.HttpServletRequest;
|
||||||
import jakarta.servlet.http.HttpServletResponse;
|
import jakarta.servlet.http.HttpServletResponse;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.apache.catalina.connector.ClientAbortException;
|
import org.apache.catalina.connector.ClientAbortException;
|
||||||
|
import org.springframework.http.ContentDisposition;
|
||||||
import org.springframework.http.HttpHeaders;
|
import org.springframework.http.HttpHeaders;
|
||||||
|
import org.springframework.http.HttpRange;
|
||||||
|
|
||||||
import java.io.BufferedOutputStream;
|
import java.io.*;
|
||||||
import java.io.File;
|
import java.nio.charset.StandardCharsets;
|
||||||
import java.io.IOException;
|
import java.util.List;
|
||||||
import java.io.RandomAccessFile;
|
|
||||||
|
|
||||||
|
|
||||||
@Slf4j
|
@Slf4j
|
||||||
public class FileDownload {
|
public class FileDownload {
|
||||||
public static void export(HttpServletRequest request, HttpServletResponse response, String path) {
|
public static void export(HttpServletRequest request, HttpServletResponse response, String path) {
|
||||||
File file = new File(path);
|
File file = new File(path);
|
||||||
|
if (!file.isFile()) {
|
||||||
String fileName = file.getName();
|
response.setStatus(HttpServletResponse.SC_NOT_FOUND);
|
||||||
|
return;
|
||||||
String range = request.getHeader(HttpHeaders.RANGE);
|
|
||||||
|
|
||||||
String rangeSeparator = "-";
|
|
||||||
// 开始下载位置
|
|
||||||
long startByte = 0;
|
|
||||||
// 结束下载位置
|
|
||||||
long endByte = file.length() - 1;
|
|
||||||
|
|
||||||
// 如果是断点续传
|
|
||||||
if (range != null && range.contains("bytes=") && range.contains(rangeSeparator)) {
|
|
||||||
// 设置响应状态码为 206
|
|
||||||
response.setStatus(HttpServletResponse.SC_PARTIAL_CONTENT);
|
|
||||||
|
|
||||||
range = range.substring(range.lastIndexOf("=") + 1).trim();
|
|
||||||
String[] ranges = range.split(rangeSeparator);
|
|
||||||
try {
|
|
||||||
// 判断 range 的类型
|
|
||||||
if (ranges.length == 1) {
|
|
||||||
// 类型一:bytes=-2343
|
|
||||||
if (range.startsWith(rangeSeparator)) {
|
|
||||||
endByte = Long.parseLong(ranges[0]);
|
|
||||||
}
|
|
||||||
// 类型二:bytes=2343-
|
|
||||||
else if (range.endsWith(rangeSeparator)) {
|
|
||||||
startByte = Long.parseLong(ranges[0]);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
// 类型三:bytes=22-2343
|
|
||||||
else if (ranges.length == 2) {
|
|
||||||
startByte = Long.parseLong(ranges[0]);
|
|
||||||
endByte = Long.parseLong(ranges[1]);
|
|
||||||
}
|
|
||||||
} catch (NumberFormatException e) {
|
|
||||||
// 传参不规范,则直接返回所有内容
|
|
||||||
startByte = 0;
|
|
||||||
endByte = file.length() - 1;
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// 没有 ranges 即全部一次性传输,需要用 200 状态码,这一行应该可以省掉,因为默认返回是 200 状态码
|
|
||||||
response.setStatus(HttpServletResponse.SC_OK);
|
|
||||||
}
|
}
|
||||||
|
// Size and content refer to the same opened file, even if a cache is replaced.
|
||||||
//要下载的长度(endByte 为总长度 -1,这时候要加回去)
|
try (RandomAccessFile input = new RandomAccessFile(file, "r")) {
|
||||||
long contentLength = endByte - startByte + 1;
|
long size = input.length();
|
||||||
//文件类型
|
long start = 0;
|
||||||
String contentType = request.getServletContext().getMimeType(fileName);
|
long end = size - 1;
|
||||||
|
boolean partial = false;
|
||||||
if (StrUtil.isEmpty(contentType)) {
|
String range = request.getHeader(HttpHeaders.RANGE);
|
||||||
contentType = "attachment";
|
if (range != null && range.startsWith("bytes=")) {
|
||||||
}
|
try {
|
||||||
|
List<HttpRange> ranges = HttpRange.parseRanges(range);
|
||||||
response.setHeader(HttpHeaders.ACCEPT_RANGES, "bytes");
|
// Multiple ranges are intentionally ignored; send the full representation.
|
||||||
response.setHeader(HttpHeaders.CONTENT_TYPE, contentType);
|
if (ranges.size() == 1) {
|
||||||
// 这里文件名换你想要的,inline 表示浏览器可以直接使用
|
if (size == 0) throw new IllegalArgumentException("empty file");
|
||||||
// 参考资料:https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Headers/Content-Disposition
|
start = ranges.getFirst().getRangeStart(size);
|
||||||
response.setHeader(HttpHeaders.CONTENT_DISPOSITION, contentType + ";filename=\"" + URLUtil.encode(fileName) + "\"");
|
end = ranges.getFirst().getRangeEnd(size);
|
||||||
response.setHeader(HttpHeaders.CONTENT_LENGTH, String.valueOf(contentLength));
|
if (start < 0 || start >= size || end < start)
|
||||||
// [要下载的开始位置]-[结束位置]/[文件总大小]
|
throw new IllegalArgumentException("unsatisfiable range");
|
||||||
response.setHeader(HttpHeaders.CONTENT_RANGE, "bytes " + startByte + rangeSeparator + endByte + "/" + file.length());
|
partial = true;
|
||||||
|
}
|
||||||
BufferedOutputStream outputStream;
|
} catch (IllegalArgumentException e) {
|
||||||
//已传送数据大小
|
response.setStatus(HttpServletResponse.SC_REQUESTED_RANGE_NOT_SATISFIABLE);
|
||||||
long transmitted = 0;
|
response.setHeader(HttpHeaders.CONTENT_RANGE, "bytes */" + size);
|
||||||
try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
|
response.setContentLengthLong(0);
|
||||||
try {
|
return;
|
||||||
outputStream = new BufferedOutputStream(response.getOutputStream());
|
|
||||||
byte[] buff = new byte[4096];
|
|
||||||
int len = 0;
|
|
||||||
randomAccessFile.seek(startByte);
|
|
||||||
while ((transmitted + len) <= contentLength && (len = randomAccessFile.read(buff)) != -1) {
|
|
||||||
outputStream.write(buff, 0, len);
|
|
||||||
transmitted += len;
|
|
||||||
// 本地测试, 防止下载速度过快
|
|
||||||
// Thread.sleep(1);
|
|
||||||
}
|
}
|
||||||
// 处理不足 buff.length 部分
|
|
||||||
if (transmitted < contentLength) {
|
|
||||||
len = randomAccessFile.read(buff, 0, (int) (contentLength - transmitted));
|
|
||||||
outputStream.write(buff, 0, len);
|
|
||||||
}
|
|
||||||
|
|
||||||
outputStream.flush();
|
|
||||||
response.flushBuffer();
|
|
||||||
randomAccessFile.close();
|
|
||||||
// log.trace("下载完毕: {}-{}, 已传输 {}", startByte, endByte, transmitted);
|
|
||||||
} catch (ClientAbortException e) {
|
|
||||||
// ignore 用户停止下载
|
|
||||||
// log.trace("用户停止下载: {}-{}, 已传输 {}", startByte, endByte, transmitted);
|
|
||||||
} catch (IOException e) {
|
|
||||||
log.error("文件下载IO错误: {}", path, e);
|
|
||||||
}
|
}
|
||||||
|
long remaining = end - start + 1;
|
||||||
|
response.setStatus(partial ? HttpServletResponse.SC_PARTIAL_CONTENT : HttpServletResponse.SC_OK);
|
||||||
|
response.setHeader(HttpHeaders.ACCEPT_RANGES, "bytes");
|
||||||
|
String mime = request.getServletContext().getMimeType(file.getName());
|
||||||
|
response.setContentType(mime == null ? "application/octet-stream" : mime);
|
||||||
|
response.setHeader(HttpHeaders.CONTENT_DISPOSITION,
|
||||||
|
ContentDisposition.inline().filename(file.getName(), StandardCharsets.UTF_8).build().toString());
|
||||||
|
response.setContentLengthLong(remaining);
|
||||||
|
if (partial)
|
||||||
|
response.setHeader(HttpHeaders.CONTENT_RANGE, "bytes " + start + "-" + end + "/" + size);
|
||||||
|
if ("HEAD".equalsIgnoreCase(request.getMethod()))
|
||||||
|
return;
|
||||||
|
input.seek(start);
|
||||||
|
BufferedOutputStream output = new BufferedOutputStream(response.getOutputStream());
|
||||||
|
byte[] buffer = new byte[8192];
|
||||||
|
while (remaining > 0) {
|
||||||
|
int count = input.read(buffer, 0, (int) Math.min(buffer.length, remaining));
|
||||||
|
if (count == -1)
|
||||||
|
throw new EOFException("File changed during download");
|
||||||
|
output.write(buffer, 0, count);
|
||||||
|
remaining -= count;
|
||||||
|
}
|
||||||
|
output.flush();
|
||||||
|
response.flushBuffer();
|
||||||
|
} catch (ClientAbortException e) {
|
||||||
|
// The client cancelled its download.
|
||||||
} catch (IOException e) {
|
} catch (IOException e) {
|
||||||
log.warn("关闭RandomAccessFile失败: {}", path, e);
|
log.warn("文件下载失败: {}", path, e);
|
||||||
|
if (!response.isCommitted()) {
|
||||||
|
response.reset();
|
||||||
|
response.setStatus(HttpServletResponse.SC_INTERNAL_SERVER_ERROR);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5,6 +5,9 @@ import com.lion.lionwebsite.Domain.Gallery;
|
|||||||
import com.lion.lionwebsite.Domain.ImageKeyCache;
|
import com.lion.lionwebsite.Domain.ImageKeyCache;
|
||||||
import com.lion.lionwebsite.Exception.ResolutionNotMatchException;
|
import com.lion.lionwebsite.Exception.ResolutionNotMatchException;
|
||||||
import org.apache.http.HttpEntity;
|
import org.apache.http.HttpEntity;
|
||||||
|
import org.apache.http.client.config.RequestConfig;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
import org.apache.http.client.entity.EntityBuilder;
|
import org.apache.http.client.entity.EntityBuilder;
|
||||||
import org.apache.http.client.methods.CloseableHttpResponse;
|
import org.apache.http.client.methods.CloseableHttpResponse;
|
||||||
import org.apache.http.client.methods.HttpGet;
|
import org.apache.http.client.methods.HttpGet;
|
||||||
@@ -44,7 +47,9 @@ public class GalleryUtil {
|
|||||||
static ConcurrentHashMap<String, String> gid2MpvKey = new ConcurrentHashMap<>();
|
static ConcurrentHashMap<String, String> gid2MpvKey = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
/** Reusable HTTP client —不要每次请求新建 */
|
/** Reusable HTTP client —不要每次请求新建 */
|
||||||
private static final CloseableHttpClient httpClient = HttpClients.createDefault();
|
private static final CloseableHttpClient httpClient = HttpClients.custom()
|
||||||
|
.setDefaultRequestConfig(RequestConfig.custom().setConnectTimeout(5_000)
|
||||||
|
.setConnectionRequestTimeout(5_000).setSocketTimeout(15_000).build()).build();
|
||||||
|
|
||||||
/** E-Hentai Cookie, injected from application.yaml via CustomBean */
|
/** E-Hentai Cookie, injected from application.yaml via CustomBean */
|
||||||
private static String ehentaiCookie = "";
|
private static String ehentaiCookie = "";
|
||||||
@@ -189,10 +194,13 @@ public class GalleryUtil {
|
|||||||
|
|
||||||
public static String getMpvKey(String url){
|
public static String getMpvKey(String url){
|
||||||
String gid = String.valueOf(parseGid(url));
|
String gid = String.valueOf(parseGid(url));
|
||||||
return gid2MpvKey.computeIfAbsent(gid, k -> {
|
String key = gid2MpvKey.get(gid);
|
||||||
|
if (key == null) {
|
||||||
|
// refreshMpvKey writes the cache itself; never call it inside computeIfAbsent.
|
||||||
refreshMpvKey(url);
|
refreshMpvKey(url);
|
||||||
return gid2MpvKey.get(k);
|
key = gid2MpvKey.get(gid);
|
||||||
});
|
}
|
||||||
|
return key;
|
||||||
}
|
}
|
||||||
|
|
||||||
public static void refreshMpvKey(String url) {
|
public static void refreshMpvKey(String url) {
|
||||||
@@ -205,7 +213,7 @@ public class GalleryUtil {
|
|||||||
content = requests(mpvUrl, "get", header, null);
|
content = requests(mpvUrl, "get", header, null);
|
||||||
}catch (Exception e){
|
}catch (Exception e){
|
||||||
log.error("刷新mpvKey失败, url: {}", url, e);
|
log.error("刷新mpvKey失败, url: {}", url, e);
|
||||||
gid2MpvKey.put(parseGid(url) + "", null);
|
gid2MpvKey.remove(parseGid(url) + "");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
Document document = Jsoup.parse(content);
|
Document document = Jsoup.parse(content);
|
||||||
@@ -240,15 +248,38 @@ public class GalleryUtil {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public static String convertImg(String imagePath, String suffix){
|
public static String convertImg(String imagePath, String suffix){
|
||||||
Runtime rt = Runtime.getRuntime();
|
Path source = Path.of(imagePath);
|
||||||
|
Path target = source.resolveSibling(source.getFileName().toString().replaceFirst("\\Q" + suffix + "\\E$", ".avif"));
|
||||||
|
if (source.equals(target)) return imagePath;
|
||||||
|
Path temporary = null;
|
||||||
|
Process process = null;
|
||||||
try {
|
try {
|
||||||
Process exec = rt.exec(new String[]{"convert", imagePath, imagePath.replace(suffix, ".avif")});
|
temporary = Files.createTempFile(target.toAbsolutePath().getParent(), ".convert-", ".avif");
|
||||||
exec.waitFor();
|
process = new ProcessBuilder("convert", source.toString(), temporary.toString())
|
||||||
boolean ignored = new File(imagePath).delete();
|
.redirectErrorStream(true).redirectOutput(ProcessBuilder.Redirect.DISCARD).start();
|
||||||
return imagePath.replace(suffix, ".avif");
|
if (!process.waitFor(60, TimeUnit.SECONDS))
|
||||||
} catch (IOException| InterruptedException e) {
|
throw new IOException("图片转换超时");
|
||||||
log.error("文件{}转换失败", imagePath, e);
|
if (process.exitValue() != 0 || Files.size(temporary) == 0)
|
||||||
|
throw new IOException("图片转换失败");
|
||||||
|
try {
|
||||||
|
Files.move(temporary, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
|
||||||
|
} catch (AtomicMoveNotSupportedException e) {
|
||||||
|
Files.move(temporary, target, StandardCopyOption.REPLACE_EXISTING);
|
||||||
|
}
|
||||||
|
Files.deleteIfExists(source);
|
||||||
|
return target.toString();
|
||||||
|
} catch (InterruptedException e) {
|
||||||
|
Thread.currentThread().interrupt();
|
||||||
return null;
|
return null;
|
||||||
|
} catch (IOException e) {
|
||||||
|
log.warn("文件{}转换失败", imagePath, e);
|
||||||
|
return null;
|
||||||
|
} finally {
|
||||||
|
if (process != null && process.isAlive()) process.destroyForcibly();
|
||||||
|
if (temporary != null) {
|
||||||
|
try { Files.deleteIfExists(temporary); }
|
||||||
|
catch (IOException e) { log.warn("清理图片转换临时文件失败", e); }
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -315,22 +346,21 @@ public class GalleryUtil {
|
|||||||
}
|
}
|
||||||
httpResponse = httpClient.execute(httpPost);
|
httpResponse = httpClient.execute(httpPost);
|
||||||
}
|
}
|
||||||
HttpEntity responseEntity = httpResponse.getEntity();
|
try (httpResponse) {
|
||||||
int statusCode = httpResponse.getStatusLine().getStatusCode();
|
HttpEntity responseEntity = httpResponse.getEntity();
|
||||||
StringBuilder stringBuilder = new StringBuilder();
|
int statusCode = httpResponse.getStatusLine().getStatusCode();
|
||||||
|
StringBuilder stringBuilder = new StringBuilder();
|
||||||
if(statusCode == 200){
|
if (statusCode == 200 && responseEntity != null) {
|
||||||
BufferedReader reader = new BufferedReader(new InputStreamReader(responseEntity.getContent()));
|
try (BufferedReader reader = new BufferedReader(new InputStreamReader(responseEntity.getContent()))) {
|
||||||
String str;
|
String str;
|
||||||
while((str = reader.readLine()) != null)
|
while ((str = reader.readLine()) != null)
|
||||||
stringBuilder.append(str).append("\n");
|
stringBuilder.append(str).append("\n");
|
||||||
} else{
|
}
|
||||||
log.warn("{}:{}", url, statusCode);
|
} else {
|
||||||
|
log.warn("{}:{}", url, statusCode);
|
||||||
|
}
|
||||||
|
return stringBuilder.toString();
|
||||||
}
|
}
|
||||||
|
|
||||||
httpResponse.close();
|
|
||||||
|
|
||||||
return stringBuilder.toString();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public static Integer parseGid(String link){
|
public static Integer parseGid(String link){
|
||||||
|
|||||||
@@ -0,0 +1,67 @@
|
|||||||
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
|
import java.io.*;
|
||||||
|
import java.net.URI;
|
||||||
|
import java.net.URLConnection;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.Locale;
|
||||||
|
import java.util.concurrent.Callable;
|
||||||
|
|
||||||
|
/** Downloads into private temporary files and publishes only completed images. */
|
||||||
|
public final class ImageFileCache {
|
||||||
|
private static final SingleFlight<Path, Path> downloads = new SingleFlight<>();
|
||||||
|
|
||||||
|
private ImageFileCache() { }
|
||||||
|
|
||||||
|
public static Path find(Path directory, String name) {
|
||||||
|
for (String suffix : new String[]{".avif", ".gif"}) {
|
||||||
|
Path path = directory.resolve(name + suffix);
|
||||||
|
if (Files.isRegularFile(path) && path.toFile().length() > 0)
|
||||||
|
return path;
|
||||||
|
}
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
|
||||||
|
public static Path get(Path directory, String name, Callable<String> sourceUrl) throws Exception {
|
||||||
|
Path key = directory.resolve(name).toAbsolutePath().normalize();
|
||||||
|
return downloads.run(key, () -> {
|
||||||
|
Path cached = find(directory, name);
|
||||||
|
if (cached != null) return cached;
|
||||||
|
Files.createDirectories(directory);
|
||||||
|
String url = sourceUrl.call();
|
||||||
|
if (url == null) throw new IOException("图片地址不存在");
|
||||||
|
URI source = new URI(url);
|
||||||
|
boolean gif = source.getPath().toLowerCase(Locale.ROOT).endsWith(".gif");
|
||||||
|
String suffix = gif ? ".gif" : ".img";
|
||||||
|
Path temporary = Files.createTempFile(directory, ".download-", suffix);
|
||||||
|
Path converted = null;
|
||||||
|
try {
|
||||||
|
URLConnection connection = source.toURL().openConnection();
|
||||||
|
connection.setConnectTimeout(5_000);
|
||||||
|
connection.setReadTimeout(15_000);
|
||||||
|
try (InputStream input = connection.getInputStream();
|
||||||
|
OutputStream output = Files.newOutputStream(temporary)) {
|
||||||
|
input.transferTo(output);
|
||||||
|
}
|
||||||
|
if (Files.size(temporary) == 0) throw new IOException("图片内容为空");
|
||||||
|
if (gif) {
|
||||||
|
converted = temporary;
|
||||||
|
} else {
|
||||||
|
String result = GalleryUtil.convertImg(temporary.toString(), suffix);
|
||||||
|
if (result == null) throw new IOException("图片转换失败");
|
||||||
|
converted = Path.of(result);
|
||||||
|
}
|
||||||
|
Path target = directory.resolve(name + (gif ? ".gif" : ".avif"));
|
||||||
|
try {
|
||||||
|
Files.move(converted, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
|
||||||
|
} catch (AtomicMoveNotSupportedException e) {
|
||||||
|
Files.move(converted, target, StandardCopyOption.REPLACE_EXISTING);
|
||||||
|
}
|
||||||
|
return target;
|
||||||
|
} finally {
|
||||||
|
Files.deleteIfExists(temporary);
|
||||||
|
if (converted != null && !converted.equals(temporary)) Files.deleteIfExists(converted);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,30 @@
|
|||||||
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
|
import java.util.concurrent.*;
|
||||||
|
|
||||||
|
/** Concurrent callers for the same key share one in-flight operation, including its failure. */
|
||||||
|
public final class SingleFlight<K, V> {
|
||||||
|
private final ConcurrentHashMap<K, CompletableFuture<V>> running = new ConcurrentHashMap<>();
|
||||||
|
|
||||||
|
public V run(K key, Callable<V> operation) throws Exception {
|
||||||
|
CompletableFuture<V> mine = new CompletableFuture<>();
|
||||||
|
CompletableFuture<V> existing = running.putIfAbsent(key, mine);
|
||||||
|
if (existing != null) {
|
||||||
|
try { return existing.get(); }
|
||||||
|
catch (ExecutionException e) {
|
||||||
|
if (e.getCause() instanceof Exception cause) throw cause;
|
||||||
|
throw new IllegalStateException(e.getCause());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
V result = operation.call();
|
||||||
|
mine.complete(result);
|
||||||
|
return result;
|
||||||
|
} catch (Exception | Error e) {
|
||||||
|
mine.completeExceptionally(e);
|
||||||
|
throw e;
|
||||||
|
} finally {
|
||||||
|
running.remove(key, mine);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,57 @@
|
|||||||
|
package com.lion.lionwebsite.Service;
|
||||||
|
|
||||||
|
import com.lion.lionwebsite.Dao.normal.*;
|
||||||
|
import com.lion.lionwebsite.Dao.cache.ImageCacheMapper;
|
||||||
|
import com.lion.lionwebsite.Domain.*;
|
||||||
|
import com.lion.lionwebsite.Util.GalleryUtil;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import java.util.concurrent.atomic.AtomicReference;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
import static org.mockito.Mockito.*;
|
||||||
|
|
||||||
|
class GallerySubmissionTest {
|
||||||
|
@Test
|
||||||
|
void immediateCompletionSeesPersistedTask() throws Exception {
|
||||||
|
checkSubmission((byte) 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void timeoutRetainsTaskForRetry() throws Exception {
|
||||||
|
checkSubmission((byte) -1);
|
||||||
|
}
|
||||||
|
|
||||||
|
private void checkSubmission(byte ack) throws Exception {
|
||||||
|
GalleryMapper galleries = mock(GalleryMapper.class);
|
||||||
|
UserMapper users = mock(UserMapper.class);
|
||||||
|
RemoteService remote = mock(RemoteService.class);
|
||||||
|
CustomConfigurationMapper configuration = mock(CustomConfigurationMapper.class);
|
||||||
|
GalleryManageService service = new GalleryManageService(galleries, mock(CollectMapper.class), configuration,
|
||||||
|
users, mock(ShareFileMapper.class), mock(ImageCacheMapper.class), remote, mock(PushService.class));
|
||||||
|
User user = new User();
|
||||||
|
user.setId(7);
|
||||||
|
user.setUsername("test");
|
||||||
|
when(users.selectUserByAuthCode("test")).thenReturn(user);
|
||||||
|
AtomicReference<Gallery> saved = new AtomicReference<>();
|
||||||
|
when(galleries.selectGalleryByGid(123)).thenAnswer(call -> saved.get());
|
||||||
|
doAnswer(call -> { saved.set(call.getArgument(0)); return null; }).when(galleries).insertGallery(any());
|
||||||
|
Gallery gallery = new Gallery();
|
||||||
|
gallery.setGid(123);
|
||||||
|
gallery.setName("sample [123]");
|
||||||
|
gallery.setStatus("已提交");
|
||||||
|
when(remote.addGalleryToQueue(any())).thenAnswer(call -> {
|
||||||
|
assertNotNull(saved.get(), "node status must find a persisted record");
|
||||||
|
assertEquals(7, saved.get().getDownloader());
|
||||||
|
if (ack == 0) saved.get().setStatus("下载完成");
|
||||||
|
return ack;
|
||||||
|
});
|
||||||
|
try (var parser = mockStatic(GalleryUtil.class)) {
|
||||||
|
parser.when(() -> GalleryUtil.parse("https://example.org/g/123/key/", true, "original")).thenReturn(gallery);
|
||||||
|
String response = service.createTask("https://example.org/g/123/key/", "original", "test");
|
||||||
|
assertNotNull(saved.get());
|
||||||
|
assertTrue(response.contains(ack == 0 ? "success" : "任务已保存"));
|
||||||
|
assertEquals(ack == 0 ? "下载完成" : "已提交", saved.get().getStatus());
|
||||||
|
verify(galleries, never()).deleteGalleryByGid(any());
|
||||||
|
verify(galleries, times(1)).insertGallery(gallery);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,27 @@
|
|||||||
|
package com.lion.lionwebsite.Service;
|
||||||
|
|
||||||
|
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.HashMap;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
class PersonalArchiveTest {
|
||||||
|
@Test void archiveContainsCompleteFilesAndCanBeOpenedImmediately(@TempDir Path root) throws Exception {
|
||||||
|
Path source = Files.createDirectory(root.resolve("source"));
|
||||||
|
Files.createDirectory(source.resolve("nested"));
|
||||||
|
for (int i = 0; i < 100; i++) Files.writeString(source.resolve("nested/" + i + ".txt"), "content-" + i);
|
||||||
|
Path archive = root.resolve("result.tar");
|
||||||
|
PersonalService.writeTar(source, archive);
|
||||||
|
var contents = new HashMap<String, String>();
|
||||||
|
try (var input = new TarArchiveInputStream(Files.newInputStream(archive))) {
|
||||||
|
org.apache.commons.compress.archivers.tar.TarArchiveEntry entry;
|
||||||
|
while ((entry = input.getNextTarEntry()) != null) {
|
||||||
|
if (!entry.isDirectory()) contents.put(entry.getName(), new String(input.readAllBytes(), java.nio.charset.StandardCharsets.UTF_8));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
assertEquals(100, contents.size());
|
||||||
|
for (int i = 0; i < 100; i++) assertEquals("content-" + i, contents.get("nested/" + i + ".txt"));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,62 @@
|
|||||||
|
package com.lion.lionwebsite.Service;
|
||||||
|
|
||||||
|
import com.lion.lionwebsite.Dao.normal.GalleryMapper;
|
||||||
|
import com.lion.lionwebsite.Message.*;
|
||||||
|
import io.netty.channel.*;
|
||||||
|
import io.netty.channel.embedded.EmbeddedChannel;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import java.io.IOException;
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
import static org.mockito.Mockito.*;
|
||||||
|
|
||||||
|
class RemoteServiceTest {
|
||||||
|
private static ResponseMessage response(int messageId, byte result) {
|
||||||
|
ResponseMessage message = new ResponseMessage();
|
||||||
|
message.setMessageId(messageId);
|
||||||
|
message.setResult(result);
|
||||||
|
return message;
|
||||||
|
}
|
||||||
|
|
||||||
|
private RemoteService service() {
|
||||||
|
return new RemoteService(mock(GalleryMapper.class), mock(PushService.class),
|
||||||
|
mock(WebSocketService.class), mock(SubscriptionStandbySnapshotService.class));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void immediateResponseHasRegisteredWaiter() {
|
||||||
|
RemoteService service = service();
|
||||||
|
EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
|
||||||
|
@Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
|
||||||
|
AbstractMessage request = (AbstractMessage) msg;
|
||||||
|
ctx.fireChannelRead(response(request.messageId, (byte) 0));
|
||||||
|
promise.setSuccess();
|
||||||
|
}
|
||||||
|
}, service.new MyChannelInboundHandlerAdapter());
|
||||||
|
service.channel = channel;
|
||||||
|
try {
|
||||||
|
assertEquals(0, service.checkAvailability());
|
||||||
|
assertTrue(service.promiseHashMap.isEmpty());
|
||||||
|
} finally { service.shutdownResources(); channel.finishAndReleaseAll(); }
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void timeoutAndWriteFailureRemoveWaiters() {
|
||||||
|
RemoteService service = service();
|
||||||
|
EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
|
||||||
|
@Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
|
||||||
|
promise.setSuccess(); // no reply
|
||||||
|
}
|
||||||
|
});
|
||||||
|
service.channel = channel;
|
||||||
|
try {
|
||||||
|
assertEquals(-1, service.sendRequest(new AvailableCheckMessage(), 1, TimeUnit.MILLISECONDS));
|
||||||
|
assertTrue(service.promiseHashMap.isEmpty());
|
||||||
|
channel.pipeline().addLast(new ChannelOutboundHandlerAdapter() {
|
||||||
|
@Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
|
||||||
|
promise.setFailure(new IOException("test failure"));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
assertEquals(-1, service.sendRequest(new AvailableCheckMessage(), 1, TimeUnit.SECONDS));
|
||||||
|
assertTrue(service.promiseHashMap.isEmpty());
|
||||||
|
} finally { service.shutdownResources(); channel.finishAndReleaseAll(); }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,53 @@
|
|||||||
|
package com.lion.lionwebsite.Service;
|
||||||
|
|
||||||
|
import com.lion.lionwebsite.Dao.normal.SubMapper;
|
||||||
|
import com.lion.lionwebsite.Domain.SubscriptionAccount;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.List;
|
||||||
|
import java.util.concurrent.*;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
import static org.mockito.Mockito.*;
|
||||||
|
|
||||||
|
class SubscriptionRefreshServiceTest {
|
||||||
|
@Test
|
||||||
|
void stalledDownloadDoesNotHoldStateLockAndStaleResultIsDiscarded(@TempDir Path directory) throws Exception {
|
||||||
|
SubMapper mapper = mock(SubMapper.class);
|
||||||
|
SubscriptionAccount original = new SubscriptionAccount(1, "sample", "old", false, true,
|
||||||
|
null, null, null, null, 0, null, null);
|
||||||
|
SubscriptionAccount changed = new SubscriptionAccount(1, "sample", "new", false, true,
|
||||||
|
null, null, null, null, 0, null, null);
|
||||||
|
when(mapper.selectSubscriptionAccount(1)).thenReturn(original);
|
||||||
|
var coordinator = new SubscriptionStateCoordinator();
|
||||||
|
CountDownLatch started = new CountDownLatch(1);
|
||||||
|
CountDownLatch release = new CountDownLatch(1);
|
||||||
|
var service = new SubscriptionRefreshService(mapper, coordinator) {
|
||||||
|
@Override List<String> download(String url) throws java.io.IOException {
|
||||||
|
started.countDown();
|
||||||
|
try {
|
||||||
|
if (!release.await(5, TimeUnit.SECONDS)) throw new java.io.IOException("test timed out");
|
||||||
|
} catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new java.io.IOException(e); }
|
||||||
|
return url.contains("v2") ? List.of("bm9kZQ==") : List.of("proxies:");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
service.v2UrlTemplate = "https://example.invalid/v2/{key}";
|
||||||
|
service.clashUrlTemplate = "https://example.invalid/clash/{key}";
|
||||||
|
service.cacheRoot = directory.toString();
|
||||||
|
ExecutorService worker = Executors.newSingleThreadExecutor();
|
||||||
|
try {
|
||||||
|
Future<Boolean> refresh = worker.submit(() -> service.refresh(1));
|
||||||
|
assertTrue(started.await(2, TimeUnit.SECONDS));
|
||||||
|
var lock = coordinator.writeLock();
|
||||||
|
assertTrue(lock.tryLock(1, TimeUnit.SECONDS), "management must remain available during downloads");
|
||||||
|
try {
|
||||||
|
when(mapper.selectSubscriptionAccount(1)).thenReturn(changed);
|
||||||
|
service.invalidateCache(1);
|
||||||
|
} finally { lock.unlock(); }
|
||||||
|
release.countDown();
|
||||||
|
assertFalse(refresh.get(2, TimeUnit.SECONDS));
|
||||||
|
assertFalse(service.hasCompleteCache(1));
|
||||||
|
verify(mapper, never()).markSubscriptionRefreshSuccess(any());
|
||||||
|
} finally { release.countDown(); worker.shutdownNow(); }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,62 @@
|
|||||||
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
|
import org.springframework.mock.web.MockHttpServletRequest;
|
||||||
|
import org.springframework.mock.web.MockHttpServletResponse;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.Arrays;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
class FileDownloadTest {
|
||||||
|
@TempDir Path directory;
|
||||||
|
|
||||||
|
private MockHttpServletResponse download(String range, int size, String method) throws Exception {
|
||||||
|
byte[] bytes = new byte[size];
|
||||||
|
for (int i = 0; i < size; i++) bytes[i] = (byte) i;
|
||||||
|
Path file = directory.resolve("sample.bin");
|
||||||
|
Files.write(file, bytes);
|
||||||
|
MockHttpServletRequest request = new MockHttpServletRequest(method, "/file");
|
||||||
|
if (range != null) request.addHeader("Range", range);
|
||||||
|
MockHttpServletResponse response = new MockHttpServletResponse();
|
||||||
|
FileDownload.export(request, response, file.toString());
|
||||||
|
return response;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void smallRangeDoesNotOverread() throws Exception {
|
||||||
|
var response = download("bytes=10-109", 10_000, "GET");
|
||||||
|
assertEquals(206, response.getStatus());
|
||||||
|
assertEquals(100, response.getContentAsByteArray().length);
|
||||||
|
assertEquals("bytes 10-109/10000", response.getHeader("Content-Range"));
|
||||||
|
assertEquals(10, response.getContentAsByteArray()[0]);
|
||||||
|
assertEquals(109, response.getContentAsByteArray()[99]);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void supportsSuffixAndOpenEndedRanges() throws Exception {
|
||||||
|
var suffix = download("bytes=-10", 100, "GET");
|
||||||
|
assertEquals("bytes 90-99/100", suffix.getHeader("Content-Range"));
|
||||||
|
assertArrayEquals(download("bytes=90-", 100, "GET").getContentAsByteArray(), suffix.getContentAsByteArray());
|
||||||
|
assertEquals(10, suffix.getContentAsByteArray().length);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void clampsEndAndRejectsInvalidRanges() throws Exception {
|
||||||
|
assertEquals(10, download("bytes=90-999", 100, "GET").getContentAsByteArray().length);
|
||||||
|
for (String range : Arrays.asList("bytes=100-", "bytes=9-2", "bytes=-0", "bytes=oops")) {
|
||||||
|
var response = download(range, 100, "GET");
|
||||||
|
assertEquals(416, response.getStatus(), range);
|
||||||
|
assertEquals("bytes */100", response.getHeader("Content-Range"));
|
||||||
|
assertEquals(0, response.getContentAsByteArray().length);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void handlesFullEmptyHeadAndMultipleRanges() throws Exception {
|
||||||
|
var full = download(null, 100, "GET");
|
||||||
|
assertEquals(200, full.getStatus());
|
||||||
|
assertNull(full.getHeader("Content-Range"));
|
||||||
|
assertEquals(100, full.getContentAsByteArray().length);
|
||||||
|
assertEquals(0, download(null, 0, "GET").getContentAsByteArray().length);
|
||||||
|
assertEquals(416, download("bytes=0-", 0, "GET").getStatus());
|
||||||
|
assertEquals(0, download(null, 100, "HEAD").getContentAsByteArray().length);
|
||||||
|
assertEquals(200, download("bytes=0-1,5-6", 100, "GET").getStatus());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,38 @@
|
|||||||
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import java.io.IOException;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
import static org.mockito.Mockito.*;
|
||||||
|
import static org.mockito.ArgumentMatchers.*;
|
||||||
|
|
||||||
|
class GalleryKeyCacheTest {
|
||||||
|
@Test void coldCacheAllowsRefreshToPopulateIt() {
|
||||||
|
String url = "https://example.org/g/987654321/key/";
|
||||||
|
GalleryUtil.gid2MpvKey.remove("987654321");
|
||||||
|
try (var methods = mockStatic(GalleryUtil.class)) {
|
||||||
|
methods.when(() -> GalleryUtil.parseGid(url)).thenReturn(987654321);
|
||||||
|
methods.when(() -> GalleryUtil.getMpvKey(url)).thenCallRealMethod();
|
||||||
|
methods.when(() -> GalleryUtil.refreshMpvKey(url)).thenAnswer(call -> {
|
||||||
|
GalleryUtil.gid2MpvKey.put("987654321", "cached-key");
|
||||||
|
return null;
|
||||||
|
});
|
||||||
|
assertEquals("cached-key", GalleryUtil.getMpvKey(url));
|
||||||
|
assertEquals("cached-key", GalleryUtil.getMpvKey(url));
|
||||||
|
methods.verify(() -> GalleryUtil.refreshMpvKey(url), times(1));
|
||||||
|
} finally { GalleryUtil.gid2MpvKey.remove("987654321"); }
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void failedRefreshRemovesOldKeyWithoutInsertingNull() {
|
||||||
|
String url = "https://example.org/g/987654321/key/";
|
||||||
|
GalleryUtil.gid2MpvKey.put("987654321", "old-key");
|
||||||
|
try (var methods = mockStatic(GalleryUtil.class)) {
|
||||||
|
methods.when(() -> GalleryUtil.parseGid(url)).thenReturn(987654321);
|
||||||
|
methods.when(() -> GalleryUtil.refreshMpvKey(url)).thenCallRealMethod();
|
||||||
|
methods.when(() -> GalleryUtil.requests(anyString(), anyString(), any(), any()))
|
||||||
|
.thenThrow(new IOException("test failure"));
|
||||||
|
assertDoesNotThrow(() -> GalleryUtil.refreshMpvKey(url));
|
||||||
|
assertFalse(GalleryUtil.gid2MpvKey.containsKey("987654321"));
|
||||||
|
} finally { GalleryUtil.gid2MpvKey.remove("987654321"); }
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,63 @@
|
|||||||
|
package com.lion.lionwebsite.Util;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.junit.jupiter.api.io.TempDir;
|
||||||
|
import java.nio.file.*;
|
||||||
|
import java.util.concurrent.*;
|
||||||
|
import java.util.concurrent.atomic.*;
|
||||||
|
import java.time.Duration;
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
class ImageFileCacheTest {
|
||||||
|
@Test void publishesClosedGifAndReusesIt(@TempDir Path root) throws Exception {
|
||||||
|
Path source = root.resolve("source.gif");
|
||||||
|
byte[] content = "GIF89a test image".getBytes(java.nio.charset.StandardCharsets.UTF_8);
|
||||||
|
Files.write(source, content);
|
||||||
|
Path cache = root.resolve("cache");
|
||||||
|
Path result = ImageFileCache.get(cache, "1", () -> source.toUri().toString());
|
||||||
|
assertArrayEquals(content, Files.readAllBytes(result));
|
||||||
|
assertEquals(result, ImageFileCache.get(cache, "1", () -> { throw new AssertionError("cache must avoid source access"); }));
|
||||||
|
try (var files = Files.list(cache)) { assertEquals(1, files.count()); }
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void failedDownloadLeavesNoPublishedOrTemporaryFile(@TempDir Path root) throws Exception {
|
||||||
|
Path cache = root.resolve("cache");
|
||||||
|
assertThrows(Exception.class, () -> ImageFileCache.get(cache, "1", () -> root.resolve("missing.gif").toUri().toString()));
|
||||||
|
assertNull(ImageFileCache.find(cache, "1"));
|
||||||
|
try (var files = Files.list(cache)) { assertEquals(0, files.count()); }
|
||||||
|
Path source = root.resolve("retry.gif");
|
||||||
|
Files.writeString(source, "GIF89a retry");
|
||||||
|
assertNotNull(ImageFileCache.get(cache, "1", () -> source.toUri().toString()));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test void concurrentCallersShareTheSameOperation() throws Exception {
|
||||||
|
SingleFlight<String, String> flight = new SingleFlight<>();
|
||||||
|
CountDownLatch entered = new CountDownLatch(1);
|
||||||
|
CountDownLatch release = new CountDownLatch(1);
|
||||||
|
AtomicInteger calls = new AtomicInteger();
|
||||||
|
AtomicReference<Thread> followerThread = new AtomicReference<>();
|
||||||
|
ExecutorService workers = Executors.newFixedThreadPool(2);
|
||||||
|
try {
|
||||||
|
Future<String> leader = workers.submit(() -> flight.run("image", () -> {
|
||||||
|
calls.incrementAndGet(); entered.countDown();
|
||||||
|
if (!release.await(5, TimeUnit.SECONDS)) throw new IllegalStateException("test timed out");
|
||||||
|
return "completed";
|
||||||
|
}));
|
||||||
|
assertTrue(entered.await(2, TimeUnit.SECONDS));
|
||||||
|
Future<String> follower = workers.submit(() -> {
|
||||||
|
followerThread.set(Thread.currentThread());
|
||||||
|
return flight.run("image", () -> { calls.incrementAndGet(); return "duplicate"; });
|
||||||
|
});
|
||||||
|
assertTimeoutPreemptively(Duration.ofSeconds(2), () -> {
|
||||||
|
while (followerThread.get() == null || followerThread.get().getState() != Thread.State.WAITING) {
|
||||||
|
assertFalse(follower.isDone(), "follower must wait for the first operation");
|
||||||
|
Thread.sleep(1);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
release.countDown();
|
||||||
|
assertEquals("completed", leader.get(2, TimeUnit.SECONDS));
|
||||||
|
assertEquals("completed", follower.get(2, TimeUnit.SECONDS));
|
||||||
|
assertEquals(1, calls.get());
|
||||||
|
} finally { release.countDown(); workers.shutdownNow(); }
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user