Winify – Diff between revs 59 and 61

Subversion Repositories:

?pathlinks?

Rev 59 – Rev 62 – Go to most recent revision – Details – Blame – Last modification – View Log

 ï»¿using System;
+using System.Collections.Generic;
+using System.Diagnostics;
 using System.Drawing;
+using System.Globalization;
 using System.IO;
+using System.Linq;
 using System.Net;
 using System.Net.Http;
 using System.Net.Http.Headers;
+using System.Runtime.Caching;
+using System.Runtime.CompilerServices;
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+using System.Windows.Forms;
 using Newtonsoft.Json;
 using Serilog;
 using Servers;
 using WebSocketSharp;
 using WebSocketSharp.Net;
+using Winify.Utilities;
 using ErrorEventArgs = WebSocketSharp.ErrorEventArgs;
 using NetworkCredential = System.Net.NetworkCredential;
 namespace Winify.Gotify
         private readonly Uri _httpUri;
         private WebSocket _webSocketSharp;
         private readonly Configuration.Configuration _configuration;
+        private Task _initTask;
+        private IDisposable _tplRetrievePastMessagesLink;
+        private IDisposable _tplWebSocketsBufferBlockTransformLink;
+        private IDisposable _tplWebSocketsTransformActionLink;
+        private IDisposable _tplWebSocketsTransformActionNullLink;
+        private readonly BufferBlock<byte[]> _webSocketMessageBufferBlock;
+        private readonly Stopwatch _webSocketsClientPingStopWatch;
+        private readonly ScheduledContinuation _webSocketsServerResponseScheduledContinuation;
         private Task _initTask;
         private readonly MemoryCache _applicationImageCache;
         #endregion
+        #region Constructors, Destructors and Finalizers
+        private GotifyConnection()
+        {
+            _applicationImageCache = new MemoryCache("GotifyApplicationImageCache");
+            _webSocketsServerResponseScheduledContinuation = new ScheduledContinuation();
+            _webSocketsClientPingStopWatch = new Stopwatch();
+            _webSocketMessageBufferBlock = new BufferBlock<byte[]>(new DataflowBlockOptions { CancellationToken = _cancellationToken });
+            var webSocketTransformBlock = new TransformBlock<byte[], GotifyMessage>(bytes =>
+            {
+                if (bytes.Length == 0)
+                {
+                    return null;
+                }
+                var message = Encoding.UTF8.GetString(bytes, 0, bytes.Length);
+                GotifyMessage gotifyNotification;
+                try
+                {
+                    gotifyNotification = JsonConvert.DeserializeObject<GotifyMessage>(message);
+                }
+                catch (JsonSerializationException exception)
+                {
+                    Log.Warning($"Could not deserialize notification: {exception.Message}");
+                    return null;
+                }
+                if (gotifyNotification == null)
+                {
+                    Log.Warning($"Could not deserialize gotify notification: {message}");
+                    return null;
+                }
+                return gotifyNotification;
+            }, new ExecutionDataflowBlockOptions { CancellationToken = _cancellationToken });
+            var webSocketActionBlock = new ActionBlock<GotifyMessage>(async message =>
+            {
+                message.Server = _server;
+                var cachedImage = _applicationImageCache.Get($"{message.AppId}");
+                if (cachedImage is Image applicationImage)
+                {
+                    GotifyNotification?.Invoke(this,
+                        new GotifyNotificationEventArgs(message, applicationImage));
+                    return;
+                }
+                using (var imageStream = await RetrieveGotifyApplicationImage(message.AppId, _cancellationToken))
+                {
+                    if (imageStream == null || imageStream.Length == 0)
+                    {
+                        Log.Warning("Could not find any application image for notification");
+                        return;
+                    }
+                    var image = Image.FromStream(imageStream);
+                    _applicationImageCache.Add($"{message.AppId}", image.Clone(),
+                        new CacheItemPolicy
+                        {
+                            SlidingExpiration = TimeSpan.FromHours(1)
+                        });
+                    GotifyNotification?.Invoke(this,
+                        new GotifyNotificationEventArgs(message, image));
+                }
+                Log.Debug($"Notification message received: {message.Message}");
+            }, new ExecutionDataflowBlockOptions { CancellationToken = _cancellationToken });
+            _tplWebSocketsBufferBlockTransformLink = _webSocketMessageBufferBlock.LinkTo(webSocketTransformBlock,
+                new DataflowLinkOptions { PropagateCompletion = true });
-        #region Constructors, Destructors and Finalizers
+            _tplWebSocketsTransformActionLink = webSocketTransformBlock.LinkTo(webSocketActionBlock,
                 new DataflowLinkOptions { PropagateCompletion = true }, message => message != null);
-        private GotifyConnection()
+            _tplWebSocketsTransformActionNullLink = webSocketTransformBlock.LinkTo(DataflowBlock.NullTarget<GotifyMessage>(),
                 new DataflowLinkOptions() { PropagateCompletion = true });
+            {
                 _cancellationTokenSource.Dispose();
                 _cancellationTokenSource = null;
+            }
+            if (_tplWebSocketsBufferBlockTransformLink != null)
+            {
+                _tplWebSocketsBufferBlockTransformLink.Dispose();
+                _tplWebSocketsBufferBlockTransformLink = null;
+            }
+            if (_tplWebSocketsTransformActionLink != null)
+            {
+                _tplWebSocketsTransformActionLink.Dispose();
+                _tplWebSocketsTransformActionLink = null;
+            }
+            if (_tplWebSocketsTransformActionNullLink != null)
+            {
+                _tplWebSocketsTransformActionNullLink.Dispose();
+                _tplWebSocketsTransformActionNullLink = null;
+            }
+            if (_tplRetrievePastMessagesLink != null)
+            {
+                _tplRetrievePastMessagesLink.Dispose();
+                _tplRetrievePastMessagesLink = null;
+            }
             if (_webSocketSharp != null)
+            {
                 _webSocketSharp.Close();
                 _webSocketSharp = null;
             if (_configuration.RetrievePastNotificationHours != 0)
+            {
                 _initTask = RetrievePastMessages(_cancellationToken);
+            }
-            _runTask = Run(_cancellationToken);
+            _runTask = HeartBeat(_cancellationToken);
+        }
         private void Connect()
+        {
+            _webSocketSharp = new WebSocket(_webSocketsUri.AbsoluteUri);
             _webSocketSharp.EmitOnPing = true;
-            _webSocketSharp = new WebSocket(_webSocketsUri.AbsoluteUri);
+            _webSocketSharp.WaitTime = TimeSpan.FromMinutes(1);
             _webSocketSharp.SslConfiguration = new ClientSslConfiguration(_webSocketsUri.Host,
                 new X509CertificateCollection(new X509Certificate[] { }), SslProtocols.Tls12, false);
             if (_configuration.Proxy.Enable)
+        }
         private void WebSocketSharp_OnOpen(object sender, EventArgs e)
+        {
+            Log.Information($"WebSockets connection to server {_webSocketsUri.AbsoluteUri} is now open");
+            _webSocketsServerResponseScheduledContinuation.Schedule(TimeSpan.FromMinutes(1), OnUnresponsiveServer, _cancellationToken);
+        }
+        private void OnUnresponsiveServer()
+        {
-            Log.Information($"WebSockets connection to server {_webSocketsUri.AbsoluteUri} is now open");
+            Log.Warning($"Server {_server} has not responded in a long while...");
+        }
         private async void WebSocketSharp_OnError(object sender, ErrorEventArgs e)
                 Stop();
                 return;
+            }
+            await Task.Delay(TimeSpan.FromSeconds(1), _cancellationToken);
             await Task.Delay(TimeSpan.FromSeconds(1), _cancellationToken);
             Log.Information($"Reconnecting to websocket server {_webSocketsUri.AbsoluteUri}");
             Connect();
+        }
         private async void WebSocketSharp_OnMessage(object sender, MessageEventArgs e)
         {
-            if (e.RawData.Length == 0)
-            {
-                Log.Warning("Empty message received from server");
-                return;
-            }
-            var message = Encoding.UTF8.GetString(e.RawData, 0, e.RawData.Length);
-            GotifyMessage gotifyNotification;
+            if (e.IsPing)
-            try
-            {
-                gotifyNotification = JsonConvert.DeserializeObject<GotifyMessage>(message);
-            }
-            catch (JsonSerializationException exception)
-            {
-                Log.Warning($"Could not deserialize notification: {exception.Message}");
-                return;
-            }
+            {
-            if (gotifyNotification == null)
+                Log.Information($"Server {_server} sent PING message");
-            {
-                Log.Warning($"Could not deserialize gotify notification: {message}");
-                return;
-            }
-            gotifyNotification.Server = _server;
-            var applicationUriBuilder = new UriBuilder(_httpUri);
-            try
-            {
-                applicationUriBuilder.Path = Path.Combine(applicationUriBuilder.Path, "application");
-            }
-            catch (ArgumentException exception)
-            {
-                Log.Warning("Could not build an URI to an application");
-                return;
-            }
-            using (var imageStream =
-                   await RetrieveGotifyApplicationImage(gotifyNotification.AppId, applicationUriBuilder.Uri,
-                       _cancellationToken))
-            {
-                if (imageStream == null)
-                {
-                    Log.Warning("Could not find any application image for notification");
-                    return;
-                }
-                var image = Image.FromStream(imageStream);
                 _webSocketsServerResponseScheduledContinuation.Schedule(TimeSpan.FromMinutes(1), OnUnresponsiveServer, _cancellationToken);
                 GotifyNotification?.Invoke(this,
-                    new GotifyNotificationEventArgs(gotifyNotification, image));
+                return;
+            }
         #region Private Methods
         private async Task RetrievePastMessages(CancellationToken cancellationToken)
+        {
+            var messageUriBuilder = new UriBuilder(_httpUri);
-            var messageUriBuilder = new UriBuilder(_httpUri);
+            var gotifyApplicationBufferBlock = new BufferBlock<GotifyApplication>(new DataflowBlockOptions { CancellationToken = cancellationToken });
-            foreach (var application in await RetrieveGotifyApplications(cancellationToken))
+            var gotifyApplicationActionBlock = new ActionBlock<GotifyApplication>(async application =>
+            {
                 try
+                {
                     messageUriBuilder.Path = Path.Combine(messageUriBuilder.Path, "application", $"{application.Id}",
                         "message");
+                }
                 catch (ArgumentException exception)
+                {
                     Log.Error($"No application URL could be built for {_server.Url} due to {exception.Message}");
+                    return;
+                }
-                    continue;
+                HttpResponseMessage messagesResponse;
+                try
+                {
+                    messagesResponse = await _httpClient.GetAsync(messageUriBuilder.Uri, cancellationToken);
+                }
+                catch (Exception exception)
+                {
+                    Log.Error($"Could not get application {application.Id} due to {exception.Message}");
                     return;
+                }
                 catch (JsonSerializationException exception)
+                {
                     Log.Warning($"Could not deserialize the message response: {exception.Message}");
-                    continue;
+                    return;
                 }
                 var applicationUriBuilder = new UriBuilder(_httpUri);
-                try
-                {
-                    applicationUriBuilder.Path = Path.Combine(applicationUriBuilder.Path, "application");
+                foreach (var message in gotifyMessageQuery.Messages.Where(message => message.Date >= DateTime.Now - TimeSpan.FromHours(_configuration.RetrievePastNotificationHours)))
-                }
-                catch (ArgumentException exception)
+                {
+                    message.Server = _server;
+                    var cachedImage = _applicationImageCache.Get($"{message.AppId}");
+                    if (cachedImage is Image applicationImage)
-                {
+                    {
-                    Log.Warning($"Could not build an URI to an application: {exception}");
+                        GotifyNotification?.Invoke(this,
+                            new GotifyNotificationEventArgs(message, applicationImage));
+                        return;
-                    return;
+                    }
-                }
                     using var imageStream = await RetrieveGotifyApplicationImage(message.AppId, _cancellationToken);
+                    if (imageStream == null || imageStream.Length == 0)
-                foreach (var message in gotifyMessageQuery.Messages)
+                    {
-                {
-                    if (message.Date < DateTime.Now - TimeSpan.FromHours(_configuration.RetrievePastNotificationHours))
+                        Log.Warning("Could not find any application image for notification");
-                        continue;
-                    message.Server = _server;
+                        continue;
+                    }
-                    using (var imageStream =
-                           await RetrieveGotifyApplicationImage(message.AppId, applicationUriBuilder.Uri,
-                               _cancellationToken))
+                    var image = Image.FromStream(imageStream);
-                    {
-                        if (imageStream == null)
-                        {
-                            Log.Warning("Could not find any application image for notification");
+                    _applicationImageCache.Add($"{message.AppId}", image.Clone(),
-                            return;
                         new CacheItemPolicy
+                        {
+                            SlidingExpiration = TimeSpan.FromHours(1)
+                        });
+                    GotifyNotification?.Invoke(this,
+                        new GotifyNotificationEventArgs(message, image));
+                }
+            }, new ExecutionDataflowBlockOptions { CancellationToken = cancellationToken });
+            gotifyApplicationBufferBlock.LinkTo(gotifyApplicationActionBlock,
+                new DataflowLinkOptions { PropagateCompletion = true });
+            await foreach (var application in RetrieveGotifyApplications(cancellationToken))
                         var image = Image.FromStream(imageStream);
+                await gotifyApplicationBufferBlock.SendAsync(application, cancellationToken);
                         GotifyNotification?.Invoke(this,
                             new GotifyNotificationEventArgs(message, image));
             gotifyApplicationBufferBlock.Complete();
             await gotifyApplicationActionBlock.Completion;
         }
+        private async Task HeartBeat(CancellationToken cancellationToken)
+        {
+            try
+            {
+                do
+                {
+                    await Task.Delay(TimeSpan.FromMinutes(1), cancellationToken);
+                    _webSocketsClientPingStopWatch.Restart();
+                    if (!_webSocketSharp.Ping())
+                    {
+                        Log.Warning($"Server {_server} did not respond to PING message.");
                         continue;
         private async Task Run(CancellationToken cancellationToken)
-            try
+                    var delta = _webSocketsClientPingStopWatch.ElapsedMilliseconds;
-                do
+                    Log.Information($"PING response latency for {_server} is {delta}ms");
-                    await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
+                    _webSocketsServerResponseScheduledContinuation.Schedule(TimeSpan.FromMinutes(1), OnUnresponsiveServer, _cancellationToken);
                 } while (!cancellationToken.IsCancellationRequested);
+            }
             catch (Exception exception) when (exception is OperationCanceledException ||
                                               exception is ObjectDisposedException)
+            {
+            }
             catch (Exception exception)
+            {
-                Log.Warning(exception, "Failure running connection loop");
+                Log.Warning(exception, $"Heartbeat for server {_server} has failed due to {exception.Message}");
+            }
+        }
-        private async Task<GotifyApplication[]> RetrieveGotifyApplications(CancellationToken cancellationToken)
+        private async IAsyncEnumerable<GotifyApplication> RetrieveGotifyApplications([EnumeratorCancellation] CancellationToken cancellationToken)
+        {
+            var applicationsUriBuilder = new UriBuilder(_httpUri);
             try
-            var applicationsUriBuilder = new UriBuilder(_httpUri);
+            {
-            try
-            {
-                applicationsUriBuilder.Path = Path.Combine(applicationsUriBuilder.Path, "application");
                 applicationsUriBuilder.Path = Path.Combine(applicationsUriBuilder.Path, "application");
             catch (ArgumentException exception)
-            {
-                Log.Error($"No application URL could be built for {_server.Url} due to {exception}");
+            catch (ArgumentException exception)
+            {
+                Log.Error($"No application URL could be built for {_server.Url} due to {exception}");
             var applicationsResponse = await _httpClient.GetAsync(applicationsUriBuilder.Uri, cancellationToken);
+                yield break;
-            var applications = await applicationsResponse.Content.ReadAsStringAsync();
+            }
-            GotifyApplication[] gotifyApplications;
-            try
-            {
-                gotifyApplications =
-                    JsonConvert.DeserializeObject<GotifyApplication[]>(applications);
-            }
-            catch (JsonSerializationException exception)
-            {
-                Log.Warning($"Could not deserialize the list of applications from the server: {exception}");
+            HttpResponseMessage applicationsResponse;
-                return null;
+            try
+            {
                 applicationsResponse = await _httpClient.GetAsync(applicationsUriBuilder.Uri, cancellationToken);
             return gotifyApplications;
             catch (Exception exception)
+            {
-        private async Task<Stream> RetrieveGotifyApplicationImage(int appId, Uri applicationUri,
+                Log.Error($"Could not retrieve applications: {exception.Message}");
-            CancellationToken cancellationToken)
-        {
+                yield break;
             var applicationResponse = await _httpClient.GetAsync(applicationUri, cancellationToken);
-            var applications = await applicationResponse.Content.ReadAsStringAsync();
+            var applications = await applicationsResponse.Content.ReadAsStringAsync();
+            GotifyApplication[] gotifyApplications;
+            try
+            {
+                gotifyApplications =
+                    JsonConvert.DeserializeObject<GotifyApplication[]>(applications);
+            }
+            catch (JsonSerializationException exception)
+            {
-            GotifyApplication[] gotifyApplications;
+                Log.Warning($"Could not deserialize the list of applications from the server: {exception}");
-            try
-                gotifyApplications =
+                yield break;
                     JsonConvert.DeserializeObject<GotifyApplication[]>(applications);
+            foreach (var application in gotifyApplications)
+            {
+                yield return application;
+            }
-            catch (JsonSerializationException exception)
+        }
+        private async Task<Stream> RetrieveGotifyApplicationImage(int appId, CancellationToken cancellationToken)
+        {
+            await foreach (var application in RetrieveGotifyApplications(cancellationToken))
+            {
+                if (application.Id != appId) continue;
                 if (!Uri.TryCreate(Path.Combine($"{_httpUri}", $"{application.Image}"), UriKind.Absolute, out var applicationImageUri))
                 Log.Warning($"Could not deserialize the list of applications from the server: {exception.Message}");
+                    Log.Warning("Could not build URL path to application icon");
+                    continue;
+                }
                 return null;
-            }
+                HttpResponseMessage imageResponse;
-            foreach (var application in gotifyApplications)
+                try
+                {
-                if (application.Id != appId) continue;
+                    imageResponse = await _httpClient.GetAsync(applicationImageUri, cancellationToken);
+                }