Merge pull request #595 from theolivenbaum/switch-to-new-socket-lib

Switch to async socket lib

Author: GregorBiswanger

ElectronNET.API/App.cs

@@ -1,44 +1,318 @@
-using Quobject.SocketIoClientDotNet.Client;
-using System;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Newtonsoft.Json;
+using Newtonsoft.Json.Serialization;
+using SocketIOClient;
+using SocketIOClient.JsonSerializer;
+using SocketIOClient.Newtonsoft.Json;
 
 namespace ElectronNET.API
 {
     internal static class BridgeConnector
     {
-        private static Socket _socket;
+        internal static class EventTasks<T>
+        {
+            //Although SocketIO already manage event handlers, we need to manage this here as well for the OnResult calls,
+            //because SocketIO will simply replace the existing event handler on every call to On(key, ...) , which means there is 
+            //a race condition between On / Off calls that can lead to tasks deadlocking forever without ever triggering their On handler
+
+            private static readonly Dictionary<string, TaskCompletionSource<T>> _taskCompletionSources = new();
+            private static readonly Dictionary<string, string> _eventKeys = new();
+            private static readonly object _lock = new();
+
+            /// <summary>
+            /// Get or add a new TaskCompletionSource<typeparamref name="T"/> for a given event key
+            /// </summary>
+            /// <param name="key"></param>
+            /// <param name="eventKey"></param>
+            /// <param name="taskCompletionSource"></param>
+            /// <param name="waitThisFirstAndThenTryAgain"></param>
+            /// <returns>Returns true if a new TaskCompletionSource<typeparamref name="T"/> was added to the dictionary</returns>
+            internal static bool TryGetOrAdd(string key, string eventKey, out TaskCompletionSource<T> taskCompletionSource, out Task waitThisFirstAndThenTryAgain)
+            {
+                lock (_lock)
+                {
+                    if (!_taskCompletionSources.TryGetValue(key, out taskCompletionSource))
+                    {
+                        taskCompletionSource = new(TaskCreationOptions.RunContinuationsAsynchronously);
+                        _taskCompletionSources[key] = taskCompletionSource;
+                        _eventKeys[key] = eventKey;
+                        waitThisFirstAndThenTryAgain = null;
+                        return true; //Was added, so we need to also register the socket events
+                    }
+
+                    if(_eventKeys.TryGetValue(key, out var existingEventKey) && existingEventKey == eventKey)
+                    {
+                        waitThisFirstAndThenTryAgain = null;
+                        return false; //No need to register the socket events twice
+                    }
+
+                    waitThisFirstAndThenTryAgain = taskCompletionSource.Task; //Will need to try again after the previous existing one is done
+
+                    taskCompletionSource = null;
+
+                    return true; //Need to register the socket events, but must first await the previous task to complete
+                }
+            }
+
+            /// <summary>
+            /// Clean up the TaskCompletionSource<typeparamref name="T"/> from the dictionary if and only if it is the same as the passed argument
+            /// </summary>
+            /// <param name="key"></param>
+            /// <param name="eventKey"></param>
+            /// <param name="taskCompletionSource"></param>
+            internal static void DoneWith(string key, string eventKey, TaskCompletionSource<T> taskCompletionSource)
+            {
+                lock (_lock)
+                {
+                    if (_taskCompletionSources.TryGetValue(key, out var existingTaskCompletionSource)
+                        && ReferenceEquals(existingTaskCompletionSource, taskCompletionSource))
+                    {
+                        _taskCompletionSources.Remove(key);
+                    }
+
+                    if (_eventKeys.TryGetValue(key, out var existingEventKey) && existingEventKey == eventKey)
+                    {
+                        _eventKeys.Remove(key);
+                    }
+                }
+            }
+        }
+
+        private static SocketIO _socket;
+
         private static object _syncRoot = new object();
 
-        public static Socket Socket
+        public static void Emit(string eventString, params object[] args)
         {
-            get
+            //We don't care about waiting for the event to be emitted, so this doesn't need to be async 
+
+            Task.Run(async () =>
+            {
+                if (App.SocketDebug)
+                {
+                    Console.WriteLine($"Sending event {eventString}");
+                }
+
+                await Socket.EmitAsync(eventString, args);
+
+                if (App.SocketDebug)
+                {
+                    Console.WriteLine($"Sent event {eventString}");
+                }
+            });
+        }
+
+        /// <summary>
+        /// This method is only used on places where we need to be sure the event was sent on the socket, such as Quit, Exit, Relaunch and QuitAndInstall methods
+        /// </summary>
+        /// <param name="eventString"></param>
+        /// <param name="args"></param>
+        internal static void EmitSync(string eventString, params object[] args)
+        {
+            if (App.SocketDebug)
+            {
+                Console.WriteLine($"Sending event {eventString}");
+            }
+
+            Socket.EmitAsync(eventString, args).Wait();
+
+            if (App.SocketDebug)
+            {
+                Console.WriteLine($"Sent event {eventString}");
+            }
+        }
+
+        public static void Off(string eventString)
+        {
+            Socket.Off(eventString);
+        }
+
+        public static void On(string eventString, Action fn)
+        {
+            Socket.On(eventString, _ => fn());
+        }
+
+        public static void On<T>(string eventString, Action<T> fn)
+        {
+            Socket.On(eventString, (o) => fn(o.GetValue<T>(0)));
+        }
+
+        public static void Once<T>(string eventString, Action<T> fn)
+        {
+            On<T>(eventString, (o) =>
+            {
+                Off(eventString);
+                fn(o);
+            });
+        }
+
+        public static async Task<T> OnResult<T>(string triggerEvent, string completedEvent, params object[] args)
+        {
+            string eventKey = completedEvent;
+
+            if (args is object && args.Length > 0) // If there are arguments passed, we generate a unique event key with the arguments
+                                                   // this allow us to wait for previous events first before registering new ones
+            {
+                var hash = new HashCode();
+                foreach(var obj in args)
+                {
+                    hash.Add(obj);
+                }
+                eventKey = $"{eventKey}-{(uint)hash.ToHashCode()}";
+            }
+
+            if (EventTasks<T>.TryGetOrAdd(completedEvent, eventKey, out var taskCompletionSource, out var waitThisFirstAndThenTryAgain))
+            {
+                if (waitThisFirstAndThenTryAgain is object)
+                {
+                    //There was a pending call with different parameters, so we need to wait that first and then call here again
+                    try
+                    {
+                        await waitThisFirstAndThenTryAgain;
+                    }
+                    catch
+                    {
+                        //Ignore any exceptions here so we can set a new event below
+                        //The exception will also be visible to the original first caller due to taskCompletionSource.Task
+                    }
+
+                    //Try again to set the event
+                    return await OnResult<T>(triggerEvent, completedEvent, args);
+                }
+                else
+                {
+                    //A new TaskCompletionSource was added, so we need to register the completed event here
+
+                    On<T>(completedEvent, (result) =>
+                    {
+                        Off(completedEvent);
+                        taskCompletionSource.SetResult(result);
+                        EventTasks<T>.DoneWith(completedEvent, eventKey, taskCompletionSource);
+                    });
+
+                    Emit(triggerEvent, args);
+                }
+            }
+
+            return await taskCompletionSource.Task;
+        }
+
+
+        public static async Task<T> OnResult<T>(string triggerEvent, string completedEvent, CancellationToken cancellationToken, params object[] args)
+        {
+            string eventKey = completedEvent;
+
+            if (args is object && args.Length > 0) // If there are arguments passed, we generate a unique event key with the arguments
+                                                   // this allow us to wait for previous events first before registering new ones
             {
-                if(_socket == null && HybridSupport.IsElectronActive)
+                var hash = new HashCode();
+                foreach (var obj in args)
+                {
+                    hash.Add(obj);
+                }
+                eventKey = $"{eventKey}-{(uint)hash.ToHashCode()}";
+            }
+
+            if (EventTasks<T>.TryGetOrAdd(completedEvent, eventKey, out var taskCompletionSource, out var waitThisFirstAndThenTryAgain))
+            {
+                if (waitThisFirstAndThenTryAgain is object)
+                {
+                    //There was a pending call with different parameters, so we need to wait that first and then call here again
+                    try
+                    {
+                        await Task.Run(() => waitThisFirstAndThenTryAgain, cancellationToken);
+                    }
+                    catch
+                    {
+                        //Ignore any exceptions here so we can set a new event below
+                        //The exception will also be visible to the original first caller due to taskCompletionSource.Task
+                    }
+
+                    //Try again to set the event
+                    return await OnResult<T>(triggerEvent, completedEvent, cancellationToken, args);
+                }
+                else
                 {
-                    lock (_syncRoot)
+                    using (cancellationToken.Register(() => taskCompletionSource.TrySetCanceled()))
                     {
-                        if (_socket == null && HybridSupport.IsElectronActive)
+                        //A new TaskCompletionSource was added, so we need to register the completed event here
+
+                        On<T>(completedEvent, (result) =>
                         {
-                            _socket = IO.Socket("http://localhost:" + BridgeSettings.SocketPort);
-                            _socket.On(Socket.EVENT_CONNECT, () =>
-                            {
-                                Console.WriteLine("BridgeConnector connected!");
-                            });
-                        }
+                            Off(completedEvent);
+                            taskCompletionSource.SetResult(result);
+                            EventTasks<T>.DoneWith(completedEvent, eventKey, taskCompletionSource);
+                        });
+
+                        Emit(triggerEvent, args);
                     }
                 }
-                else if(_socket == null && !HybridSupport.IsElectronActive)
+            }
+
+            return await taskCompletionSource.Task;
+        }
+        private static SocketIO Socket
+        {
+            get
+            {
+                if (_socket is null)
                 {
-                    lock (_syncRoot)
+                    if (HybridSupport.IsElectronActive)
                     {
-                        if (_socket == null && !HybridSupport.IsElectronActive)
+
+                        lock (_syncRoot)
                         {
-                            _socket = IO.Socket(new Uri("http://localhost"), new IO.Options { AutoConnect = false });
+                            if (_socket is null && HybridSupport.IsElectronActive)
+                            {
+                                var socket = new SocketIO($"http://localhost:{BridgeSettings.SocketPort}", new SocketIOOptions()
+                                {
+                                    EIO = 3
+                                });
+
+                                socket.JsonSerializer = new CamelCaseNewtonsoftJsonSerializer(socket.Options.EIO);
+
+
+                                socket.OnConnected += (_, __) =>
+                                {
+                                    Console.WriteLine("BridgeConnector connected!");
+                                };
+
+                                socket.ConnectAsync().Wait();
+
+                                _socket = socket;
+                            }
                         }
                     }
+                    else
+                    {
+                        throw new Exception("Missing Socket Port");
+                    }
                 }
 
                 return _socket;
             }
         }
+
+        private class CamelCaseNewtonsoftJsonSerializer : NewtonsoftJsonSerializer
+        {
+            public CamelCaseNewtonsoftJsonSerializer(int eio) : base(eio)
+            {
+            }
+
+            public override JsonSerializerSettings CreateOptions()
+            {
+                return new JsonSerializerSettings()
+                {
+                    ContractResolver = new CamelCasePropertyNamesContractResolver(),
+                    NullValueHandling = NullValueHandling.Ignore,
+                    DefaultValueHandling = DefaultValueHandling.Ignore
+                };
+            }
+        }
     }
 }

ElectronNET.API/AutoUpdater.cs

@@ -20,5 +20,16 @@ public static class BridgeSettings
         /// The web port.
         /// </value>
         public static string WebPort { get; internal set; }
+
+        /// <summary>
+        /// Manually set the port values instead of using the UseElectron extension method
+        /// </summary>
+        /// <param name="socketPort"></param>
+        /// <param name="webPort"></param>
+        public static void InitializePorts(int socketPort, int webPort)
+        {
+            SocketPort = socketPort.ToString();
+            WebPort    = webPort.ToString();
+        }
     }
 }