Switch Authentication to asynchronous streaming calls (#16)

* add base grpc service and swap auth service to streaming * remove Authorize from hub itself * remove unused usings * heave files server to net 7, add exception handling in grpc auth stream Co-authored-by: rootdarkarchon <root.darkarchon@outlook.com>
2022-10-13 16:55:23 +02:00
parent d37c1208fe
commit c98e2b2dd6
20 changed files with 313 additions and 159 deletions
--- a/MareSynchronosServer/MareSynchronosShared/Services/GrpcAuthenticationService.cs
+++ b/MareSynchronosServer/MareSynchronosShared/Services/GrpcAuthenticationService.cs
@@ -0,0 +1,109 @@
+using System.Collections.Concurrent;
+using System.Security.Cryptography;
+using MareSynchronosShared.Protos;
+using Microsoft.Extensions.Logging;
+
+namespace MareSynchronosShared.Services;
+
+public class GrpcAuthenticationService : GrpcBaseService
+{
+    private record AuthRequestInternal
+    {
+        public AuthRequest Request { get; set; }
+        public long Id { get; set; }
+    }
+
+    private readonly AuthService.AuthServiceClient _authClient;
+    private readonly ConcurrentQueue<AuthRequestInternal> _requestQueue = new();
+    private readonly ConcurrentDictionary<long, AuthReply> _authReplies = new();
+    private long _requestId = 0;
+
+    public GrpcAuthenticationService(ILogger<GrpcAuthenticationService> logger, AuthService.AuthServiceClient authClient) : base(logger)
+    {
+        _authClient = authClient;
+    }
+
+    public async Task<AuthReply> AuthorizeAsync(string ip, string secretKey)
+    {
+        using var sha1 = SHA1.Create();
+        var id = Interlocked.Increment(ref _requestId);
+        _requestQueue.Enqueue(new AuthRequestInternal()
+        {
+            Id = id,
+            Request = new AuthRequest()
+            {
+                Ip = ip,
+                SecretKey = secretKey,
+            }
+        });
+
+        using CancellationTokenSource cts = new(TimeSpan.FromSeconds(30));
+        AuthReply response = null;
+
+        while (!GrpcIsFaulty && !cts.IsCancellationRequested && !_authReplies.TryRemove(id, out response))
+        {
+            await Task.Delay(10, cts.Token).ConfigureAwait(false);
+        }
+
+        return response ?? new AuthReply
+        {
+            Success = false,
+        };
+    }
+
+    public async Task GrpcAuthStream(CancellationToken token)
+    {
+        try
+        {
+            using var stream = _authClient.Authorize(cancellationToken: token);
+            while (!token.IsCancellationRequested)
+            {
+                while (_requestQueue.TryDequeue(out var request))
+                {
+                    await stream.RequestStream.WriteAsync(request.Request, token).ConfigureAwait(false);
+                    await stream.ResponseStream.MoveNext(token).ConfigureAwait(false);
+                    _authReplies[request.Id] = stream.ResponseStream.Current;
+                }
+
+                await Task.Delay(10, token).ConfigureAwait(false);
+            }
+        }
+        catch
+        {
+            SetGrpcFaulty();
+        }
+    }
+
+    protected override Task OnGrpcRestore()
+    {
+        return Task.CompletedTask;
+    }
+
+    protected override Task PostStartStream()
+    {
+        return Task.CompletedTask;
+    }
+
+    protected override Task PreStartStream()
+    {
+        _requestQueue.Clear();
+        _authReplies.Clear();
+        return Task.CompletedTask;
+    }
+
+    protected override Task StartAsyncInternal(CancellationToken cancellationToken)
+    {
+        return Task.CompletedTask;
+    }
+
+    protected override Task StartStream(CancellationToken ct)
+    {
+        _ = GrpcAuthStream(ct);
+        return Task.CompletedTask;
+    }
+
+    protected override Task StopAsyncInternal(CancellationToken cancellationToken)
+    {
+        return Task.CompletedTask;
+    }
+}
--- a/MareSynchronosServer/MareSynchronosShared/Services/GrpcBaseService.cs
+++ b/MareSynchronosServer/MareSynchronosShared/Services/GrpcBaseService.cs
@@ -0,0 +1,129 @@
+using Grpc.Core;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+
+namespace MareSynchronosShared.Services;
+
+public abstract class GrpcBaseService : IHostedService, IDisposable
+{
+    protected GrpcBaseService(ILogger logger)
+    {
+        _logger = logger;
+    }
+
+    private CancellationTokenSource _faultCheckCts = new();
+    private CancellationTokenSource _streamCts = new();
+    private readonly ILogger _logger;
+    protected bool GrpcIsFaulty { get; private set; }
+
+    protected abstract Task StartAsyncInternal(CancellationToken cancellationToken);
+    protected abstract Task StopAsyncInternal(CancellationToken cancellationToken);
+    protected abstract Task OnGrpcRestore();
+    protected abstract Task PreStartStream();
+    protected abstract Task StartStream(CancellationToken ct);
+    protected abstract Task PostStartStream();
+
+    public async Task StartAsync(CancellationToken cancellationToken)
+    {
+        _ = RestartStreams();
+        _ = CheckGrpcFaults(_faultCheckCts.Token);
+        await StartAsyncInternal(cancellationToken).ConfigureAwait(false);
+    }
+
+    public async Task StopAsync(CancellationToken cancellationToken)
+    {
+        _faultCheckCts.Cancel();
+        _streamCts.Cancel();
+        await StopAsyncInternal(cancellationToken).ConfigureAwait(false);
+    }
+
+    private async Task RestartStreams()
+    {
+        _streamCts?.Cancel();
+        _streamCts?.Dispose();
+        _streamCts = new();
+        if (!GrpcIsFaulty)
+        {
+            try
+            {
+                await PreStartStream().ConfigureAwait(false);
+                await StartStream(_streamCts.Token).ConfigureAwait(false);
+                await PostStartStream().ConfigureAwait(false);
+            }
+            catch
+            {
+                SetGrpcFaulty();
+            }
+        }
+    }
+
+    protected void SetGrpcFaulty()
+    {
+        if (!GrpcIsFaulty)
+        {
+            GrpcIsFaulty = true;
+            _logger.LogWarning("GRPC connection is faulty");
+        }
+    }
+
+    private async Task CheckGrpcFaults(CancellationToken ct)
+    {
+        while (!ct.IsCancellationRequested)
+        {
+            try
+            {
+                await CheckFaultStateAndResend().ConfigureAwait(false);
+            }
+            catch { SetGrpcFaulty(); }
+            await Task.Delay(250).ConfigureAwait(false);
+        }
+    }
+
+    private async Task CheckFaultStateAndResend()
+    {
+        if (GrpcIsFaulty)
+        {
+            await RestartStreams().ConfigureAwait(false);
+            await OnGrpcRestore().ConfigureAwait(false);
+            _logger.LogInformation("GRPC connection is restored");
+            GrpcIsFaulty = false;
+        }
+    }
+
+    protected async Task<T> InvokeOnGrpc<T>(AsyncUnaryCall<T> toExecute)
+    {
+        try
+        {
+            var result = await toExecute.ConfigureAwait(false);
+
+            await CheckFaultStateAndResend().ConfigureAwait(false);
+
+            return result;
+        }
+        catch
+        {
+            SetGrpcFaulty();
+
+            return default;
+        }
+    }
+
+    protected async Task ExecuteOnGrpc<T>(AsyncUnaryCall<T> toExecute)
+    {
+        try
+        {
+            await toExecute.ConfigureAwait(false);
+            await CheckFaultStateAndResend().ConfigureAwait(false);
+        }
+        catch
+        {
+            SetGrpcFaulty();
+        }
+    }
+
+    public void Dispose()
+    {
+        _streamCts?.Dispose();
+        _faultCheckCts?.Dispose();
+    }
+}