Processing incoming messages in MtApi5

This commit is contained in:
Viacheslav Demydiuk
2024-01-12 18:22:26 +02:00
parent ecd09ad6a8
commit e6731d8828
+127 -87
View File
@@ -3,6 +3,9 @@ using MtApi5.Requests;
using Newtonsoft.Json; using Newtonsoft.Json;
using MtApi5.Events; using MtApi5.Events;
using MtClient; using MtClient;
using System.ComponentModel.Design;
using System.Reflection.Metadata;
using System.Collections.Generic;
namespace MtApi5 namespace MtApi5
{ {
@@ -3344,13 +3347,13 @@ namespace MtApi5
_connectionState = Mt5ConnectionState.Connecting; _connectionState = Mt5ConnectionState.Connecting;
} }
string message = $"Connecting to {host}:{port}"; string message = $"Connect: connecting to {host}:{port}";
ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(Mt5ConnectionState.Connecting, message)); ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(Mt5ConnectionState.Connecting, message));
var client = new MtRpcClient(host, port); var client = new MtRpcClient(host, port);
client.MessageReceived += _client_OnMessageReceived; client.MessageReceived += Client_OnMessageReceived;
client.ConnectionFailed += _client_OnConnectionFailed; client.ConnectionFailed += Client_OnConnectionFailed;
client.Disconnected += _client_Disconnected; client.Disconnected += Client_Disconnected;
var state = Mt5ConnectionState.Failed; var state = Mt5ConnectionState.Failed;
try try
@@ -3360,15 +3363,16 @@ namespace MtApi5
} }
catch (Exception e) catch (Exception e)
{ {
Log?.Warn($"Failed connection to {host}:{port}. {e.Message}"); Log?.Warn($"Connect: Failed connection to {host}:{port}. {e.Message}");
} }
Log?.Info($"Connect: connection to {host}:{port} is {state}");
lock (_locker) lock (_locker)
{ {
if (state == Mt5ConnectionState.Connected) if (state == Mt5ConnectionState.Connected)
{ {
_client = client; _client = client;
Log?.Info($"Connected to {host}:{port}");
} }
_connectionState = state; _connectionState = state;
@@ -3378,35 +3382,30 @@ namespace MtApi5
OnConnected(); OnConnected();
ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(state, message)); ConnectionStateChanged?.Invoke(this, new Mt5ConnectionEventArgs(state, message));
Log?.Info($"Connected finished");
} }
private void _client_OnMessageReceived(object? o, MtMessage msg) private void Client_OnMessageReceived(object? o, MtMessage msg)
{ {
Log?.Debug($"Message received: {msg}"); Log?.Debug($"Message received: type = {msg}");
Task.Run(() => switch (msg.MsgType)
{ {
switch (msg.MsgType) case MessageType.ExpertList:
{ Task.Run(() => ProcessExpertList(msg as MtExpertListMsg)); //must be runned on another thread to avoid blocking
case MessageType.ExpertList: break;
ProcessExpertList(msg as MtExpertListMsg); case MessageType.ExpertAdded:
break; Task.Run(() => ProcessExpertAdded(msg as MtExpertAddedMsg)); //must be runned on another thread to avoid blocking
case MessageType.ExpertAdded: break;
//ProcessExpertAdded(msg as MtExpertAddedMsg); case MessageType.ExpertRemoved:
break; ProcessExpertRemoved(msg as MtExpertRemovedMsg);
case MessageType.ExpertRemoved: break;
//ProcessExpertRemoved(msg as MtExpertRemovedMsg); case MessageType.Event:
break; ProcessEvent(msg as MtEvent);
case MessageType.Event: break;
//ProcessEvent(msg as MtEvent); case MessageType.Response:
break; ProcessResponse(msg as MtResponse);
case MessageType.Response: break;
ProcessResponse(msg as MtResponse); }
break;
}
});
} }
private void ProcessExpertList(MtExpertListMsg? msg) private void ProcessExpertList(MtExpertListMsg? msg)
@@ -3417,19 +3416,13 @@ namespace MtApi5
HashSet<int> experts = msg.Experts; HashSet<int> experts = msg.Experts;
if (experts == null || experts.Count == 0) if (experts == null || experts.Count == 0)
{ {
Console.WriteLine($"ProcessExpertList: expert list invalid"); Log?.Warn("ProcessExpertList: expert list invalid or empty");
return; return;
} }
lock(_locker)
{
_experts = experts;
}
Dictionary<int, Mt5Quote> quotes = []; Dictionary<int, Mt5Quote> quotes = [];
foreach (var handle in experts) foreach (var handle in experts)
{ {
Console.WriteLine($"ProcessExpertList: {handle}");
var quote = GetQuote(handle); var quote = GetQuote(handle);
if (quote != null) if (quote != null)
quotes[handle] = quote; quotes[handle] = quote;
@@ -3437,11 +3430,82 @@ namespace MtApi5
lock (_locker) lock (_locker)
{ {
_quotes= quotes; _experts = experts;
_quotes = quotes;
} }
_quotesWaiter.Set(); _quotesWaiter.Set();
QuoteList?.Invoke(this, _quotes.Values.ToList()); QuoteList?.Invoke(this, quotes.Values.ToList());
}
private void ProcessExpertAdded(MtExpertAddedMsg? msg)
{
if (msg == null)
return;
int handle = msg.ExpertHandle;
Log?.Debug($"ProcessExpertAdded: {handle}");
bool added;
lock (_locker)
{
added = _experts.Add(handle);
}
if (added)
{
var quote = GetQuote(handle);
if (quote != null)
{
lock (_locker)
{
_quotes[handle] = quote;
}
QuoteAdded?.Invoke(this, new Mt5QuoteEventArgs(quote));
}
else
Log?.Warn($"ProcessExpertAdded: failed to get quote for expert {handle}");
}
else
Log?.Warn($"ProcessExpertAdded: expert handle {handle} is already exist");
}
private void ProcessExpertRemoved(MtExpertRemovedMsg? msg)
{
if (msg == null)
return;
int handle = msg.ExpertHandle;
Log?.Debug($"ProcessExpertRemoved: {handle}");
Mt5Quote? quote = null;
lock (_locker)
{
_experts.Remove(handle);
if (_quotes.TryGetValue(handle, out quote))
_quotes.Remove(handle);
}
if (quote != null)
QuoteRemoved?.Invoke(this, new Mt5QuoteEventArgs(quote));
}
private void ProcessEvent(MtEvent? msg)
{
if (msg == null)
return;
var handle = msg.ExpertHandle;
var eventType = (Mt5EventTypes)msg.EventType;
var payload = msg.Payload;
Log?.Debug($"ProcessEvent: {handle}, {eventType}, {payload}");
_mtEventHandlers[eventType](handle, payload);
} }
private void ProcessResponse(MtResponse? msg) private void ProcessResponse(MtResponse? msg)
@@ -3461,29 +3525,27 @@ namespace MtApi5
value.SetResponse(payload); value.SetResponse(payload);
} }
} }
private Mt5Quote? GetQuote(int expertHandle) private Mt5Quote? GetQuote(int expertHandle)
{ {
var response = SendCommand<Mt5Quote>(expertHandle, Mt5CommandType.GetQuote); Log?.Debug($"GetQuote: expertHandle = {expertHandle}");
return response; var quote = SendCommand<Mt5Quote>(expertHandle, Mt5CommandType.GetQuote);
if (quote != null)
quote.ExpertHandle = expertHandle;
return quote;
} }
private void _client_OnConnectionFailed(object? sender, EventArgs e) private void Client_OnConnectionFailed(object? sender, EventArgs e)
{ {
Log?.Info("Received connection failed");
Disconnect(true); Disconnect(true);
} }
private void _client_Disconnected(object? sender, EventArgs e) private void Client_Disconnected(object? sender, EventArgs e)
{ {
Log?.Info("Received normal disconnection");
Disconnect(false); Disconnect(false);
} }
//private void _client_MtEventReceived(MtEvent e)
//{
// var eventType = (Mt5EventTypes)e.EventType;
// _mtEventHandlers[eventType](e.ExpertHandle, e.Payload);
//}
private void ReceivedOnTradeTransactionEvent(int expertHandler, string payload) private void ReceivedOnTradeTransactionEvent(int expertHandler, string payload)
{ {
var e = JsonConvert.DeserializeObject<OnTradeTransactionEvent>(payload); var e = JsonConvert.DeserializeObject<OnTradeTransactionEvent>(payload);
@@ -3515,6 +3577,9 @@ namespace MtApi5
var e = JsonConvert.DeserializeObject<OnTickEvent>(payload); var e = JsonConvert.DeserializeObject<OnTickEvent>(payload);
if (e == null || string.IsNullOrEmpty(e.Instrument) || e.Tick == null) if (e == null || string.IsNullOrEmpty(e.Instrument) || e.Tick == null)
return; return;
QuoteUpdated?.Invoke(this, e.Instrument, e.Tick.bid, e.Tick.ask);
var quote = new Mt5Quote(e.Instrument, e.Tick.bid, e.Tick.ask) var quote = new Mt5Quote(e.Instrument, e.Tick.bid, e.Tick.ask)
{ {
ExpertHandle = expertHandler, ExpertHandle = expertHandler,
@@ -3522,8 +3587,6 @@ namespace MtApi5
Time = e.Tick.time, Time = e.Tick.time,
Last = e.Tick.last Last = e.Tick.last
}; };
QuoteUpdated?.Invoke(this, quote.Instrument, quote.Bid, quote.Ask);
QuoteUpdate?.Invoke(this, new Mt5QuoteEventArgs(quote)); QuoteUpdate?.Invoke(this, new Mt5QuoteEventArgs(quote));
} }
@@ -3609,16 +3672,19 @@ namespace MtApi5
var payloadJson = JsonConvert.SerializeObject(payload); var payloadJson = JsonConvert.SerializeObject(payload);
MtCommand command = new(expertHandle, (int)commandType, _commandId++, payloadJson); MtCommand command = new(expertHandle, (int)commandType, _commandId++, payloadJson);
_tasks[command.CommandId] = new CommandTask(); CommandTask commandTask = new();
lock (_locker)
Log?.Debug($"SendCommand: {command.CommandId}");
client.Send(command);
var responseJson = _tasks[command.CommandId].WaitResponse(10000); // 10 sec
if (_tasks.Remove(command.CommandId) == false)
{ {
Log?.Warn($"SendCommand: task {command.CommandId} is not found in collection"); _tasks[command.CommandId] = commandTask;
return default; }
Log?.Debug($"SendCommand: sending {command.CommandId} ...");
client.Send(command);
var responseJson = commandTask.WaitResponse(10000); // 10 sec
lock(_locker)
{
_tasks.Remove(command.CommandId);
} }
Log?.Debug($"SendCommand: received response JSON [{responseJson}]"); Log?.Debug($"SendCommand: received response JSON [{responseJson}]");
@@ -3722,32 +3788,6 @@ namespace MtApi5
// QuoteUpdated?.Invoke(this, quote.Instrument, quote.Bid, quote.Ask); // QuoteUpdated?.Invoke(this, quote.Instrument, quote.Bid, quote.Ask);
//} //}
//private void _client_ServerDisconnected(object sender, EventArgs e)
//{
// Disconnect(false);
//}
//private void _client_ServerFailed(object sender, EventArgs e)
//{
// Disconnect(true);
//}
//private void _client_QuoteRemoved(MtQuote quote)
//{
// if (quote != null)
// {
// QuoteRemoved?.Invoke(this, new Mt5QuoteEventArgs(new Mt5Quote(quote)));
// }
//}
//private void _client_QuoteAdded(MtQuote quote)
//{
// if (quote != null)
// {
// QuoteAdded?.Invoke(this, new Mt5QuoteEventArgs(new Mt5Quote(quote)));
// }
//}
private void OnConnected() private void OnConnected()
{ {
Log?.Debug("OnConnected: begin"); Log?.Debug("OnConnected: begin");