using System; using System.Collections.Generic; using System.Linq; using System.Net; using System.Threading; using System.Threading.Tasks; using ZeroLevel.Models; using ZeroLevel.Network.SDL; namespace ZeroLevel.Network { /// /// Provides data exchange between services /// internal sealed class Exchange : IExchange { private readonly ServiceRouteStorage _dicovery_aliases = new ServiceRouteStorage(); private readonly ServiceRouteStorage _user_aliases = new ServiceRouteStorage(); private readonly ExClientServerCachee _cachee = new ExClientServerCachee(); public IServiceRoutesStorage RoutesStorage => _user_aliases; public IServiceRoutesStorage DiscoveryStorage => _dicovery_aliases; private readonly IZeroService _owner; #region Ctor public Exchange(IZeroService owner) { _owner = owner; } #endregion Ctor #region IMultiClient public bool Peek(string alias, string inbox) { return CallService(alias, (transport) => transport.Send(inbox)); } /// /// Sending a message to the service /// /// Service key or url /// Message /// public bool Send(string alias, T data) { return CallService(alias, (transport) => transport.Send(BaseSocket.DEFAULT_MESSAGE_INBOX, data)); } /// /// Sending a message to the service /// /// Service key or url /// Inbox name /// Message /// public bool Send(string alias, string inbox, T data) { return CallService(alias, (transport) => transport.Send(inbox, data)); } /// /// Sending a message to all services with the specified key, to the default handler /// /// Message type /// Service key /// Message /// true - on successful submission public bool SendBroadcast(string serviceKey, T data) => SendBroadcast(serviceKey, BaseSocket.DEFAULT_MESSAGE_INBOX, data); /// /// Sending a message to all services with the specified key to the specified handler /// /// Message type /// Service key /// Inbox name /// Message /// true - on successful submission public bool SendBroadcast(string alias, string inbox, T data) { var result = false; try { var clients = GetClientEnumerator(alias).ToArray(); var tasks = new Task[clients.Length]; int index = 0; foreach (var client in clients) { tasks[index++] = Task.Run(() => { try { result |= client.Send(inbox, data); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcast] Error broadcast send data to services '{alias}'. Inbox '{inbox}'"); } }); } Task.WaitAll(tasks); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcast] Error broadcast send data in service '{alias}'. Inbox '{inbox}'"); } return result; } /// /// Sending a message to all services of a specific type to the specified handler /// /// Message type /// Service type /// Inbox name /// Message /// true - on successful submission public bool SendBroadcastByType(string type, string inbox, T data) { var result = false; try { var clients = GetClientEnumeratorByType(type).ToArray(); var tasks = new Task[clients.Length]; int index = 0; foreach (var client in clients) { tasks[index++] = Task.Run(() => { try { result |= client.Send(inbox, data); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcastByType] Error broadcast send data to services with type '{type}'. Inbox '{inbox}'"); } }); } Task.WaitAll(tasks); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcastByType] Error broadcast send data to services with type '{type}'. Inbox '{inbox}'"); } return result; } /// /// Sending a message to all services of a particular type, to the default handler /// /// Message type /// Service type /// Message /// true - on successful submission public bool SendBroadcastByType(string type, T data) => SendBroadcastByType(type, BaseSocket.DEFAULT_MESSAGE_INBOX, data); /// /// Sending a message to all services of a specific group to the specified handler /// /// Message type /// Service group /// Inbox name /// Message /// true - on successful submission public bool SendBroadcastByGroup(string group, string inbox, T data) { var result = false; try { var clients = GetClientEnumeratorByGroup(group).ToArray(); var tasks = new Task[clients.Length]; int index = 0; foreach (var client in clients) { tasks[index++] = Task.Run(() => { try { result |= client.Send(inbox, data); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcastByGroup] Error broadcast send data to services with type '{group}'. Inbox '{inbox}'"); } }); } Task.WaitAll(tasks); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.SendBroadcastByGroup] Error broadcast send data to services with type '{group}'. Inbox '{inbox}'"); } return result; } /// /// Sending a message to all services of a specific group in the default handler /// /// Message type /// Service group /// Messsage /// true - on successful submission public bool SendBroadcastByGroup(string serviceGroup, T data) => SendBroadcastByGroup(serviceGroup, BaseSocket.DEFAULT_MESSAGE_INBOX, data); public bool Request(string alias, Action callback) => Request(alias, BaseSocket.DEFAULT_REQUEST_WITHOUT_ARGS_INBOX, callback); public bool Request(string alias, string inbox, Action callback) { try { return CallService(alias, (transport) => { try { return transport.Request(inbox, callback); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.Request] Error request to service '{alias}'. Inbox '{inbox}'"); } return false; }); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.Request] Error request to service '{alias}'. Inbox '{inbox}'"); } return false; } public bool Request(string alias, Trequest request, Action callback) => Request(alias, BaseSocket.DEFAULT_REQUEST_INBOX, callback); public bool Request(string alias, string inbox, Trequest request, Action callback) { try { return CallService(alias, (transport) => { try { return transport.Request(inbox, request, callback); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.Request] Error request to service '{alias}'. Inbox '{inbox}'"); } return false; }); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.Request] Error request to service '{alias}'. Inbox '{inbox}'"); } return false; } /// /// Broadcast polling of services by key, without message of request, to default handler /// /// Response message type /// Service key /// Response handler /// true - in case of successful mailing public bool RequestBroadcast(string alias, Action> callback) => RequestBroadcast(alias, BaseSocket.DEFAULT_REQUEST_WITHOUT_ARGS_INBOX, callback); /// /// Broadcast polling services by key /// /// Response message type /// Service key /// Inbox name /// Request message /// Response handler /// true - in case of successful mailing public bool RequestBroadcast(string alias, string inbox, Action> callback) { try { var clients = GetClientEnumerator(alias).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.RequestBroadcast] Error broadcast request to service '{alias}'. Inbox '{inbox}'"); } return false; } public bool RequestBroadcast(string alias, Trequest data, Action> callback) => RequestBroadcast(alias, BaseSocket.DEFAULT_REQUEST_INBOX, data, callback); public bool RequestBroadcast(string alias, string inbox, Trequest data , Action> callback) { try { var clients = GetClientEnumerator(alias).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox, data)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.RequestBroadcast] Error broadcast request to service '{alias}'. Inbox '{inbox}'"); } return false; } public bool RequestBroadcastByGroup(string serviceGroup, Action> callback) => RequestBroadcastByGroup(serviceGroup, BaseSocket.DEFAULT_REQUEST_INBOX, callback); public bool RequestBroadcastByGroup(string serviceGroup, string inbox, Action> callback) { try { var clients = GetClientEnumeratorByGroup(serviceGroup).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange] Error broadcast request to service by group '{serviceGroup}'. Inbox '{inbox}'"); } return false; } public bool RequestBroadcastByGroup(string serviceGroup, Trequest data, Action> callback) => RequestBroadcastByGroup(serviceGroup, BaseSocket.DEFAULT_REQUEST_INBOX, data, callback); public bool RequestBroadcastByGroup(string serviceGroup, string inbox, Trequest data , Action> callback) { try { var clients = GetClientEnumeratorByGroup(serviceGroup).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox, data)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange] Error broadcast request to service by group '{serviceGroup}'. Inbox '{inbox}'"); } return false; } public bool RequestBroadcastByType(string serviceType, Action> callback) => RequestBroadcastByType(serviceType, BaseSocket.DEFAULT_REQUEST_WITHOUT_ARGS_INBOX, callback); public bool RequestBroadcastByType(string serviceType, string inbox, Action> callback) { try { var clients = GetClientEnumeratorByType(serviceType).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange] Error broadcast request to service by type '{serviceType}'. Inbox '{inbox}'"); } return false; } /// /// Broadcast polling services by type of service, to default handler /// /// Request message type /// Response message type /// Service type /// Request message /// Response handler /// true - in case of successful mailing public bool RequestBroadcastByType(string serviceType, Trequest data , Action> callback) => RequestBroadcastByType(serviceType, BaseSocket.DEFAULT_REQUEST_INBOX, data, callback); /// /// Broadcast polling services by type of service /// /// Request message type /// Response message type /// Service type /// Inbox name /// Request message /// Response handler /// true - in case of successful mailing public bool RequestBroadcastByType(string serviceType, string inbox, Trequest data , Action> callback) { try { var clients = GetClientEnumeratorByType(serviceType).ToList(); if (clients.Count > 0) { callback(_RequestBroadcast(clients, inbox, data)); return true; } } catch (Exception ex) { Log.SystemError(ex, $"[Exchange] Error broadcast request to service by type '{serviceType}'. Inbox '{inbox}'"); } return false; } #endregion #region Discovery private long _update_discovery_table_task = -1; private long _register_in_discovery_table_task = -1; private static TimeSpan _update_discovery_table_period = TimeSpan.FromSeconds(15); private static TimeSpan _register_in_discovery_table_period = TimeSpan.FromSeconds(15); public void UseDiscovery() { try { var discoveryEndpoint = Configuration.Default.First("discovery"); _user_aliases.Set(BaseSocket.DISCOVERY_ALIAS, NetUtils.CreateIPEndPoint(discoveryEndpoint)); RestartDiscoveryTasks(); } catch (Exception ex) { Log.Error(ex, "[Exchange.UseDiscovery]"); } } public void UseDiscovery(string discoveryEndpoint) { try { _user_aliases.Set(BaseSocket.DISCOVERY_ALIAS, NetUtils.CreateIPEndPoint(discoveryEndpoint)); RestartDiscoveryTasks(); } catch (Exception ex) { Log.Error(ex, "[Exchange.UseDiscovery]"); } } public void UseDiscovery(IPEndPoint discoveryEndpoint) { try { _user_aliases.Set(BaseSocket.DISCOVERY_ALIAS, discoveryEndpoint); RestartDiscoveryTasks(); } catch (Exception ex) { Log.Error(ex, "[Exchange.UseDiscovery]"); } } private void RestartDiscoveryTasks() { if (_update_discovery_table_task != -1) { Sheduller.Remove(_update_discovery_table_task); } if (_register_in_discovery_table_task != -1) { Sheduller.Remove(_register_in_discovery_table_task); } UpdateServiceListFromDiscovery(); _register_in_discovery_table_task = Sheduller.RemindEvery(TimeSpan.FromMilliseconds(500), _update_discovery_table_period, RegisterServicesInDiscovery); _update_discovery_table_task = Sheduller.RemindEvery(TimeSpan.FromMilliseconds(750), _register_in_discovery_table_period, UpdateServiceListFromDiscovery); } private void RegisterServicesInDiscovery() { var discovery_endpoint = _user_aliases.Get(BaseSocket.DISCOVERY_ALIAS); if (discovery_endpoint.Success) { var discoveryClient = _cachee.GetClient(discovery_endpoint.Value, true); if (discoveryClient != null) { foreach (var service in _cachee.ServerList) { var request = discoveryClient.Request("register" , new ServiceRegisterInfo { Port = service.LocalEndpoint.Port, ServiceInfo = _owner.ServiceInfo } , r => { if (!r.Success) { Log.SystemWarning($"[Exchange.RegisterServicesInDiscovery] Register canceled. {r.Comment}"); } }); if (request == false) { Log.SystemWarning($"[Exchange.RegisterServicesInDiscovery] Register canceled."); } } } } } private void UpdateServiceListFromDiscovery() { var discovery_endpoint = _user_aliases.Get(BaseSocket.DISCOVERY_ALIAS); if (discovery_endpoint.Success) { var discoveryClient = _cachee.GetClient(discovery_endpoint.Value, true); if (discoveryClient != null) { try { var ir = discoveryClient.Request>("services", records => { if (records == null) { Log.SystemWarning("[Exchange.UpdateServiceListFromDiscovery] UpdateServiceListInfo. Discrovery response is empty"); return; } _dicovery_aliases.BeginUpdate(); try { foreach (var service in records) { _dicovery_aliases.Set(service.ServiceInfo.ServiceKey , service.ServiceInfo.ServiceType , service.ServiceInfo.ServiceGroup , NetUtils.CreateIPEndPoint(service.Endpoint)); } _dicovery_aliases.Commit(); } catch { _dicovery_aliases.Rollback(); } }); if (!ir) { Log.SystemWarning($"[Exchange.UpdateServiceListFromDiscovery] Error request to inbox 'services'."); } } catch (Exception ex) { Log.SystemError(ex, "[Exchange.UpdateServiceListFromDiscovery] Discovery service response is absent"); } } } } #endregion public IClient GetConnection(string alias) { if (_update_discovery_table_task != -1) { var address = _dicovery_aliases.Get(alias); if (address.Success) { return _cachee.GetClient(address.Value, true); } } else { var address = _user_aliases.Get(alias); if (address.Success) { return _cachee.GetClient(address.Value, true); } try { var endpoint = NetUtils.CreateIPEndPoint(alias); return _cachee.GetClient(endpoint, true); } catch (Exception ex) { Log.SystemError(ex, "[Exchange.GetConnection]"); } } return null; } public IClient GetConnection(IPEndPoint endpoint) { try { return _cachee.GetClient(endpoint, true); } catch (Exception ex) { Log.SystemError(ex, "[Exchange.GetConnection]"); } return null; } public IClient GetConnection(ISocketClient client) { try { return new ExClient(client); } catch (Exception ex) { Log.SystemError(ex, "[Exchange.GetConnection]"); } return null; } #region Host service public IRouter UseHost() { return MakeHost(new IPEndPoint(IPAddress.Any, NetUtils.GetFreeTcpPort())); } public IRouter UseHost(int port) { return MakeHost(new IPEndPoint(IPAddress.Any, port)); } public IRouter UseHostV6() { return MakeHost(new IPEndPoint(IPAddress.IPv6Any, NetUtils.GetFreeTcpPort())); } public IRouter UseHostV6(int port) { return MakeHost(new IPEndPoint(IPAddress.IPv6Any, port)); } public IRouter UseHost(IPEndPoint endpoint) { return MakeHost(endpoint); } private IRouter MakeHost(IPEndPoint endpoint) { var server = _cachee.GetServer(endpoint, new Router()); server.RegisterInbox("__service_description__", _ => CollectServiceDescription()); return server; } #endregion #region Private private IEnumerable GetAllAddresses(string serviceKey) { if (_update_discovery_table_task != -1) { var dr = _dicovery_aliases.GetAll(serviceKey); var ur = _user_aliases.GetAll(serviceKey); if (dr.Success && ur.Success) { return Enumerable.Union(dr.Value, ur.Value); } else if (dr.Success) { return dr.Value; } else if (ur.Success) { return ur.Value; } } else { var result = _user_aliases.GetAll(serviceKey); if (result.Success) { return result.Value; } } return null; } private IEnumerable GetAllAddressesByType(string serviceType) { if (_update_discovery_table_task != -1) { var dr = _dicovery_aliases.GetAllByType(serviceType); var ur = _user_aliases.GetAllByType(serviceType); if (dr.Success && ur.Success) { return Enumerable.Union(dr.Value, ur.Value); } else if (dr.Success) { return dr.Value; } else if (ur.Success) { return ur.Value; } } else { var result = _user_aliases.GetAllByType(serviceType); if (result.Success) { return result.Value; } } return null; } private IEnumerable GetAllAddressesByGroup(string serviceGroup) { if (_update_discovery_table_task != -1) { var dr = _dicovery_aliases.GetAllByGroup(serviceGroup); var ur = _user_aliases.GetAllByGroup(serviceGroup); if (dr.Success && ur.Success) { return Enumerable.Union(dr.Value, ur.Value); } else if (dr.Success) { return dr.Value; } else if (ur.Success) { return ur.Value; } } else { var result = _user_aliases.GetAllByGroup(serviceGroup); if (result.Success) { return result.Value; } } return null; } private IEnumerable GetClientEnumerator(string serviceKey) { IEnumerable candidates; try { candidates = GetAllAddresses(serviceKey); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumerator] Error when trying get endpoints for service key '{serviceKey}'"); candidates = null; } if (candidates != null && candidates.Any()) { foreach (var endpoint in candidates) { IClient transport; try { transport = _cachee.GetClient(endpoint, true); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumerator] Can't get transport for endpoint '{endpoint}'"); continue; } if (transport == null) continue; yield return transport; } } else { Log.Debug($"[Exchange.GetClientEnumerator] Not found endpoints for service key '{serviceKey}'"); } } private IEnumerable GetClientEnumeratorByType(string serviceType) { IEnumerable candidates; try { candidates = GetAllAddressesByType(serviceType); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumeratorByType] Error when trying get endpoints for service type '{serviceType}'"); candidates = null; } if (candidates != null && candidates.Any()) { foreach (var endpoint in candidates) { IClient transport; try { transport = _cachee.GetClient(endpoint, true); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumeratorByType] Can't get transport for endpoint '{endpoint}'"); continue; } if (transport == null) continue; yield return transport; } } else { Log.Debug($"[Exchange.GetClientEnumeratorByType] Not found endpoints for service type '{serviceType}'"); } } private IEnumerable GetClientEnumeratorByGroup(string serviceGroup) { IEnumerable candidates; try { candidates = GetAllAddressesByGroup(serviceGroup); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumeratorByGroup] Error when trying get endpoints for service group '{serviceGroup}'"); candidates = null; } if (candidates != null && candidates.Any()) { foreach (var service in candidates) { IClient transport; try { transport = _cachee.GetClient(service, true); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.GetClientEnumeratorByGroup] Can't get transport for endpoint '{service}'"); continue; } if (transport == null) continue; yield return transport; } } else { Log.Debug($"[Exchange.GetClientEnumeratorByGroup] Not found endpoints for service group '{serviceGroup}'"); } } /// /// Call service with round-robin balancing /// /// Service key /// Service call code /// true - service called succesfully private bool CallService(string serviceKey, Func callHandler) { IEnumerable candidates; try { candidates = GetAllAddresses(serviceKey); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.CallService] Error when trying get endpoints for service key '{serviceKey}'"); return false; } if (candidates == null || candidates.Any() == false) { Log.Debug($"[Exchange.CallService] Not found endpoints for service key '{serviceKey}'"); return false; } var success = false; foreach (var endpoint in candidates) { IClient transport; try { transport = _cachee.GetClient(endpoint, true); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.CallService] Can't get transport for service '{serviceKey}'"); continue; } if (transport == null) continue; try { success = callHandler(transport); } catch (Exception ex) { Log.SystemError(ex, $"[Exchange.CallService] Error send/request data in service '{serviceKey}'. Endpoint '{endpoint}'"); success = false; } if (success) { break; } } return success; } private IEnumerable _RequestBroadcast(List clients, string inbox, Treq data) { var response = new List(); using (var waiter = new CountdownEvent(clients.Count)) { foreach (var client in clients) { Task.Run(() => { try { if (false == client.Request(inbox, data, resp => { response.Add(resp); waiter.Signal(); })) { 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 _RequestBroadcast(List clients, string inbox) { var response = new List(); using (var waiter = new CountdownEvent(clients.Count)) { foreach (var client in clients) { Task.Run(() => { try { if (false == client.Request(inbox, resp => { response.Add(resp); waiter.Signal(); })) { 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 private ServiceDescription CollectServiceDescription() { return new ServiceDescription { ServiceInfo = this._owner?.ServiceInfo, Inboxes = _cachee.ServerList .SelectMany(se => se .CollectInboxInfo() .Select(i => { i.Port = se.LocalEndpoint.Port; return i; })).ToList() }; } public void Dispose() { if (_update_discovery_table_task != -1) { Sheduller.Remove(_update_discovery_table_task); } if (_register_in_discovery_table_task != -1) { Sheduller.Remove(_register_in_discovery_table_task); } _cachee.Dispose(); } } }