From 9a7d3d6a36a204eb66bad1e1ba62ca36a73f2387 Mon Sep 17 00:00:00 2001
From: "a.bozhenov" <ogoun@users.noreply.github.com>
Date: Fri, 5 Jul 2019 21:19:48 +0300
Subject: [PATCH] Discovery update

---
 ZeroLevel/Services/BaseZeroService.cs         |   3 +-
 .../Services/Network/Contracts/IClientSet.cs  |  44 +++
 ZeroLevel/Services/Network/ExClient.cs        | 192 ++++++++++++
 ZeroLevel/Services/Network/ExServiceHost.cs   | 288 ------------------
 4 files changed, 238 insertions(+), 289 deletions(-)
 create mode 100644 ZeroLevel/Services/Network/Contracts/IClientSet.cs

diff --git a/ZeroLevel/Services/BaseZeroService.cs b/ZeroLevel/Services/BaseZeroService.cs
index b84464e..dd5268e 100644
--- a/ZeroLevel/Services/BaseZeroService.cs
+++ b/ZeroLevel/Services/BaseZeroService.cs
@@ -141,6 +141,7 @@ namespace ZeroLevel.Services.Applications
             {
                 Sheduller.Remove(_register_in_discovery_table_task);
             }
+            RegisterServicesInDiscovery();
             _update_discovery_table_task = Sheduller.RemindEvery(_update_discovery_table_period, RegisterServicesInDiscovery);
             _register_in_discovery_table_task = Sheduller.RemindEvery(_register_in_discovery_table_period, () => { });
         }
@@ -418,7 +419,7 @@ namespace ZeroLevel.Services.Applications
 
 
         public void StoreConnection(string endpoint)
-        {
+        { 
             if (_state == ZeroServiceStatus.Running ||
                 _state == ZeroServiceStatus.Initialized)
             {
diff --git a/ZeroLevel/Services/Network/Contracts/IClientSet.cs b/ZeroLevel/Services/Network/Contracts/IClientSet.cs
new file mode 100644
index 0000000..dc066ac
--- /dev/null
+++ b/ZeroLevel/Services/Network/Contracts/IClientSet.cs
@@ -0,0 +1,44 @@
+using System;
+using System.Collections.Generic;
+using ZeroLevel.Models;
+
+namespace ZeroLevel.Network
+{
+    public interface IClientSet
+    {
+        InvokeResult Send<T>(string alias, T data);
+        InvokeResult Send<T>(string alias, string inbox, T data);
+        InvokeResult Request<Tresponse>(string alias, Action<Tresponse> callback);
+        InvokeResult Request<Tresponse>(string alias, string inbox, Action<Tresponse> callback);
+        InvokeResult Request<Trequest, Tresponse>(string alias, Trequest request, Action<Tresponse> callback);
+        InvokeResult Request<Trequest, Tresponse>(string alias, string inbox, Trequest request, Action<Tresponse> callback);
+
+
+        InvokeResult SendBroadcast<T>(string alias, T data);
+        InvokeResult SendBroadcast<T>(string alias, string inbox, T data);
+
+        InvokeResult SendBroadcastByType<T>(string serviceType, T data);
+        InvokeResult SendBroadcastByType<T>(string serviceType, string inbox, T data);
+
+        InvokeResult SendBroadcastByGroup<T>(string serviceGroup, T data);
+        InvokeResult SendBroadcastByGroup<T>(string serviceGroup, string inbox, T data);
+
+        InvokeResult RequestBroadcast<Tresponse>(string alias, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcast<Tresponse>(string alias, string inbox, Action<IEnumerable<Tresponse>> callback);
+
+        InvokeResult RequestBroadcast<Trequest, Tresponse>(string alias, Trequest data, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcast<Trequest, Tresponse>(string alias, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback);
+
+        InvokeResult RequestBroadcastByType<Tresponse>(string serviceType, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcastByType<Tresponse>(string serviceType, string inbox, Action<IEnumerable<Tresponse>> callback);
+
+        InvokeResult RequestBroadcastByType<Trequest, Tresponse>(string serviceType, Trequest data, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcastByType<Trequest, Tresponse>(string serviceType, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback);
+
+        InvokeResult RequestBroadcastByGroup<Tresponse>(string serviceGroup, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcastByGroup<Tresponse>(string serviceGroup, string inbox, Action<IEnumerable<Tresponse>> callback);
+
+        InvokeResult RequestBroadcastByGroup<Trequest, Tresponse>(string serviceGroup, Trequest data, Action<IEnumerable<Tresponse>> callback);
+        InvokeResult RequestBroadcastByGroup<Trequest, Tresponse>(string serviceGroup, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback);
+    }
+}
diff --git a/ZeroLevel/Services/Network/ExClient.cs b/ZeroLevel/Services/Network/ExClient.cs
index f2a1ba5..310d5ad 100644
--- a/ZeroLevel/Services/Network/ExClient.cs
+++ b/ZeroLevel/Services/Network/ExClient.cs
@@ -1,10 +1,202 @@
 using System;
+using System.Collections.Generic;
 using System.Net;
+using System.Threading;
+using System.Threading.Tasks;
 using ZeroLevel.Models;
 using ZeroLevel.Services.Serialization;
 
 namespace ZeroLevel.Network
 {
+    public sealed class ExClientSet
+        : IClientSet, IDisposable
+    {
+        public void Dispose()
+        {
+            throw new NotImplementedException();
+        }
+
+        #region IMultiClient
+        public InvokeResult Request<Tresponse>(string alias, Action<Tresponse> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult Request<Tresponse>(string alias, string inbox, Action<Tresponse> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult Request<Trequest, Tresponse>(string alias, Trequest request, Action<Tresponse> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult Request<Trequest, Tresponse>(string alias, string inbox, Trequest request, Action<Tresponse> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcast<Tresponse>(string alias, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcast<Tresponse>(string alias, string inbox, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcast<Trequest, Tresponse>(string alias, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcast<Trequest, Tresponse>(string alias, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByGroup<Tresponse>(string serviceGroup, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByGroup<Tresponse>(string serviceGroup, string inbox, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByGroup<Trequest, Tresponse>(string serviceGroup, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByGroup<Trequest, Tresponse>(string serviceGroup, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByType<Tresponse>(string serviceType, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByType<Tresponse>(string serviceType, string inbox, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByType<Trequest, Tresponse>(string serviceType, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult RequestBroadcastByType<Trequest, Tresponse>(string serviceType, string inbox, Trequest data, Action<IEnumerable<Tresponse>> callback)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult Send<T>(string alias, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult Send<T>(string alias, string inbox, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcast<T>(string alias, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcast<T>(string alias, string inbox, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcastByGroup<T>(string serviceGroup, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcastByGroup<T>(string serviceGroup, string inbox, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcastByType<T>(string serviceType, T data)
+        {
+            throw new NotImplementedException();
+        }
+
+        public InvokeResult SendBroadcastByType<T>(string serviceType, string inbox, T data)
+        {
+            throw new NotImplementedException();
+        }
+        #endregion
+
+        #region Private
+        private IEnumerable<Tresp> _RequestBroadcast<Treq, Tresp>(List<ExClient> clients, string inbox, Treq data)
+        {
+            var response = new List<Tresp>();
+            using (var waiter = new CountdownEvent(clients.Count))
+            {
+                foreach (var client in clients)
+                {
+                    Task.Run(() =>
+                    {
+                        try
+                        {
+                            if (false == client.Request<Treq, Tresp>(inbox, data, resp => { waiter.Signal(); response.Add(resp); }).Success)
+                            {
+                                waiter.Signal();
+                            }
+                        }
+                        catch (Exception ex)
+                        {
+                            Log.SystemError(ex, $"[ExClientSet._RequestBroadcast] Error direct request to service '{client.EndPoint}' in broadcast request. Inbox '{inbox}'");
+                            waiter.Signal();
+                        }
+                    });
+                }
+                waiter.Wait(BaseSocket.MAX_REQUEST_TIME_MS);
+            }
+            return response;
+        }
+
+        private IEnumerable<Tresp> _RequestBroadcast<Tresp>(List<ExClient> clients, string inbox)
+        {
+            var response = new List<Tresp>();
+            using (var waiter = new CountdownEvent(clients.Count))
+            {
+                foreach (var client in clients)
+                {
+                    Task.Run(() =>
+                    {
+                        try
+                        {
+                            if (false == client.Request<Tresp>(inbox, resp => { waiter.Signal(); response.Add(resp); }).Success)
+                            {
+                                waiter.Signal();
+                            }
+                        }
+                        catch (Exception ex)
+                        {
+                            Log.SystemError(ex, $"[ExClientSet._RequestBroadcast] Error direct request to service '{client.EndPoint}' in broadcast request. Inbox '{inbox}'");
+                            waiter.Signal();
+                        }
+                    });
+                }
+                waiter.Wait(BaseSocket.MAX_REQUEST_TIME_MS);
+            }
+            return response;
+        }
+        #endregion
+    }
+
     public sealed class ExClient
         : IClient, IDisposable
     {
diff --git a/ZeroLevel/Services/Network/ExServiceHost.cs b/ZeroLevel/Services/Network/ExServiceHost.cs
index 272d014..d0de213 100644
--- a/ZeroLevel/Services/Network/ExServiceHost.cs
+++ b/ZeroLevel/Services/Network/ExServiceHost.cs
@@ -11,294 +11,6 @@ namespace ZeroLevel.Network
     public sealed class ExServiceHost
         : IDisposable
     {
-        private class MetaService
-        {
-            public ExServiceInfo ServiceInfo { get; set; }
-            public IExService Server { get; set; }
-        }
-
-        private bool _disposed = false;
-        private readonly long _registerTaskKey = -1;
-        private readonly IDiscoveryClient _discoveryClient;
-
-        private readonly ConcurrentDictionary<string, MetaService> _services
-            = new ConcurrentDictionary<string, MetaService>();
-
-        public ExServiceHost(IDiscoveryClient client)
-        {
-            _discoveryClient = client;
-            _registerTaskKey = Sheduller.RemindEvery(TimeSpan.FromMilliseconds(50), TimeSpan.FromSeconds(55), RegisterServicesInDiscovery);
-        }
-
-        public IExService RegisterService(IExchangeService service)
-        {
-            try
-            {
-                if (_disposed) throw new ObjectDisposedException("ExServiceHost");
-                if (service == null) throw new ArgumentNullException(nameof(service));
-                ValidateService(service);
-                if (_services.ContainsKey(service.Key))
-                {
-                    throw new Exception($"[ExServiceHost] Service {service.Key} already registered");
-                }
-                var server = ExchangeTransportFactory.GetServer();
-                if (false == _services.TryAdd(service.Key, new MetaService
-                {
-                    Server = server,
-                    ServiceInfo = new ExServiceInfo
-                    {
-                        Port = server.Endpoint.Port,
-                        ServiceKey = service.Key,
-                        Version = service.Version,
-                        ServiceGroup = service.Group,
-                        ServiceType = service.Type
-                    }
-                }))
-                {
-                    server.Dispose();
-                    return null;
-                }
-
-                RegisterServiceInboxes(service);
-
-                return server;
-            }
-            catch (Exception ex)
-            {
-                Log.SystemError(ex, "[ExServiceHost] Fault register service");
-                return null;
-            }
-        }
-
-        public IExService RegisterService(ExServiceInfo serviceInfo)
-        {
-            try
-            {
-                if (_disposed) throw new ObjectDisposedException("ExServiceHost");
-                if (serviceInfo == null) throw new ArgumentNullException(nameof(serviceInfo));
-                ValidateService(serviceInfo);
-
-                if (_services.ContainsKey(serviceInfo.ServiceKey))
-                {
-                    throw new Exception($"[ExServiceHost] Service {serviceInfo.ServiceKey} already registered");
-                }
-
-                var server = ExchangeTransportFactory.GetServer();
-                if (false == _services.TryAdd(serviceInfo.ServiceKey, new MetaService
-                {
-                    Server = server,
-                    ServiceInfo = new ExServiceInfo
-                    {
-                        Port = server.Endpoint.Port,
-                        ServiceKey = serviceInfo.ServiceKey,
-                        Version = serviceInfo.Version,
-                        ServiceGroup = serviceInfo.ServiceGroup,
-                        ServiceType = serviceInfo.ServiceType
-                    }
-                }))
-                {
-                    server.Dispose();
-                    return null;
-                }
-                return server;
-            }
-            catch (Exception ex)
-            {
-                Log.SystemError(ex, "[ExServiceHost] Fault register service");
-                return null;
-            }
-        }
-
-        #region Private methods
-
-        private void ValidateService(IExchangeService service)
-        {
-            if (string.IsNullOrWhiteSpace(service.Key))
-            {
-                throw new ArgumentNullException("Service.Key");
-            }
-        }
-
-        private void ValidateService(ExServiceInfo service)
-        {
-            if (string.IsNullOrWhiteSpace(service.ServiceKey))
-            {
-                throw new ArgumentNullException("ServiceKey");
-            }
-        }
-
-        private void RegisterServiceInboxes(IExchangeService service)
-        {
-            MethodInfo[] methods = service.
-                GetType().
-                GetMethods(BindingFlags.NonPublic | BindingFlags.Public |
-                BindingFlags.Instance |
-                BindingFlags.FlattenHierarchy |
-                BindingFlags.Instance);
-
-            var registerHandler = this.GetType().GetMethod("RegisterHandler");
-            var registerReplier = this.GetType().GetMethod("RegisterReplier");
-            var registerReplierWithNoRequestBody = this.GetType().GetMethod("RegisterReplierWithNoRequestBody");
-
-            foreach (MethodInfo mi in methods)
-            {
-                try
-                {
-                    foreach (Attribute attr in Attribute.GetCustomAttributes(mi, typeof(ExchangeAttribute)))
-                    {
-                        if (attr.GetType() == typeof(ExchangeMainHandlerAttribute))
-                        {
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerHandler.MakeGenericMethod(firstArgType);
-                            genericMethod.Invoke(this, new object[] { ZBaseNetwork.DEFAULT_MESSAGE_INBOX, CreateDelegate(mi, service) });
-                        }
-                        else if (attr.GetType() == typeof(ExchangeHandlerAttribute))
-                        {
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerHandler.MakeGenericMethod(firstArgType);
-                            genericMethod.Invoke(this, new object[] { (attr as ExchangeHandlerAttribute).Inbox, CreateDelegate(mi, service) });
-                        }
-                        else if (attr.GetType() == typeof(ExchangeMainReplierAttribute))
-                        {
-                            var returnType = mi.ReturnType;
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerReplier.MakeGenericMethod(firstArgType, returnType);
-                            genericMethod.Invoke(this, new object[] { ZBaseNetwork.DEFAULT_REQUEST_INBOX, CreateDelegate(mi, service) });
-                        }
-                        else if (attr.GetType() == typeof(ExchangeReplierAttribute))
-                        {
-                            var returnType = mi.ReturnType;
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerReplier.MakeGenericMethod(firstArgType, returnType);
-                            genericMethod.Invoke(this, new object[] { (attr as ExchangeReplierAttribute).Inbox, CreateDelegate(mi, service) });
-                        }
-                        else if (attr.GetType() == typeof(ExchangeMainReplierWithoutArgAttribute))
-                        {
-                            var returnType = mi.ReturnType;
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerReplierWithNoRequestBody.MakeGenericMethod(returnType);
-                            genericMethod.Invoke(this, new object[] { ZBaseNetwork.DEFAULT_REQUEST_INBOX, CreateDelegate(mi, service) });
-                        }
-                        else if (attr.GetType() == typeof(ExchangeReplierWithoutArgAttribute))
-                        {
-                            var returnType = mi.ReturnType;
-                            var firstArgType = mi.GetParameters().First().ParameterType;
-                            MethodInfo genericMethod = registerReplierWithNoRequestBody.MakeGenericMethod(returnType);
-                            genericMethod.Invoke(this, new object[] { (attr as ExchangeReplierAttribute).Inbox, CreateDelegate(mi, service) });
-                        }
-                    }
-                }
-                catch (Exception ex)
-                {
-                    Log.Debug($"[ZExchange] Can't register method {mi.Name} as inbox handler. {ex}");
-                }
-            }
-        }
-
-        private void RegisterServicesInDiscovery()
-        {
-            var services = _services.
-                Values.
-                Select(s => s.ServiceInfo).
-                ToList();
-            foreach (var service in services)
-            {
-                _discoveryClient.Register(service);
-            }
-        }
-
-        #endregion Private methods
-
-        #region Utils
-
-        private static Delegate CreateDelegate(MethodInfo methodInfo, object target)
-        {
-            Func<Type[], Type> getType;
-            var isAction = methodInfo.ReturnType.Equals((typeof(void)));
-            var types = methodInfo.GetParameters().Select(p => p.ParameterType);
-            if (isAction)
-            {
-                getType = Expression.GetActionType;
-            }
-            else
-            {
-                getType = Expression.GetFuncType;
-                types = types.Concat(new[] { methodInfo.ReturnType });
-            }
-            if (methodInfo.IsStatic)
-            {
-                return Delegate.CreateDelegate(getType(types.ToArray()), methodInfo);
-            }
-            return Delegate.CreateDelegate(getType(types.ToArray()), target, methodInfo.Name);
-        }
-
-        #endregion Utils
-
-        #region Inboxes
-
-        /// <summary>
-        /// Registering an Inbox Handler
-        /// </summary>
-        /// <typeparam name="T">Message type</typeparam>
-        /// <param name="protocol">Protocol</param>
-        /// <param name="inbox">Inbox name</param>
-        /// <param name="handler">Handler</param>
-        private void RegisterHandler<T>(MetaService meta, string inbox, Action<T, long, IZBackward> handler)
-        {
-            if (_disposed) return;
-            try
-            {
-                meta.Server.RegisterInbox(inbox, handler);
-            }
-            catch (Exception ex)
-            {
-                Log.SystemError(ex, $"[Exchange] Register inbox handler error. Inbox '{inbox}'. Service '{meta.ServiceInfo.ServiceKey}'");
-            }
-        }
-
-        /// <summary>
-        /// Registration method responding to an incoming request
-        /// </summary>
-        /// <typeparam name="Treq">Request message type</typeparam>
-        /// <typeparam name="Tresp">Response message type</typeparam>
-        /// <param name="protocol">Protocol</param>
-        /// <param name="inbox">Inbox name</param>
-        /// <param name="replier">Handler</param>
-        private void RegisterReplier<Treq, Tresp>(MetaService meta, string inbox, Func<Treq, long, IZBackward, Tresp> handler)
-        {
-            if (_disposed) return;
-            try
-            {
-                meta.Server.RegisterInbox(inbox, handler);
-            }
-            catch (Exception ex)
-            {
-                Log.SystemError(ex, $"[Exchange] Register inbox replier error. Inbox '{inbox}'. Service '{meta.ServiceInfo.ServiceKey}'");
-            }
-        }
-
-        /// <summary>
-        /// Registration of the method of responding to the incoming request, not receiving incoming data
-        /// </summary>
-        /// <typeparam name="Tresp">Response message type</typeparam>
-        /// <param name="protocol">Protocol</param>
-        /// <param name="inbox">Inbox name</param>
-        /// <param name="replier">Handler</param>
-        private void RegisterReplierWithNoRequestBody<Tresp>(MetaService meta, string inbox, Func<long, IZBackward, Tresp> handler)
-        {
-            if (_disposed) return;
-            try
-            {
-                meta.Server.RegisterInbox(inbox, handler);
-            }
-            catch (Exception ex)
-            {
-                Log.SystemError(ex, $"[Exchange] Register inbox replier error. Inbox '{inbox}'. Service '{meta.ServiceInfo.ServiceKey}'");
-            }
-        }
-
-        #endregion Inboxes
-
         #region Transport helpers
 
         /// <summary>