diff --git a/MtApi/MtApiClient.cs b/MtApi/MtApiClient.cs index 83d4a714..654444a2 100755 --- a/MtApi/MtApiClient.cs +++ b/MtApi/MtApiClient.cs @@ -27,7 +27,6 @@ namespace MtApi private readonly Dictionary> _mtEventHandlers = []; private HashSet _experts = []; private Dictionary _quotes = []; - private readonly EventWaitHandle _quotesWaiter = new AutoResetEvent(false); private MtConnectionState _connectionState = MtConnectionState.Disconnected; private int _executorHandle; @@ -83,7 +82,6 @@ namespace MtApi /// public List GetQuotes() { - _quotesWaiter.WaitOne(10000); // wait 10 sec for loading all quotes from MetaTrader lock (_locker) { return _quotes.Values.ToList(); @@ -3041,7 +3039,6 @@ namespace MtApi ConnectionStateChanged?.Invoke(this, new MtConnectionEventArgs(MtConnectionState.Connecting, message)); var client = new MtRpcClient(host, port, new RpcClientLogger(Log)); - client.ExpertList += Client_ExpertList; client.ExpertAdded += Client_ExpertAdded; client.ExpertRemoved += Client_ExpertRemoved; client.MtEventReceived += Client_MtEventReceived; @@ -3052,13 +3049,44 @@ namespace MtApi { await client.Connect(); Log.Info($"Connected to {host}:{port}"); + + var experts = client.RequestExpertsList(); + if (experts == null || experts.Count == 0) + { + var errorMessage = "Failed to load expert list"; + Log.Error(errorMessage); + client.Disconnect(); + + ConnectionStateChanged?.Invoke(this, new MtConnectionEventArgs(MtConnectionState.Failed, errorMessage)); + throw new Exception($"Connection to {host}:{port} failed. Error: {errorMessage}"); + } + + // Load quotes + Dictionary quotes = []; + foreach (var handle in experts) + { + var quote = GetQuote(client, handle); + if (quote != null) + quotes[handle] = quote; + } + lock (_locker) { _client = client; + _experts = experts; + _quotes = quotes; + if (_executorHandle == 0) + _executorHandle = (_experts.Count > 0) ? _experts.ElementAt(0) : 0; _connectionState = MtConnectionState.Connected; } - client.NotifyClientReady(); + + if (IsTesting()) + { + BacktestingReady(); + } + ConnectionStateChanged?.Invoke(this, new MtConnectionEventArgs(MtConnectionState.Connected, $"Connected to {host}:{port}")); + QuoteList?.Invoke(this, new(quotes.Values.ToList())); } catch (Exception e) { @@ -3102,7 +3130,11 @@ namespace MtApi private T? SendCommand(int expertHandle, MtCommandType commandType, object? payload = null) { - var client = Client; + return SendCommand(Client, expertHandle, commandType, payload); + } + + private T? SendCommand(MtRpcClient? client, int expertHandle, MtCommandType commandType, object? payload = null) + { if (client == null) { Log.Warn("SendCommand: No connection"); @@ -3143,11 +3175,6 @@ namespace MtApi Task.Run(() => _mtEventHandlers[(MtEventTypes)e.EventType](e.ExpertHandle, e.Payload)); } - private void Client_ExpertList(object? sender, MtExpertListEventArgs e) - { - Task.Run(() => ProcessExpertList(e.Experts)); - } - private void Client_ExpertAdded(object? sender, MtExpertEventArgs e) { Task.Run(() => ProcessExpertAdded(e.Expert)); @@ -3213,39 +3240,6 @@ namespace MtApi OnLockTicks?.Invoke(this, new MtLockTicksEventArgs(expertHandle, e.Instrument)); } - private void ProcessExpertList(HashSet experts) - { - if (experts == null || experts.Count == 0) - { - Log.Warn("ProcessExpertList: expert list invalid or empty"); - return; - } - - Dictionary quotes = []; - foreach (var handle in experts) - { - var quote = GetQuote(handle); - if (quote != null) - quotes[handle] = quote; - } - - lock (_locker) - { - _experts = experts; - _quotes = quotes; - if (_executorHandle == 0) - _executorHandle = (_experts.Count > 0) ? _experts.ElementAt(0) : 0; - } - _quotesWaiter.Set(); - - QuoteList?.Invoke(this, new(quotes.Values.ToList())); - - if (IsTesting()) - { - BacktestingReady(); - } - } - private void ProcessExpertAdded(int handle) { Log.Debug($"ProcessExpertAdded: {handle}"); @@ -3260,7 +3254,7 @@ namespace MtApi if (added) { - var quote = GetQuote(handle); + var quote = GetQuote(Client, handle); if (quote != null) { lock (_locker) @@ -3295,19 +3289,19 @@ namespace MtApi QuoteRemoved?.Invoke(this, new MtQuoteEventArgs(quote)); } - private MtQuote? GetQuote(int expertHandle) + private MtQuote? GetQuote(MtRpcClient? client, int expertHandle) { Log.Debug($"GetQuote: expertHandle = {expertHandle}"); - var e = SendCommand(expertHandle, MtCommandType.GetQuote); - if (e == null || string.IsNullOrEmpty(e.Instrument) || e.Tick == null) + var q = SendCommand(client, expertHandle, MtCommandType.GetQuote); + if (q == null || string.IsNullOrEmpty(q.Instrument) || q.Tick == null) return null; MtQuote quote = new() { - Instrument = e.Instrument, - Bid = e.Tick.Bid, - Ask = e.Tick.Ask, + Instrument = q.Instrument, + Bid = q.Tick.Bid, + Ask = q.Tick.Ask, ExpertHandle = expertHandle, }; diff --git a/MtApi5/MtApi5Client.cs b/MtApi5/MtApi5Client.cs index fae82223..95972560 100755 --- a/MtApi5/MtApi5Client.cs +++ b/MtApi5/MtApi5Client.cs @@ -31,7 +31,6 @@ namespace MtApi5 private HashSet _experts = []; private Dictionary _quotes = []; - private readonly EventWaitHandle _quotesWaiter = new AutoResetEvent(false); #endregion #region Public Methods @@ -94,17 +93,13 @@ namespace MtApi5 { if (_connectionState == Mt5ConnectionState.Connected || _connectionState == Mt5ConnectionState.Connecting) - { return; - } - _connectionState = Mt5ConnectionState.Connecting; } ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(Mt5ConnectionState.Connecting, $"Connecting to {host}:{port}")); var client = new MtRpcClient(host, port, new RpcClientLogger(Log)); - client.ExpertList += Client_ExpertList; client.ExpertAdded += Client_ExpertAdded; client.ExpertRemoved += Client_ExpertRemoved; client.MtEventReceived += Client_MtEventReceived; @@ -115,13 +110,42 @@ namespace MtApi5 { await client.Connect(); Log.Info($"Connected to {host}:{port}"); + + var experts = client.RequestExpertsList(); + if (experts == null || experts.Count == 0) + { + var errorMessage = "Failed to load expert list"; + Log.Error(errorMessage); + client.Disconnect(); + + ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(Mt5ConnectionState.Failed, errorMessage)); + throw new Exception($"Connection to {host}:{port} failed. Error: {errorMessage}"); + } + + // Load quotes + Dictionary quotes = []; + foreach (var handle in experts) + { + var quote = GetQuote(client, handle); + if (quote != null) + quotes[handle] = quote; + } + lock (_locker) { _client = client; + _experts = experts; + _quotes = quotes; + if (_executorHandle == 0) + _executorHandle = (_experts.Count > 0) ? _experts.ElementAt(0) : 0; _connectionState = Mt5ConnectionState.Connected; } - client.NotifyClientReady(); + + if (IsTesting()) + BacktestingReady(); + ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(Mt5ConnectionState.Connected, $"Connected to {host}:{port}")); + QuoteList?.Invoke(this, new(quotes.Values.ToList())); } catch (Exception e) { @@ -154,14 +178,10 @@ namespace MtApi5 } /// - ///Load quotes connected into MetaTrader API (deprecated) + ///Load quotes connected into MetaTrader API /// public IEnumerable GetQuotes() { - // this function is deprecated. - // should be used event OnQuoteList - - _quotesWaiter.WaitOne(10000); // wait 10 sec for loading all quotes from MetaTrader lock (_locker) { return _quotes.Values.ToList(); @@ -3345,11 +3365,6 @@ namespace MtApi5 Task.Run(() => _mtEventHandlers[(Mt5EventTypes)e.EventType](e.ExpertHandle, e.Payload)); } - private void Client_ExpertList(object? sender, MtExpertListEventArgs e) - { - Task.Run(()=>ProcessExpertList(e.Experts)); - } - private void Client_ExpertAdded(object? sender, MtExpertEventArgs e) { Task.Run(() => ProcessExpertAdded(e.Expert)); @@ -3360,39 +3375,6 @@ namespace MtApi5 Task.Run(() => ProcessExpertRemoved(e.Expert)); } - private void ProcessExpertList(HashSet experts) - { - if (experts == null || experts.Count == 0) - { - Log.Warn("ProcessExpertList: expert list invalid or empty"); - return; - } - - Dictionary quotes = []; - foreach (var handle in experts) - { - var quote = GetQuote(handle); - if (quote != null) - quotes[handle] = quote; - } - - lock (_locker) - { - _experts = experts; - _quotes = quotes; - if (_executorHandle == 0) - _executorHandle = (_experts.Count > 0) ? _experts.ElementAt(0) : 0; - } - _quotesWaiter.Set(); - - QuoteList?.Invoke(this, new(quotes.Values.ToList())); - - if (IsTesting()) - { - BacktestingReady(); - } - } - private void ProcessExpertAdded(int handle) { Log.Debug($"ProcessExpertAdded: {handle}"); @@ -3407,7 +3389,7 @@ namespace MtApi5 if (added) { - var quote = GetQuote(handle); + var quote = GetQuote(Client, handle); if (quote != null) { lock (_locker) @@ -3444,23 +3426,23 @@ namespace MtApi5 QuoteRemoved?.Invoke(this, new Mt5QuoteEventArgs(quote)); } - private Mt5Quote? GetQuote(int expertHandle) + private Mt5Quote? GetQuote(MtRpcClient? client, int expertHandle) { Log.Debug($"GetQuote: expertHandle = {expertHandle}"); - var e = SendCommand(expertHandle, Mt5CommandType.GetQuote); - if (e == null || string.IsNullOrEmpty(e.Instrument) || e.Tick == null) + var q = SendCommand(client, expertHandle, Mt5CommandType.GetQuote); + if (q == null || string.IsNullOrEmpty(q.Instrument) || q.Tick == null) return null; Mt5Quote quote = new() { - Instrument = e.Instrument, - Bid = e.Tick.Bid, - Ask = e.Tick.Ask, + Instrument = q.Instrument, + Bid = q.Tick.Bid, + Ask = q.Tick.Ask, ExpertHandle = expertHandle, - Volume = e.Tick.Volume, - Time = Mt5TimeConverter.ConvertFromMtTime(e.Tick.Time), - Last = e.Tick.Last + Volume = q.Tick.Volume, + Time = Mt5TimeConverter.ConvertFromMtTime(q.Tick.Time), + Last = q.Tick.Last }; return quote; @@ -3572,7 +3554,11 @@ namespace MtApi5 private T? SendCommand(int expertHandle, Mt5CommandType commandType, object? payload = null) { - var client = Client; + return SendCommand(Client, expertHandle, commandType, payload); + } + + private T? SendCommand(MtRpcClient? client, int expertHandle, Mt5CommandType commandType, object? payload = null) + { if (client == null) { Log.Warn("SendCommand: No connection"); diff --git a/MtClient/MtRpcClient.cs b/MtClient/MtRpcClient.cs index a83e5845..48d06dd0 100755 --- a/MtClient/MtRpcClient.cs +++ b/MtClient/MtRpcClient.cs @@ -1,5 +1,4 @@ -using System.Diagnostics; -using System.Net.WebSockets; +using System.Net.WebSockets; using System.Text; namespace MtClient @@ -68,7 +67,7 @@ namespace MtClient public string? SendCommand(int expertHandle, int commandType, string payload) { - CommandTask commandTask = new(); + CommandTask commandTask = new(); int commandId; lock (tasks_) { @@ -88,10 +87,24 @@ namespace MtClient return response; } - public void NotifyClientReady() + public HashSet? RequestExpertsList() { + CommandTask notificationTask = new(); + lock (notification_tasks_) + { + notification_tasks_[MtNotificationType.ClientReady] = notificationTask; + } + MtNotification notification = new(MtNotificationType.ClientReady); Send(notification); + + var response = notificationTask.WaitResponse(10000); // 10 sec + lock (notification_tasks_) + { + notification_tasks_.Remove(MtNotificationType.ClientReady); + } + + return response as HashSet; } private void Send(MtMessage message) @@ -118,7 +131,7 @@ namespace MtClient { sendWaiter_.WaitOne(); continue; - } + } try { @@ -166,30 +179,12 @@ namespace MtClient else if (result.MessageType == WebSocketMessageType.Text) { var msg = Encoding.UTF8.GetString(ms.ToArray(), 0, recvCount); - //var msg = Encoding.ASCII.GetString(recvBuffer, 0, result.Count); OnReceive(msg); } ms.Seek(0, SeekOrigin.Begin); ms.Position = 0; } } - - //byte[] recvBuffer = new byte[64 * 1024]; - //while (ws_.State == WebSocketState.Open) - //{ - // var result = await ws_.ReceiveAsync(new ArraySegment(recvBuffer), CancellationToken.None); - // if (result.MessageType == WebSocketMessageType.Close) - // { - // logger_.Info($"MtRpcClient.DoReceive: close signal {result.CloseStatusDescription}"); - // Disconnected?.Invoke(this, EventArgs.Empty); - // break; - // } - // else - // { - // var msg = Encoding.ASCII.GetString(recvBuffer, 0, result.Count); - // OnReceive(msg); - // } - //} } catch (Exception ex) { @@ -216,7 +211,7 @@ namespace MtClient { logger_.Warn("MtRpcClient.OnReceive: Invalid message format."); return; - } + } MessageType msgType; try @@ -250,7 +245,7 @@ namespace MtClient break; case MessageType.ExpertList: if (message is MtExpertListMsg expertListMsg) - ExpertList?.Invoke(this, new(expertListMsg.Experts)); + ProcessExpertList(expertListMsg.Experts); break; case MessageType.ExpertAdded: if (message is MtExpertAddedMsg expertAddedMsg) @@ -273,14 +268,24 @@ namespace MtClient lock (tasks_) { - if (tasks_.TryGetValue(commandId, out CommandTask? value)) + if (tasks_.TryGetValue(commandId, out CommandTask? value)) value.SetResponse(payload); } } + private void ProcessExpertList(HashSet experts) + { + logger_.Debug($"MtRpcClient.ProcessExpertList: experts count - {experts.Count}"); + + lock (notification_tasks_) + { + if (notification_tasks_.TryGetValue(MtNotificationType.ClientReady, out CommandTask? value)) + value.SetResponse(experts); + } + } + public event EventHandler? ConnectionFailed; public event EventHandler? Disconnected; - public event EventHandler? ExpertList; public event EventHandler? ExpertAdded; public event EventHandler? ExpertRemoved; public event EventHandler? MtEventReceived; @@ -288,7 +293,6 @@ namespace MtClient private readonly ClientWebSocket ws_ = new(); private readonly string host_; private readonly int port_; - private readonly byte[] buf_ = new byte[10000]; private readonly Queue pendingMessages_ = []; private readonly Thread receiveThread_; @@ -296,18 +300,19 @@ namespace MtClient private readonly EventWaitHandle sendWaiter_ = new AutoResetEvent(false); private int nextCommandId = 0; - private readonly Dictionary tasks_ = []; + private readonly Dictionary> tasks_ = []; + private readonly Dictionary> notification_tasks_ = []; private readonly IRpcLogger logger_; } - internal class CommandTask + internal class CommandTask { private readonly EventWaitHandle responseWaiter_ = new AutoResetEvent(false); - private string? response_; + private T? response_; private readonly object locker_ = new(); - public string? WaitResponse(int time) + public T? WaitResponse(int time) { responseWaiter_.WaitOne(time); lock (locker_) @@ -316,7 +321,7 @@ namespace MtClient } } - public void SetResponse(string result) + public void SetResponse(T result) { lock (locker_) {