t0w0bi/ClubPenguinClient
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

major refactoring, maybe some bugfixes, idk

Browse Source
This commit is contained in:
Stanley Dimant
2022-06-24 00:47:47 +02:00
parent 0fe3f1cf25
commit 2dcd02d170
22 changed files with 997 additions and 949 deletions
Download Patch File Download Diff File Expand all files Collapse all files

504
MareSynchronos/WebAPI/ApiController.cs
View File

@@ -11,35 +11,57 @@ using System.Threading.Tasks;
using LZ4;
using MareSynchronos.API;
using MareSynchronos.FileCacheDB;
using MareSynchronos.Utils;
using Microsoft.AspNetCore.SignalR.Client;
namespace MareSynchronos.WebAPI
{
public class CharacterReceivedEventArgs : EventArgs
{
public CharacterReceivedEventArgs(string characterNameHash, CharacterCacheDto characterData)
{
CharacterData = characterData;
CharacterNameHash = characterNameHash;
}
public CharacterCacheDto CharacterData { get; set; }
public string CharacterNameHash { get; set; }
}
public class ApiController : IDisposable
{
public const string MainServer = "Lunae Crescere Incipientis (Central Server EU)";
private readonly Configuration _pluginConfiguration;
public const string MainServiceUri = "https://darkarchon.internet-box.ch:5001";
public string UID { get; private set; } = string.Empty;
public string SecretKey => _pluginConfiguration.ClientSecret.ContainsKey(ApiUri) ? _pluginConfiguration.ClientSecret[ApiUri] : "-";
private string CacheFolder => _pluginConfiguration.CacheFolder;
public ConcurrentDictionary<string, (long, long)> CurrentUploads { get; } = new();
readonly CancellationTokenSource _cts;
private readonly Configuration _pluginConfiguration;
private HubConnection? _fileHub;
private HubConnection? _heartbeatHub;
private CancellationTokenSource? _uploadCancellationTokenSource;
private HubConnection? _userHub;
public ApiController(Configuration pluginConfiguration)
{
Logger.Debug("Creating " + nameof(ApiController));
_pluginConfiguration = pluginConfiguration;
_cts = new CancellationTokenSource();
_ = Heartbeat();
}
public event EventHandler<CharacterReceivedEventArgs>? CharacterReceived;
public event EventHandler? Connected;
public event EventHandler? Disconnected;
public event EventHandler? PairedClientOffline;
public event EventHandler? PairedClientOnline;
public event EventHandler? PairedWithOther;
public event EventHandler? UnpairedFromOther;
public ConcurrentDictionary<string, (long, long)> CurrentDownloads { get; } = new();
public ConcurrentDictionary<string, (long, long)> CurrentUploads { get; } = new();
public bool IsConnected => !string.IsNullOrEmpty(UID);
public bool IsDownloading { get; private set; } = false;
public bool IsUploading { get; private set; } = false;
public List<ClientPairDto> PairedClients { get; set; } = new();
public string SecretKey => _pluginConfiguration.ClientSecret.ContainsKey(ApiUri) ? _pluginConfiguration.ClientSecret[ApiUri] : "-";
public bool ServerAlive =>
(_heartbeatHub?.State ?? HubConnectionState.Disconnected) == HubConnectionState.Connected;
public string UID { get; private set; } = string.Empty;
public bool UseCustomService
{
get => _pluginConfiguration.UseCustomService;
@@ -49,43 +71,91 @@ namespace MareSynchronos.WebAPI
_pluginConfiguration.Save();
}
}
private string ApiUri => UseCustomService ? _pluginConfiguration.ApiUri : MainServiceUri;
public bool ServerAlive =>
(_heartbeatHub?.State ?? HubConnectionState.Disconnected) == HubConnectionState.Connected;
public bool IsConnected => !string.IsNullOrEmpty(UID);
public event EventHandler? Connected;
public event EventHandler? Disconnected;
public event EventHandler<CharacterReceivedEventArgs>? CharacterReceived;
public event EventHandler? UnpairedFromOther;
public event EventHandler? PairedWithOther;
public event EventHandler? PairedClientOnline;
public event EventHandler? PairedClientOffline;
public List<ClientPairDto> PairedClients { get; set; } = new();
readonly CancellationTokenSource cts;
private HubConnection? _heartbeatHub;
private HubConnection? _fileHub;
private HubConnection? _userHub;
private CancellationTokenSource? uploadCancellationTokenSource;
public ApiController(Configuration pluginConfiguration)
private string CacheFolder => _pluginConfiguration.CacheFolder;
public void CancelUpload()
{
this._pluginConfiguration = pluginConfiguration;
cts = new CancellationTokenSource();
if (_uploadCancellationTokenSource != null)
{
PluginLog.Warning("Cancelling upload");
_uploadCancellationTokenSource?.Cancel();
_fileHub!.InvokeAsync("AbortUpload");
}
}
_ = Heartbeat();
public void Dispose()
{
Logger.Debug("Disposing " + nameof(ApiController));
_cts?.Cancel();
_ = DisposeHubConnections();
}
public async Task<byte[]> DownloadFile(string hash)
{
IsDownloading = true;
var reader = await _fileHub!.StreamAsChannelAsync<byte[]>("DownloadFile", hash);
List<byte> downloadedData = new();
while (await reader.WaitToReadAsync())
{
while (reader.TryRead(out var data))
{
CurrentDownloads[hash] = (CurrentDownloads[hash].Item1 + data.Length, CurrentDownloads[hash].Item2);
downloadedData.AddRange(data);
//await Task.Delay(25);
}
}
IsDownloading = false;
return downloadedData.ToArray();
}
public async Task DownloadFiles(List<FileReplacementDto> fileReplacementDto)
{
foreach (var file in fileReplacementDto)
{
var fileSize = await _fileHub!.InvokeAsync<long>("GetFileSize", file.Hash);
CurrentDownloads[file.Hash] = (0, fileSize);
}
foreach (var file in fileReplacementDto.Where(f => CurrentDownloads[f.Hash].Item2 > 0))
{
var hash = file.Hash;
var data = await DownloadFile(hash);
var extractedFile = LZ4Codec.Unwrap(data);
var ext = file.GamePaths.First().Split(".", StringSplitOptions.None).Last();
var filePath = Path.Combine(_pluginConfiguration.CacheFolder, file.Hash + "." + ext);
await File.WriteAllBytesAsync(filePath, extractedFile);
await using (var db = new FileCacheContext())
{
db.Add(new FileCache
{
Filepath = filePath.ToLower(),
Hash = file.Hash,
LastModifiedDate = DateTime.Now.Ticks.ToString(),
});
await db.SaveChangesAsync();
}
Logger.Debug("File downloaded to " + filePath);
}
CurrentDownloads.Clear();
}
public async Task GetCharacterData(Dictionary<string, int> hashedCharacterNames)
{
await _userHub!.InvokeAsync("GetCharacterData",
hashedCharacterNames);
}
public async Task Heartbeat()
{
while (!ServerAlive && !cts.Token.IsCancellationRequested)
while (!ServerAlive && !_cts.Token.IsCancellationRequested)
{
try
{
PluginLog.Debug("Attempting to establish heartbeat connection to " + ApiUri);
Logger.Debug("Attempting to establish heartbeat connection to " + ApiUri);
_heartbeatHub = new HubConnectionBuilder()
.WithUrl(ApiUri + "/heartbeat", options =>
{
@@ -105,9 +175,9 @@ namespace MareSynchronos.WebAPI
#endif
}).Build();
await _heartbeatHub.StartAsync(cts.Token);
await _heartbeatHub.StartAsync(_cts.Token);
UID = await _heartbeatHub!.InvokeAsync<string>("Heartbeat");
PluginLog.Debug("Heartbeat started: " + ApiUri);
Logger.Debug("Heartbeat started: " + ApiUri);
try
{
await InitializeHubConnections();
@@ -121,7 +191,7 @@ namespace MareSynchronos.WebAPI
_heartbeatHub.Closed += OnHeartbeatHubOnClosed;
_heartbeatHub.Reconnected += OnHeartbeatHubOnReconnected;
PluginLog.Debug("Heartbeat established to: " + ApiUri);
Logger.Debug("Heartbeat established to: " + ApiUri);
}
catch (Exception ex)
{
@@ -130,46 +200,132 @@ namespace MareSynchronos.WebAPI
}
}
private async Task LoadInitialData()
public Task ReceiveCharacterData(CharacterCacheDto character, string characterHash)
{
var pairedClients = await _userHub!.InvokeAsync<List<ClientPairDto>>("GetPairedClients");
PairedClients = pairedClients.ToList();
Logger.Debug("Received DTO for " + characterHash);
CharacterReceived?.Invoke(null, new CharacterReceivedEventArgs(characterHash, character));
return Task.CompletedTask;
}
public async Task Register()
{
if (!ServerAlive) return;
Logger.Debug("Registering at service " + ApiUri);
var response = await _userHub!.InvokeAsync<string>("Register");
_pluginConfiguration.ClientSecret[ApiUri] = response;
_pluginConfiguration.Save();
RestartHeartbeat();
}
public void RestartHeartbeat()
{
PluginLog.Debug("Restarting heartbeat");
Logger.Debug("Restarting heartbeat");
_heartbeatHub!.Closed -= OnHeartbeatHubOnClosed;
_heartbeatHub!.Reconnected -= OnHeartbeatHubOnReconnected;
Task.Run(async () =>
{
await _heartbeatHub.StopAsync(cts.Token);
await _heartbeatHub.StopAsync(_cts.Token);
await _heartbeatHub.DisposeAsync();
_heartbeatHub = null!;
_ = Heartbeat();
});
}
private async Task OnHeartbeatHubOnReconnected(string? s)
public async Task SendCharacterData(CharacterCacheDto character, List<string> visibleCharacterIds)
{
PluginLog.Debug("Reconnected: " + ApiUri);
UID = await _heartbeatHub!.InvokeAsync<string>("Heartbeat");
if (!IsConnected || SecretKey == "-") return;
Logger.Debug("Sending Character data to service " + ApiUri);
CancelUpload();
_uploadCancellationTokenSource = new CancellationTokenSource();
var uploadToken = _uploadCancellationTokenSource.Token;
Logger.Debug("New Token Created");
var filesToUpload = await _fileHub!.InvokeAsync<List<string>>("SendFiles", character.FileReplacements.Select(c => c.Hash).Distinct(), uploadToken);
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)
{
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);
while (anyUploadsOpen && !uploadToken.IsCancellationRequested)
{
anyUploadsOpen = await _fileHub!.InvokeAsync<bool>("IsUploadFinished", uploadToken);
await Task.Delay(TimeSpan.FromSeconds(0.5), uploadToken);
Logger.Debug("Waiting for uploads to finish");
}
CurrentUploads.Clear();
IsUploading = false;
if (!uploadToken.IsCancellationRequested)
{
Logger.Debug("=== Pushing character data ===");
await _userHub!.InvokeAsync("PushCharacterData", character, visibleCharacterIds, uploadToken);
}
else
{
PluginLog.Warning("=== Upload operation was cancelled ===");
}
Logger.Debug("== Upload complete for " + character.JobId);
_uploadCancellationTokenSource = null;
}
private Task OnHeartbeatHubOnClosed(Exception? exception)
public async Task<List<string>> SendCharacterName(string hashedName)
{
PluginLog.Debug("Connection closed: " + ApiUri);
Disconnected?.Invoke(null, EventArgs.Empty);
RestartHeartbeat();
return Task.CompletedTask;
return await _userHub!.InvokeAsync<List<string>>("SendCharacterNameHash", hashedName);
}
public async Task SendPairedClientAddition(string uid)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientAddition", uid);
}
public async Task SendPairedClientPauseChange(string uid, bool paused)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientPauseChange", uid, paused);
}
public async Task SendPairedClientRemoval(string uid)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientRemoval", uid);
}
public async Task UpdateCurrentDownloadSize(string hash)
{
long fileSize = await _fileHub!.InvokeAsync<long>("GetFileSize", hash);
}
private async Task DisposeHubConnections()
{
if (_fileHub != null)
{
PluginLog.Debug("Disposing File Hub");
Logger.Debug("Disposing File Hub");
CancelUpload();
await _fileHub!.StopAsync();
await _fileHub!.DisposeAsync();
@@ -177,17 +333,25 @@ namespace MareSynchronos.WebAPI
if (_userHub != null)
{
PluginLog.Debug("Disposing User Hub");
Logger.Debug("Disposing User Hub");
await _userHub.StopAsync();
await _userHub.DisposeAsync();
}
}
private async Task<(string, byte[])> GetCompressedFileData(string fileHash, CancellationToken uploadToken)
{
await using var db = new FileCacheContext();
var fileCache = db.FileCaches.First(f => f.Hash == fileHash);
return (fileHash, LZ4Codec.WrapHC(await File.ReadAllBytesAsync(fileCache.Filepath, uploadToken), 0,
(int)new FileInfo(fileCache.Filepath).Length));
}
private async Task InitializeHubConnections()
{
await DisposeHubConnections();
PluginLog.Debug("Creating User Hub");
Logger.Debug("Creating User Hub");
_userHub = new HubConnectionBuilder()
.WithUrl(ApiUri + "/user", options =>
{
@@ -209,7 +373,7 @@ namespace MareSynchronos.WebAPI
_userHub.On<string>("RemoveOnlinePairedPlayer", (s) => PairedClientOffline?.Invoke(s, EventArgs.Empty));
_userHub.On<string>("AddOnlinePairedPlayer", (s) => PairedClientOnline?.Invoke(s, EventArgs.Empty));
PluginLog.Debug("Creating File Hub");
Logger.Debug("Creating File Hub");
_fileHub = new HubConnectionBuilder()
.WithUrl(ApiUri + "/files", options =>
{
@@ -225,9 +389,27 @@ namespace MareSynchronos.WebAPI
#endif
})
.Build();
await _fileHub.StartAsync(cts.Token);
await _fileHub.StartAsync(_cts.Token);
}
private async Task LoadInitialData()
{
var pairedClients = await _userHub!.InvokeAsync<List<ClientPairDto>>("GetPairedClients");
PairedClients = pairedClients.ToList();
}
private Task OnHeartbeatHubOnClosed(Exception? exception)
{
Logger.Debug("Connection closed: " + ApiUri);
Disconnected?.Invoke(null, EventArgs.Empty);
RestartHeartbeat();
return Task.CompletedTask;
}
private async Task OnHeartbeatHubOnReconnected(string? s)
{
Logger.Debug("Reconnected: " + ApiUri);
UID = await _heartbeatHub!.InvokeAsync<string>("Heartbeat");
}
private void UpdateLocalClientPairs(ClientPairDto dto, string characterIdentifier)
{
var entry = PairedClients.SingleOrDefault(e => e.OtherUID == dto.OtherUID);
@@ -258,15 +440,6 @@ namespace MareSynchronos.WebAPI
UnpairedFromOther?.Invoke(characterIdentifier, EventArgs.Empty);
}
}
private async Task<(string, byte[])> GetCompressedFileData(string fileHash, CancellationToken uploadToken)
{
await using var db = new FileCacheContext();
var fileCache = db.FileCaches.First(f => f.Hash == fileHash);
return (fileHash, LZ4Codec.WrapHC(await File.ReadAllBytesAsync(fileCache.Filepath, uploadToken), 0,
(int)new FileInfo(fileCache.Filepath).Length));
}
private async Task UploadFile(byte[] compressedFile, string fileHash, CancellationToken uploadToken)
{
if (uploadToken.IsCancellationRequested) return;
@@ -290,184 +463,17 @@ namespace MareSynchronos.WebAPI
channel.Writer.Complete();
}
}
public async Task Register()
public class CharacterReceivedEventArgs : EventArgs
{
public CharacterReceivedEventArgs(string characterNameHash, CharacterCacheDto characterData)
{
if (!ServerAlive) return;
PluginLog.Debug("Registering at service " + ApiUri);
var response = await _userHub!.InvokeAsync<string>("Register");
_pluginConfiguration.ClientSecret[ApiUri] = response;
_pluginConfiguration.Save();
RestartHeartbeat();
CharacterData = characterData;
CharacterNameHash = characterNameHash;
}
public void CancelUpload()
{
if (uploadCancellationTokenSource != null)
{
PluginLog.Warning("Cancelling upload");
uploadCancellationTokenSource?.Cancel();
_fileHub!.InvokeAsync("AbortUpload");
}
}
public async Task SendCharacterData(CharacterCacheDto character, List<string> visibleCharacterIds)
{
if (!IsConnected || SecretKey == "-") return;
PluginLog.Debug("Sending Character data to service " + ApiUri);
CancelUpload();
uploadCancellationTokenSource = new CancellationTokenSource();
var uploadToken = uploadCancellationTokenSource.Token;
PluginLog.Debug("New Token Created");
var filesToUpload = await _fileHub!.InvokeAsync<List<string>>("SendFiles", character.FileReplacements.Select(c => c.Hash).Distinct(), uploadToken);
IsUploading = true;
PluginLog.Debug("Compressing files");
Dictionary<string, byte[]> compressedFileData = new();
foreach (var file in filesToUpload)
{
PluginLog.Debug(file);
var data = await GetCompressedFileData(file, uploadToken);
compressedFileData.Add(data.Item1, data.Item2);
CurrentUploads[data.Item1] = (0, data.Item2.Length);
}
PluginLog.Debug("Files compressed, uploading files");
foreach (var data in compressedFileData)
{
await UploadFile(data.Value, data.Key, uploadToken);
if (uploadToken.IsCancellationRequested)
{
PluginLog.Warning("Cancel in filesToUpload loop detected");
CurrentUploads.Clear();
break;
}
}
PluginLog.Debug("Upload tasks complete, waiting for server to confirm");
var anyUploadsOpen = await _fileHub!.InvokeAsync<bool>("IsUploadFinished", uploadToken);
PluginLog.Debug("Uploads open: " + anyUploadsOpen);
while (anyUploadsOpen && !uploadToken.IsCancellationRequested)
{
anyUploadsOpen = await _fileHub!.InvokeAsync<bool>("IsUploadFinished", uploadToken);
await Task.Delay(TimeSpan.FromSeconds(0.5), uploadToken);
PluginLog.Debug("Waiting for uploads to finish");
}
CurrentUploads.Clear();
IsUploading = false;
if (!uploadToken.IsCancellationRequested)
{
PluginLog.Debug("=== Pushing character data ===");
await _userHub!.InvokeAsync("PushCharacterData", character, visibleCharacterIds, uploadToken);
}
else
{
PluginLog.Warning("=== Upload operation was cancelled ===");
}
PluginLog.Debug("== Upload complete for " + character.JobId);
uploadCancellationTokenSource = null;
}
public Task ReceiveCharacterData(CharacterCacheDto character, string characterHash)
{
PluginLog.Debug("Received DTO for " + characterHash);
CharacterReceived?.Invoke(null, new CharacterReceivedEventArgs(characterHash, character));
return Task.CompletedTask;
}
public async Task UpdateCurrentDownloadSize(string hash)
{
long fileSize = await _fileHub!.InvokeAsync<long>("GetFileSize", hash);
}
public async Task DownloadFiles(List<FileReplacementDto> fileReplacementDto, string cacheFolder)
{
foreach (var file in fileReplacementDto)
{
var fileSize = await _fileHub!.InvokeAsync<long>("GetFileSize", file.Hash);
CurrentDownloads[file.Hash] = (0, fileSize);
}
foreach (var file in fileReplacementDto.Where(f => CurrentDownloads[f.Hash].Item2 > 0))
{
var hash = file.Hash;
var data = await DownloadFile(hash);
var extractedFile = LZ4.LZ4Codec.Unwrap(data);
var ext = file.GamePaths.First().Split(".", StringSplitOptions.None).Last();
var filePath = Path.Combine(cacheFolder, file.Hash + "." + ext);
await File.WriteAllBytesAsync(filePath, extractedFile);
await using (var db = new FileCacheContext())
{
db.Add(new FileCache
{
Filepath = filePath.ToLower(),
Hash = file.Hash,
LastModifiedDate = DateTime.Now.Ticks.ToString(),
});
await db.SaveChangesAsync();
}
PluginLog.Debug("File downloaded to " + filePath);
}
CurrentDownloads.Clear();
}
public async Task<byte[]> DownloadFile(string hash)
{
IsDownloading = true;
var reader = await _fileHub!.StreamAsChannelAsync<byte[]>("DownloadFile", hash);
List<byte> downloadedData = new();
while (await reader.WaitToReadAsync())
{
while (reader.TryRead(out var data))
{
CurrentDownloads[hash] = (CurrentDownloads[hash].Item1 + data.Length, CurrentDownloads[hash].Item2);
downloadedData.AddRange(data);
//await Task.Delay(25);
}
}
IsDownloading = false;
return downloadedData.ToArray();
}
public async Task GetCharacterData(Dictionary<string, int> hashedCharacterNames)
{
await _userHub!.InvokeAsync("GetCharacterData",
hashedCharacterNames);
}
public async Task SendPairedClientPauseChange(string uid, bool paused)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientPauseChange", uid, paused);
}
public async Task SendPairedClientAddition(string uid)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientAddition", uid);
}
public async Task SendPairedClientRemoval(string uid)
{
if (!IsConnected || SecretKey == "-") return;
await _userHub!.SendAsync("SendPairedClientRemoval", uid);
}
public void Dispose()
{
cts?.Cancel();
_ = DisposeHubConnections();
}
public async Task<List<string>> SendCharacterName(string hashedName)
{
return await _userHub!.InvokeAsync<List<string>>("SendCharacterNameHash", hashedName);
}
public CharacterCacheDto CharacterData { get; set; }
public string CharacterNameHash { get; set; }
}
}