CryptoExchange.Net/CryptoExchange.Net/Sockets/Query.cs

using CryptoExchange.Net.Interfaces;
using CryptoExchange.Net.Objects;
using CryptoExchange.Net.Sockets.Default;
using CryptoExchange.Net.Sockets.Interfaces;
using System;
using System.Threading;
using System.Threading.Tasks;

namespace CryptoExchange.Net.Sockets
{
    /// <summary>
    /// Query 
    /// </summary>
    public abstract class Query : IMessageProcessor
    {
        /// <summary>
        /// Unique identifier
        /// </summary>
        public int Id { get; } = ExchangeHelpers.NextId();

        /// <summary>
        /// Has this query been completed
        /// </summary>
        public bool Completed { get; set; }

        /// <summary>
        /// Timeout for the request
        /// </summary>
        public TimeSpan? RequestTimeout { get; set; }

        /// <summary>
        /// What should happen if the query times out
        /// </summary>
        public TimeoutBehavior TimeoutBehavior { get; set; } = TimeoutBehavior.Fail;

        /// <summary>
        /// The number of required responses. Can be more than 1 when for example subscribing multiple symbols streams in a single request,
        /// and each symbol receives it's own confirmation response
        /// </summary>
        public int RequiredResponses { get; set; } = 1;

        /// <summary>
        /// The current number of responses received on this query
        /// </summary>
        public int CurrentResponses { get; set; }

        /// <summary>
        /// Timestamp of when the request was send
        /// </summary>
        public DateTime RequestTimestamp { get; set; }
        
        /// <summary>
        /// Result
        /// </summary>
        public CallResult? Result { get; set; }
        
        /// <summary>
        /// Response
        /// </summary>
        public object? Response { get; set; }

        private MessageRouter _router;
        /// <summary>
        /// Router for this subscription
        /// </summary>
        public MessageRouter MessageRouter
        {
            get => _router;
            set
            {
                _router = value;
                _router.BuildRouteMap();
            }
        }

        /// <summary>
        /// The query request object
        /// </summary>
        public object Request { get; set; }

        /// <summary>
        /// If this is a private request
        /// </summary>
        public bool Authenticated { get; }

        /// <summary>
        /// Weight of the query
        /// </summary>
        public int Weight { get; }

        /// <summary>
        /// Whether the query should wait for a response or not
        /// </summary>
        public bool ExpectsResponse { get; set; } = true;

        /// <summary>
        /// Wait event for response
        /// </summary>
        protected AsyncResetEvent _event;

        /// <summary>
        /// Cancellation token
        /// </summary>
        protected CancellationTokenSource? _cts;

        /// <summary>
        /// On complete callback
        /// </summary>
        public Action? OnComplete { get; set; }

        /// <summary>
        /// ctor
        /// </summary>
#pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider adding the 'required' modifier or declaring as nullable.
        public Query(object request, bool authenticated, int weight = 1)
#pragma warning restore CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider adding the 'required' modifier or declaring as nullable.
        {
            _event = new AsyncResetEvent(false, false);

            Authenticated = authenticated;
            Request = request;
            Weight = weight;
        }

        /// <summary>
        /// Signal that the request has been send and the timeout timer should start
        /// </summary>
        public void IsSend(TimeSpan timeout)
        {
            RequestTimestamp = DateTime.UtcNow;
            if (ExpectsResponse)
            {
                // Start timeout countdown
                _cts = new CancellationTokenSource(timeout);
                _cts.Token.Register(Timeout, false);
            }
            else
            {
                Result = CallResult.SuccessResult;
                Completed = true;
                _event.Set();
            }
        }

        /// <summary>
        /// Wait until timeout or the request is completed
        /// </summary>
        /// <param name="timeout"></param>
        /// <param name="ct">Cancellation token</param>
        /// <returns></returns>
        public async Task WaitAsync(TimeSpan timeout, CancellationToken ct) => await _event.WaitAsync(timeout, ct).ConfigureAwait(false);

        /// <summary>
        /// Mark request as timeout
        /// </summary>
        public abstract void Timeout();

        /// <summary>
        /// Mark request as failed
        /// </summary>
        /// <param name="error"></param>
        public abstract void Fail(Error error);

        /// <summary>
        /// Handle a response message
        /// </summary>
        public abstract CallResult Handle(SocketConnection connection, DateTime receiveTime, string? originalData, object message, MessageRoute route);

    }

    /// <summary>
    /// Query
    /// </summary>
    /// <typeparam name="THandlerResponse">The type to be returned to the caller</typeparam>
    public abstract class Query<THandlerResponse> : Query
    {
        /// <summary>
        /// The typed call result
        /// </summary>
        public CallResult<THandlerResponse>? TypedResult => (CallResult<THandlerResponse>?)Result;

        /// <summary>
        /// ctor
        /// </summary>
        /// <param name="request"></param>
        /// <param name="authenticated"></param>
        /// <param name="weight"></param>
        protected Query(
            object request,
            bool authenticated,
            int weight = 1)
            : base(request, authenticated, weight)
        {
        }

        /// <inheritdoc />
        public override CallResult Handle(SocketConnection connection, DateTime receiveTime, string? originalData, object message, MessageRoute route)
        {
            CurrentResponses++;
            if (CurrentResponses == RequiredResponses)
                Response = message;

            if (Result?.Success != false)
            {
                // If an error result is already set don't override that
                Result = route.Handle(connection, receiveTime, originalData, message);
                if (Result == null)
                    // Null from Handle means it wasn't actually for this query
                    CurrentResponses -= 1;
            }

            if (CurrentResponses == RequiredResponses)
            {
                Completed = true;
                _event.Set();
                OnComplete?.Invoke();
            }

            return Result ?? CallResult.SuccessResult;
        }

        /// <inheritdoc />
        public override void Timeout()
        {
            if (Completed)
                return;
                        
            if (TimeoutBehavior == TimeoutBehavior.Fail)
                Result = new CallResult<THandlerResponse>(new TimeoutError());
            else
                Result = new CallResult<THandlerResponse>(default, null, default);

            Completed = true;
            _event.Set();
            OnComplete?.Invoke();
        }

        /// <inheritdoc />
        public override void Fail(Error error)
        {
            if (Completed)
                return;

            Result = new CallResult<THandlerResponse>(error);
            Completed = true;

            _event.Set();
            OnComplete?.Invoke();
        }
    }
}