clean up some disposes, remove channel and use async ienumberable for upload as well
This commit is contained in:
@@ -40,6 +40,7 @@ public class CachedPlayersManager : IDisposable
|
||||
_apiController.PairedWithOther += ApiControllerOnPairedWithOther;
|
||||
_apiController.UnpairedFromOther += ApiControllerOnUnpairedFromOther;
|
||||
_apiController.Disconnected += ApiControllerOnDisconnected;
|
||||
|
||||
_ipcManager.PenumbraDisposed += IpcManagerOnPenumbraDisposed;
|
||||
|
||||
if (clientState.IsLoggedIn)
|
||||
@@ -85,8 +86,12 @@ public class CachedPlayersManager : IDisposable
|
||||
_apiController.PairedClientOffline -= ApiControllerOnPairedClientOffline;
|
||||
_apiController.PairedWithOther -= ApiControllerOnPairedWithOther;
|
||||
_apiController.UnpairedFromOther -= ApiControllerOnUnpairedFromOther;
|
||||
_apiController.Disconnected -= ApiControllerOnDisconnected;
|
||||
|
||||
_ipcManager.PenumbraDisposed -= ApiControllerOnDisconnected;
|
||||
|
||||
_framework.Update -= FrameworkOnUpdate;
|
||||
|
||||
_clientState.Login -= ClientStateOnLogin;
|
||||
_clientState.Logout -= ClientStateOnLogout;
|
||||
}
|
||||
|
||||
@@ -49,9 +49,11 @@ namespace MareSynchronos.Managers
|
||||
{
|
||||
Logger.Debug("Disposing " + nameof(PlayerManager));
|
||||
|
||||
_ipcManager.PenumbraRedrawEvent -= IpcManager_PenumbraRedrawEvent;
|
||||
_dalamudUtil.RemovePlayerFromWatch(_dalamudUtil.PlayerName);
|
||||
_apiController.Connected -= ApiController_Connected;
|
||||
_apiController.Disconnected -= ApiController_Disconnected;
|
||||
|
||||
_ipcManager.PenumbraRedrawEvent -= IpcManager_PenumbraRedrawEvent;
|
||||
_dalamudUtil.PlayerChanged -= Watcher_PlayerChanged;
|
||||
}
|
||||
|
||||
|
||||
@@ -169,8 +169,6 @@ public class CachedPlayer
|
||||
_ipcManager.GlamourerApplyOnlyCustomization(_originalGlamourerData, PlayerName);
|
||||
_ipcManager.GlamourerApplyOnlyEquipment(_lastGlamourerData, PlayerName);
|
||||
}
|
||||
|
||||
IsVisible = false;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
@@ -183,6 +181,7 @@ public class CachedPlayer
|
||||
_apiController.CharacterReceived -= ApiControllerOnCharacterReceived;
|
||||
PlayerName = string.Empty;
|
||||
PlayerCharacter = null;
|
||||
IsVisible = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -6,7 +6,6 @@ using System.IO;
|
||||
using System.Linq;
|
||||
using System.Net.Http;
|
||||
using System.Threading;
|
||||
using System.Threading.Channels;
|
||||
using System.Threading.Tasks;
|
||||
using LZ4;
|
||||
using MareSynchronos.API;
|
||||
@@ -347,25 +346,18 @@ namespace MareSynchronos.WebAPI
|
||||
IsUploading = true;
|
||||
|
||||
Logger.Debug("Compressing files");
|
||||
Dictionary<string, byte[]> compressedFileData = new();
|
||||
foreach (var file in filesToUpload)
|
||||
{
|
||||
Logger.Debug(file);
|
||||
var data = await GetCompressedFileData(file, uploadToken);
|
||||
compressedFileData.Add(data.Item1, data.Item2);
|
||||
CurrentUploads[data.Item1] = (0, data.Item2.Length);
|
||||
}
|
||||
Logger.Debug("Files compressed, uploading files");
|
||||
foreach (var data in compressedFileData)
|
||||
{
|
||||
await UploadFile(data.Value, data.Key, uploadToken);
|
||||
if (uploadToken.IsCancellationRequested)
|
||||
{
|
||||
_ = UploadFile(data.Item2, file, uploadToken);
|
||||
if (!uploadToken.IsCancellationRequested) continue;
|
||||
PluginLog.Warning("Cancel in filesToUpload loop detected");
|
||||
CurrentUploads.Clear();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Logger.Debug("Upload tasks complete, waiting for server to confirm");
|
||||
var anyUploadsOpen = await _fileHub!.InvokeAsync<bool>("IsUploadFinished", uploadToken);
|
||||
Logger.Debug("Uploads open: " + anyUploadsOpen);
|
||||
@@ -423,6 +415,7 @@ namespace MareSynchronos.WebAPI
|
||||
return (fileHash, LZ4Codec.WrapHC(await File.ReadAllBytesAsync(fileCache.Filepath, uploadToken), 0,
|
||||
(int)new FileInfo(fileCache.Filepath).Length));
|
||||
}
|
||||
|
||||
private void UpdateLocalClientPairs(ClientPairDto dto, string characterIdentifier)
|
||||
{
|
||||
var entry = PairedClients.SingleOrDefault(e => e.OtherUID == dto.OtherUID);
|
||||
@@ -457,25 +450,22 @@ namespace MareSynchronos.WebAPI
|
||||
private async Task UploadFile(byte[] compressedFile, string fileHash, CancellationToken uploadToken)
|
||||
{
|
||||
if (uploadToken.IsCancellationRequested) return;
|
||||
|
||||
async IAsyncEnumerable<byte[]> AsyncFileData()
|
||||
{
|
||||
var chunkSize = 1024 * 512; // 512kb
|
||||
var chunks = (int)Math.Ceiling(compressedFile.Length / (double)chunkSize);
|
||||
var channel = Channel.CreateBounded<byte[]>(new BoundedChannelOptions(chunkSize)
|
||||
using var ms = new MemoryStream(compressedFile);
|
||||
var buffer = new byte[chunkSize];
|
||||
int bytesRead;
|
||||
while ((bytesRead = await ms.ReadAsync(buffer, 0, chunkSize, uploadToken)) > 0)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.Wait,
|
||||
AllowSynchronousContinuations = true,
|
||||
SingleReader = true,
|
||||
SingleWriter = true
|
||||
});
|
||||
await _fileHub!.SendAsync("UploadFile", fileHash, channel.Reader, uploadToken);
|
||||
for (var i = 0; i < chunks; i++)
|
||||
{
|
||||
var uploadChunk = compressedFile.Skip(i * chunkSize).Take(chunkSize).ToArray();
|
||||
channel.Writer.TryWrite(uploadChunk);
|
||||
CurrentUploads[fileHash] = (CurrentUploads[fileHash].Item1 + uploadChunk.Length, CurrentUploads[fileHash].Item2);
|
||||
if (uploadToken.IsCancellationRequested) break;
|
||||
CurrentUploads[fileHash] = (CurrentUploads[fileHash].Item1 + bytesRead, CurrentUploads[fileHash].Item2);
|
||||
uploadToken.ThrowIfCancellationRequested();
|
||||
yield return bytesRead == chunkSize ? buffer.ToArray() : buffer.Take(bytesRead).ToArray();
|
||||
}
|
||||
}
|
||||
|
||||
channel.Writer.Complete();
|
||||
await _fileHub!.SendAsync("UploadFileStreamAsync", fileHash, AsyncFileData(), uploadToken);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user