using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; using CryptoExchange.Net.Interfaces; using CryptoExchange.Net.Sockets; using Microsoft.Extensions.Logging; namespace CryptoExchange.Net { /// /// Base for socket client implementations /// public abstract class BaseSocketClient: BaseClient, ISocketClient { #region fields /// /// If client is disposing /// protected bool _disposing; /// public int CurrentConnections => ApiClients.OfType().Sum(c => c.CurrentConnections); /// public int CurrentSubscriptions => ApiClients.OfType().Sum(s => s.CurrentSubscriptions); /// public double IncomingKbps => ApiClients.OfType().Sum(s => s.IncomingKbps); #endregion /// /// ctor /// /// Logger /// The name of the API this client is for protected BaseSocketClient(ILoggerFactory? logger, string name) : base(logger, name) { } /// /// Unsubscribe an update subscription /// /// The id of the subscription to unsubscribe /// public virtual async Task UnsubscribeAsync(int subscriptionId) { foreach(var socket in ApiClients.OfType()) { var result = await socket.UnsubscribeAsync(subscriptionId).ConfigureAwait(false); if (result) break; } } /// /// Unsubscribe an update subscription /// /// The subscription to unsubscribe /// public virtual async Task UnsubscribeAsync(UpdateSubscription subscription) { if (subscription == null) throw new ArgumentNullException(nameof(subscription)); _logger.Log(LogLevel.Information, $"Socket {subscription.SocketId} Unsubscribing subscription " + subscription.Id); await subscription.CloseAsync().ConfigureAwait(false); } /// /// Unsubscribe all subscriptions /// /// public virtual async Task UnsubscribeAllAsync() { var tasks = new List(); foreach (var client in ApiClients.OfType()) tasks.Add(client.UnsubscribeAllAsync()); await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false); } /// /// Reconnect all connections /// /// public virtual async Task ReconnectAsync() { _logger.Log(LogLevel.Information, $"Reconnecting all {CurrentConnections} connections"); var tasks = new List(); foreach (var client in ApiClients.OfType()) { tasks.Add(client.ReconnectAsync()); } await Task.WhenAll(tasks.ToArray()).ConfigureAwait(false); } /// /// Log the current state of connections and subscriptions /// public string GetSubscriptionsState() { var result = new StringBuilder(); foreach(var client in ApiClients.OfType()) result.AppendLine(client.GetSubscriptionsState()); return result.ToString(); } } }