using FileService.Application.Ports; using FileService.Domain.IReposities; using IM.Application.Abstractions; 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 IUploadFileReposity uploadFile; private readonly IUnitOfWork uwork; private readonly IStorageRedisCache storageCache; public UploadTaskCompleteEventHandler(IObjectStorageRouter router, IUploadFileReposity uploadFile, IUnitOfWork uwork, IStorageRedisCache storageCache) { this.router = router; this.uploadFile = uploadFile; this.uwork = uwork; this.storageCache = storageCache; } public async Task Consume(ConsumeContext context) { var @event = context.Message; var storage = router.Route(@event.ProviderCode); var taskCache = await storageCache.GetAsync(@event.SessionId); 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); uploadFile.Create(new Domain.Entities.UploadFile( ownerId: @event.OperatorId, fileName: @event.FileName, fileSize: taskCache.FileSize, contentType: @event.ContentType, new Domain.ValueObjects.StorageLocation(taskCache.ProviderCode, taskCache.Bucket, taskCache.ObjectKey, taskCache.Region), checkSum: new Domain.ValueObjects.CheckSum("md5", @event.CheckSun) )); } } } }