ogoun
/
Zero
mirror of https://github.com/ogoun/Zero.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Discovery update

Browse Source
pull/1/head
a.bozhenov 5 years ago
parent 9d03789686
commit 9a7d3d6a36
4 changed files with 238 additions and 289 deletions
Show Stats Download Patch File Download Diff File

3
ZeroLevel/Services/BaseZeroService.cs
Unescape View File

@ -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)
{

44
ZeroLevel/Services/Network/Contracts/IClientSet.cs
Unescape View File

@ -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);
}
}

192
ZeroLevel/Services/Network/ExClient.cs
Unescape View File

@ -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
{

288
ZeroLevel/Services/Network/ExServiceHost.cs
Unescape View File

@ -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>

Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.