nanxun/IM_NEW
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'master' of https://gitea.nxsir.cn/nanxun/IM_NEW

Browse Source
This commit is contained in:
nanxun
2026-08-31 14:42:22 +08:00
parent e605b50367 0692ba0276
commit dc7a9a18c7
72 changed files with 2479 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FileService.Application/UploadFileTask/UploadFileTaskService.cs
+87 -24
View File
@@ -10,10 +10,11 @@ using Microsoft.Extensions.Options;
namespace FileService.Application.UploadFileTask
{
public class UploadFileTaskService(IUploadTaskReposity reposity,
IMapper mapper, IObjectStorageRouter router,
public class UploadFileTaskService(IUploadTaskReposity reposity,
IMapper mapper, IObjectStorageRouter router,
IOptions<StorageOptions> options, IStorageRedisCache redis,
IPublishEndpoint endpoint, ILocalChunkStorage localChunkStorage
IPublishEndpoint endpoint, ILocalChunkStorage localChunkStorage,
IUploadFileReposity uploadFileReposity
)
{
private readonly IUploadTaskReposity reposity = reposity;
@@ -23,11 +24,33 @@ namespace FileService.Application.UploadFileTask
private readonly IStorageRedisCache redis = redis;
private readonly IPublishEndpoint endpoint = endpoint;
private readonly ILocalChunkStorage localChunkStorage = localChunkStorage;
private readonly IUploadFileReposity uploadFileReposity = uploadFileReposity;
private readonly IObjectStoragePort storage = router.Route(options.Value.DefaultProviderCode);
public async Task<Result<TaskInitResponse>> InitTaskAsync(UploadTaskInitCommand command)
{
CancellationToken cancellationToken = CancellationToken.None;
// 秒传：相同 checksum 的文件若已存在于已完成文件表，直接返回已有记录
var existingFile = await uploadFileReposity.FindByCheckSumGlobalAsync("md5", command.checkSum);
if (existingFile != null)
{
var storageForResponse = router.Route(existingFile.StorageLocation.StorageProvider);
return Result.Success(new TaskInitResponse
{
TaskId = existingFile.Id,
UploadSessionId = existingFile.Id.ToString(),
StorageLocation = existingFile.StorageLocation
});
}
// 文件大小上限校验
var maxSize = options.Value.Providers[options.Value.DefaultProviderCode].MaxObjectSizeBytes;
if (command.FileSize > maxSize)
{
return Result.Fail<TaskInitResponse>(ResultCode.FILE_TOO_LARGE);
}
var task = command.ToUploadTask();
var date = DateTime.Now;
var storageOption = options.Value.Providers[options.Value.DefaultProviderCode];
@@ -41,6 +64,10 @@ namespace FileService.Application.UploadFileTask
null);
var initRes = await storage.InitUploadAsync(initUpdateCommand, cancellationToken);
var totalPartCount = (int)(task.FileSize % storageOption.DefaultPartSizeBytes > 0 ?
(task.FileSize / storageOption.DefaultPartSizeBytes) + 1 :
task.FileSize / storageOption.DefaultPartSizeBytes);
var res = mapper.Map<TaskInitResponse>(initRes);
res.TaskId = task.Id;
@@ -55,9 +82,7 @@ namespace FileService.Application.UploadFileTask
region: storageOption.Region,
objectKey: initUpdateCommand.ObjectKey,
fileSize: task.FileSize,
totalPartCount: (int)(task.FileSize % storageOption.DefaultPartSizeBytes > 0 ?
(task.FileSize / storageOption.DefaultPartSizeBytes) + 1 :
task.FileSize / storageOption.DefaultPartSizeBytes)
totalPartCount: totalPartCount
));
return Result.Success(res);
@@ -67,14 +92,14 @@ namespace FileService.Application.UploadFileTask
public async Task<Result<PresignedUrl>> GenerateUrlAsync(string sessionId, int partNum, Guid userId, CancellationToken token = default)
{
var taskCache = await redis.GetAsync(sessionId);
if(taskCache is null)
if (taskCache is null)
{
return Result.Fail<PresignedUrl>(ResultCode.CHUNK_NOT_FOUND);
}
if(taskCache.TotalPartCount < partNum)
if (taskCache.TotalPartCount < partNum || partNum < 1)
{
return Result.Fail<PresignedUrl>(ResultCode.CHUNK_NOT_FOUND);
return Result.Fail<PresignedUrl>(ResultCode.INVALID_PART_NUMBER);
}
var presignUrl = await storage.GenerateUploadUrlAsync(new GenerateUploadUrlCommand(
@@ -93,26 +118,27 @@ namespace FileService.Application.UploadFileTask
{
var taskCache = await redis.GetAsync(command.UploadSessionId);
if(taskCache is null)
if (taskCache is null)
{
return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
}
if(taskCache.Parts.Count < taskCache.TotalPartCount)
// 校验分片数量必须匹配
if (command.Parts.Count != taskCache.TotalPartCount)
{
return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_COMBINE_FAIL);
return Result.Fail<UploadTaskResponse>(ResultCode.PART_COUNT_MISMATCH);
}
// 校验所有分片都已在上传缓存中注册
foreach (var part in command.Parts)
{
if (!taskCache.Parts.TryGetValue(part.PartNumber, out _))
{
return Result.Fail<UploadTaskResponse>(ResultCode.CHUNK_NOT_FOUND);
}
}
var task = await reposity.FindByIdAsync(Guid.Parse(taskCache.TaskId));
//var res = await storage.CompleteUploadAsync(new CompleteUploadCommand(
// ProviderCode: taskCache.ProviderCode,
// Bucket: taskCache.Bucket,
// Region: taskCache.Region,
// ObjectKey: taskCache.ObjectKey,
// UploadSessionId: taskCache.UploadSessionId,
// Parts: command.Parts
// ), cancellationToken);
task.CompleteUpload(new Domain.ValueObjects.StorageLocation(
taskCache.ProviderCode, taskCache.Bucket,
@@ -125,7 +151,7 @@ namespace FileService.Application.UploadFileTask
Bucket = taskCache.Bucket,
FileName = task.FileName.ToString(),
ObjectKey = taskCache.ObjectKey,
Parts = command.Parts.Select(s =>
Parts = command.Parts.Select(s =>
new IM.Commons.IntegrationEvents.UploadPart(
s.PartNumber, s.ETag, s.Size, s.Checksum)
).ToList(),
@@ -144,12 +170,22 @@ namespace FileService.Application.UploadFileTask
public async Task<Result<CompleteUploadResult>> UploadPartAsync(UploadPartCommand command)
{
var taskCache = await redis.GetAsync(command.SessionId);
if(taskCache is null)
if (taskCache is null)
{
return Result.Fail<CompleteUploadResult>(ResultCode.CHUNK_NOT_FOUND);
}
var minPartSize = options.Value.Providers[options.Value.DefaultProviderCode].MinPartSizeBytes;
// 最后一个分片豁免最小值校验（仅校验非最后一片）
var isLastPart = command.PartNum == taskCache.TotalPartCount;
if (!isLastPart && command.ContentLength < minPartSize)
{
return Result.Fail<CompleteUploadResult>(ResultCode.PART_TOO_SMALL,
$"分片 {command.PartNum} 大小为 {command.ContentLength} 字节，小于最小值 {minPartSize} 字节");
}
await localChunkStorage.SavePartAsync(new SaveLocalPartCommand(
UploadSessionId: command.SessionId,
PartNumber: command.PartNum,
@@ -166,5 +202,32 @@ namespace FileService.Application.UploadFileTask
);
return Result.Success(new CompleteUploadResult(location, command.PartNum.ToString(), command.ContentLength));
}
/// <summary>
/// 查询分片上传进度。
/// </summary>
public async Task<Result<UploadProgressResponse>> GetProgressAsync(string sessionId, Guid userId)
{
var taskCache = await redis.GetAsync(sessionId);
if (taskCache is null)
{
return Result.Fail<UploadProgressResponse>(ResultCode.CHUNK_NOT_FOUND);
}
var response = new UploadProgressResponse
{
SessionId = sessionId,
TaskId = taskCache.TaskId,
FileSize = taskCache.FileSize,
TotalPartCount = taskCache.TotalPartCount,
CompletedPartCount = taskCache.Parts.Count,
UploadedBytes = taskCache.UploadedBytes,
ProgressPercent = taskCache.FileSize > 0
? (int)(taskCache.UploadedBytes * 100 / taskCache.FileSize)
: 0
};
return Result.Success(response);
}
}
}
FileService.Application/UploadFileTask/UploadProgressResponse.cs
+16
View File
@@ -0,0 +1,16 @@
namespace FileService.Application.UploadFileTask
{
/// <summary>
/// 分片上传进度响应。
/// </summary>
public class UploadProgressResponse
{
public string SessionId { get; set; }
public string TaskId { get; set; }
public long FileSize { get; set; }
public int TotalPartCount { get; set; }
public int CompletedPartCount { get; set; }
public long UploadedBytes { get; set; }
public int ProgressPercent { get; set; }
}
}