diff --git a/FileService.Application/EventHandler/UploadTaskCompleteEventHandler.cs b/FileService.Application/EventHandler/UploadTaskCompleteEventHandler.cs new file mode 100644 index 0000000..11c2dbc --- /dev/null +++ b/FileService.Application/EventHandler/UploadTaskCompleteEventHandler.cs @@ -0,0 +1,45 @@ +using FileService.Application.Ports; +using IM.Commons.IntegrationEvents; +using MassTransit; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace FileService.Application.EventHandler +{ + public class UploadTaskCompleteEventHandler : IConsumer + { + private readonly IObjectStorageRouter router; + private readonly IStorageRedisCache cache; + + public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IStorageRedisCache cache) + { + this.router = router; + this.cache = cache; + } + + public async Task Consume(ConsumeContext context) + { + var @event = context.Message; + var storage = router.Route(@event.ProviderCode); + if(@event.ProviderCode == "Local") + { + await storage.CompleteUploadAsync(new StorageContracts.CompleteUploadCommand( + ProviderCode: @event.ProviderCode, + Bucket: @event.Bucket, + Region: @event.Region, + ObjectKey: @event.ObjectKey, + UploadSessionId: @event.SessionId, + Parts: @event.Parts.Select(s => new StorageContracts.UploadPart( + PartNumber: s.PartNumber, + ETag: s.ETag, + Size: s.Size, + Checksum: s.Checksum + )).ToList() + ), context.CancellationToken); + } + } + } +} diff --git a/FileService.Application/UploadFileTask/UploadFileTaskService.cs b/FileService.Application/UploadFileTask/UploadFileTaskService.cs index cba587d..440dd63 100644 --- a/FileService.Application/UploadFileTask/UploadFileTaskService.cs +++ b/FileService.Application/UploadFileTask/UploadFileTaskService.cs @@ -4,7 +4,9 @@ using FileService.Application.StorageContracts; using FileService.Domain.Entities; using FileService.Domain.IReposities; using IM.Commons; +using IM.Commons.IntegrationEvents; using IM.InitCommon; +using MassTransit; using MassTransit.Internals; using Microsoft.Extensions.Options; using System; @@ -17,13 +19,16 @@ namespace FileService.Application.UploadFileTask { public class UploadFileTaskService(IUploadTaskReposity reposity, IMapper mapper, IObjectStorageRouter router, - IOptions options, IStorageRedisCache redis) + IOptions options, IStorageRedisCache redis, + IPublishEndpoint endpoint + ) { private readonly IUploadTaskReposity reposity = reposity; private readonly IMapper mapper = mapper; private readonly IObjectStorageRouter router = router; private readonly IOptions options = options; private readonly IStorageRedisCache redis = redis; + private readonly IPublishEndpoint endpoint = endpoint; private readonly IObjectStoragePort storage = router.Route(options.Value.DefaultProviderCode); public async Task> InitTaskAsync(UploadTaskInitCommand command) @@ -34,9 +39,9 @@ namespace FileService.Application.UploadFileTask var storageOption = options.Value.Providers[options.Value.DefaultProviderCode]; var storage = router.Route(storageOption.ProviderCode); var initRes = await storage.InitUploadAsync(new StorageContracts.InitiateUploadCommand( - ProviderCode: options.Value.ProviderCode, - Bucket: options.Value.Bucket, - ObjectKey: options.Value.Endpoint, + ProviderCode: storageOption.ProviderCode, + Bucket: storageOption.Bucket, + ObjectKey: storageOption.Endpoint, ContentType:task.ContentType.Value, ContentLength: command.FileSize, null @@ -95,20 +100,35 @@ namespace FileService.Application.UploadFileTask } 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); + //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, taskCache.ObjectKey, taskCache.Region )); + await endpoint.Publish(new UploadTaskCompleteEvent() + { + Bucket = taskCache.Bucket, + FileName = task.FileName.ToString(), + ObjectKey = taskCache.ObjectKey, + Parts = command.Parts.Select(s => + new IM.Commons.IntegrationEvents.UploadPart( + s.PartNumber, s.ETag, s.Size, s.Checksum) + ).ToList(), + ProviderCode = taskCache.ProviderCode, + Region = taskCache.Region, + SessionId = command.UploadSessionId, + TaskId = task.Id + }); + return Result.Success(mapper.Map(task)); } } diff --git a/FileService.Infrastructure/ModuleInit.cs b/FileService.Infrastructure/ModuleInit.cs index 73c0587..3b702e4 100644 --- a/FileService.Infrastructure/ModuleInit.cs +++ b/FileService.Infrastructure/ModuleInit.cs @@ -14,6 +14,7 @@ namespace FileService.Infrastructure services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); } diff --git a/FileService.Infrastructure/Storage/LocalStorageAdapter.cs b/FileService.Infrastructure/Storage/LocalStorageAdapter.cs index d0e7a44..6e66b66 100644 --- a/FileService.Infrastructure/Storage/LocalStorageAdapter.cs +++ b/FileService.Infrastructure/Storage/LocalStorageAdapter.cs @@ -3,27 +3,73 @@ using FileService.Application.StorageContracts; using FileService.Domain.ValueObjects; using IM.Commons; using IM.InitCommon; -using MassTransit; -using MassTransit.Configuration; using Microsoft.Extensions.Options; -using System; -using System.Collections.Generic; -using System.Linq; -using System.Text; -using System.Threading.Tasks; +using StackExchange.Redis; namespace FileService.Infrastructure.Storage { - public class LocalStorageAdapter(IRedisService redis, IOptions options) : IObjectStoragePort + public class LocalStorageAdapter(IStorageRedisCache redis, IOptions options) : IObjectStoragePort { - private readonly IRedisService redis = redis; + private readonly IStorageRedisCache redis = redis; private readonly IOptions options = options; public string ProviderCode => "Local"; - public Task CompleteUploadAsync(CompleteUploadCommand command, CancellationToken token) + public async Task CompleteUploadAsync(CompleteUploadCommand command, CancellationToken token) { - throw new NotImplementedException(); + var res = await MergeAsync(command.UploadSessionId, command.ObjectKey, command.Parts); + return new CompleteUploadResult(new StorageLocation( + storageProvider: command.ProviderCode, + bucket: command.Bucket, + objectKey: command.ObjectKey, + region: command.Region + ), null, command.Parts.Sum(x => x.Size).Value); + } + public async Task> MergeAsync(string sessionId, string objectKey, IReadOnlyList parts) + { + var rootPath = options.Value.Providers[options.Value.DefaultProviderCode].LocalRootPath; + var tempPath = Path.Combine(rootPath, "temp"); // 项目根目录下 uploads // 最终文件存储路径（这里可以用你之前 ObjectNameGenerator 生成的名字） + var finalPath = Path.Combine(rootPath, objectKey); + var finalDir = Path.GetDirectoryName(finalPath); + Directory.CreateDirectory(finalDir); + + var storageCache = await redis.GetAsync(sessionId); + var totalChunks = storageCache.TotalPartCount; + try + { + using (var finalStream = new FileStream(finalPath, FileMode.Create)) + { + for (var i = 1; i <= totalChunks; i++) + { + var progress = (i * 100.0 / totalChunks); + if (i % 5 == 0 || i == totalChunks) + { + //await _redis.HashSetAsync(RedisKeys.MergeStatus(taskId), new HashEntry[] + //{ + // new("status", "processing"), + // new("progress", progress.ToString("F2")) + //}); + } + var chunkPath = Path.Combine(tempPath, $"{i}.part.tmp"); + if (!File.Exists(chunkPath)) + return Result.Fail(ResultCode.CHUNK_NOT_FOUND); + using (var chunkStream = new FileStream(chunkPath, FileMode.Open)) + { + await chunkStream.CopyToAsync(finalStream); + } + } + Directory.Delete(tempPath, true); + await redis.DeleteAsync(sessionId); + } + + return Result.Success(); + } + catch (Exception e) + { + //_logger.LogError(e, e.Message); + throw; + + } } public async Task GenerateUploadUrlAsync(GenerateUploadUrlCommand command, CancellationToken token) diff --git a/IM.Commons/IntegrationEvents/UploadTaskCompleteEvent.cs b/IM.Commons/IntegrationEvents/UploadTaskCompleteEvent.cs new file mode 100644 index 0000000..6b10ea0 --- /dev/null +++ b/IM.Commons/IntegrationEvents/UploadTaskCompleteEvent.cs @@ -0,0 +1,25 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace IM.Commons.IntegrationEvents +{ + public class UploadTaskCompleteEvent + { + public Guid TaskId { get; set; } + public string SessionId { get; set; } + public string FileName { get; set; } + public string ProviderCode { get; set; } + public string Bucket { get; set; } + public string Region { get; set; } + public string ObjectKey { get; set; } + public IReadOnlyList Parts { get; set; } + } + public sealed record UploadPart( + int PartNumber, + string ETag, + long? Size = null, + string? Checksum = null); +} diff --git a/MessageService.WebApi/Application/Message/MessageService.cs b/MessageService.WebApi/Application/Message/MessageService.cs index 3d1d24b..4ebdfbf 100644 --- a/MessageService.WebApi/Application/Message/MessageService.cs +++ b/MessageService.WebApi/Application/Message/MessageService.cs @@ -73,8 +73,6 @@ namespace MessageService.WebApi.Application.Message { // 2. 处理引用逻辑（如果传了 QuoteMessageId） QuoteInfo? quote = null; - if (command.QuoteMessageId.HasValue) - { var originMsg = await reposity.FindByIdAsync(command.QuoteMessageId.Value); if (originMsg != null) { @@ -89,9 +87,6 @@ namespace MessageService.WebApi.Application.Message } message.WithQuote(quote); } - - - reposity.Create(message); return Result.Success(mapper.Map(message));