MT5RemoteBridgeAPI.mq5

//+------------------------------------------------------------------+
//|                  MT5RemoteBridgeAPI.mq5                          |
//+------------------------------------------------------------------+
#property strict
#property copyright "ding9736"
#property version   "3.0"
#property description "High-performance, fully automated encrypted, client-driven MT5RemoteBridgeAPI service framework (production-grade)"
#property link      "https://github.com/ding9736/MT5BridgeAPI "
#property service

//+------------------------------------------------------------------+
//| Core dependency imports                                          |
//+------------------------------------------------------------------+
#include "Lib/ZeroMQ/ZeroMQ.mqh"
#include "Core/Config.mqh"
#include "Core/ServiceState.mqh"
#include "Core/CommandProcessor.mqh"
#include "Core/HandshakeHandler.mqh"
#include "Core/Logger.mqh"

//+------------------------------------------------------------------+
//| Global service instances                                         |
//+------------------------------------------------------------------+
CServiceConfig g_Config;
CServiceState  g_State;

//+------------------------------------------------------------------+
//| Service start entry point                                        |
//+------------------------------------------------------------------+
void OnStart()
{
   g_Config.LoadDefaults();
   CLogger::Initialize(g_Config.ConfigPath, g_Config.LogFilename, g_Config.LogLevel, g_Config.LogToJournal);
   CLogger::LogInfo("======================================");
   CLogger::LogInfo("MT5RemoteBridgeAPI service is starting...");
   CLogger::LogInfo("======================================");
   if(!InitializeService())
   {
      CLogger::LogError("Service initialization failed, stopping now.");
      CLogger::Shutdown();
      return;
   }
   RunServiceLoop();
   CleanupService();
   CLogger::LogInfo("ZMQ service has been completely stopped.");
   CLogger::Shutdown();
}

//+------------------------------------------------------------------+
//| Initialize service                                               |
//+------------------------------------------------------------------+
bool InitializeService()
{
// Load configuration from JSON file (will override items with the same name loaded by LoadDefaults)
   g_Config.Load();
   if(!g_Config.Validate())
   {
      CLogger::LogError("Configuration validation failed: " + g_Config.LastError + " Please check the configuration file.");
      return false;
   }
   if(!g_State.Initialize(g_Config))
   {
      CLogger::LogError("Service state initialization failed.");
      return false;
   }
   g_State.SetConfigReloadCallback(ReloadConfiguration);
   CLogger::LogInfo(StringFormat("Service core components initialized successfully. Default trading parameters - MagicNumber:%d, Slippage:%d",
                                 g_Config.MagicNumber,
                                 g_Config.Slippage));
   return true;
}

//+------------------------------------------------------------------+
//| Service main loop                                                |
//+------------------------------------------------------------------+
void RunServiceLoop()
{
   CLogger::LogInfo("Entering service main loop...");
   if(!ValidateSockets())
   {
      CLogger::LogError("Socket validation failed, cannot enter main loop.");
      return;
   }
   ZmqPoller poller;
   if(!SetupPoller(poller))
   {
      CLogger::LogError("Poller setup failed.");
      return;
   }
   int poll_timeout_ms = g_Config.TimerMs > 0 ? g_Config.TimerMs : 10;
   while(!IsStopped())
   {
      if(poller.poll(poll_timeout_ms))
      {
         int processed_count = 0;
         if(ProcessHandshakeMessages() > 0)
         {
            processed_count++;
         }
         if(ProcessCommandMessages() > 0)
         {
            processed_count++;
         }
         if(processed_count > 0)
         {
            g_State.SignalActivity();
         }
      }
      g_State.ProcessAuthTasks();
      PerformPeriodicTasks();
      if(g_State.ShouldLogPerformanceStats())
      {
         ulong current_total_messages = g_State.GetTotalMessagesProcessed();
         ulong last_message_count = g_State.GetLastStatMessageCount();
         ulong messages_in_period = (current_total_messages > last_message_count) ? (current_total_messages - last_message_count) : 0;
         LogPerformanceStatistics(messages_in_period);
         g_State.ResetPerformanceStats(current_total_messages);
      }
   }
   CLogger::LogInfo("Service main loop has exited.");
}

//+------------------------------------------------------------------+
//| Validate sockets                                                 |
//+------------------------------------------------------------------+
bool ValidateSockets()
{
   if(CheckPointer(g_State.pHandshakeSocket) != POINTER_DYNAMIC || !g_State.pHandshakeSocket.isValid())
   {
      CLogger::LogError("Handshake socket is invalid.");
      return false;
   }
   if(CheckPointer(g_State.pCmdSocket) != POINTER_DYNAMIC || !g_State.pCmdSocket.isValid())
   {
      CLogger::LogError("Command socket is invalid.");
      return false;
   }
   if(CheckPointer(g_State.pPubSocket) != POINTER_DYNAMIC || !g_State.pPubSocket.isValid())
   {
      CLogger::LogError("Publish socket is invalid.");
      return false;
   }
   return true;
}

//+------------------------------------------------------------------+
//| Setup poller                                                     |
//+------------------------------------------------------------------+
bool SetupPoller(ZmqPoller &poller)
{
   if(!poller.add(*g_State.pHandshakeSocket, ZMQ_POLLIN))
   {
      CLogger::LogError("Failed to add handshake socket to poller.");
      return false;
   }
   if(!poller.add(*g_State.pCmdSocket, ZMQ_POLLIN))
   {
      CLogger::LogError("Failed to add command socket to poller.");
      return false;
   }
   CLogger::LogInfo("Poller setup successful.");
   return true;
}

//+------------------------------------------------------------------+
//| Process handshake messages                                       |
//+------------------------------------------------------------------+
int ProcessHandshakeMessages()
{
   ZmqMsg request;
   if(g_State.pHandshakeSocket.recv(request, ZMQ_FLAG_DONTWAIT))
   {
      g_State.IncrementTotalMessagesProcessed();
      string request_data = request.getData();
      if(StringLen(request_data) > 0)
      {
         CHandshakeHandler::Process(request_data, g_State);
         return 1;
      }
   }
   return 0;
}

//+------------------------------------------------------------------+
//| Process command messages                                         |
//+------------------------------------------------------------------+
int ProcessCommandMessages()
{
   ZmqMsg request;
   if(g_State.ReceiveCommand(request))
   {
      string request_data = request.getData();
      if(StringLen(request_data) > 0)
      {
         CCommandProcessor::Process(request_data, g_State);
         return 1;
      }
   }
   return 0;
}

//+------------------------------------------------------------------+
//| Perform periodic tasks                                           |
//+------------------------------------------------------------------+
void PerformPeriodicTasks()
{
   static ulong last_update_time_ms = 0;
   ulong current_time_ms = GetTickCount64();
   if (current_time_ms - last_update_time_ms < 1000) return;
   if(g_State.IsHealthy())
   {
      g_State.PublishUpdates();
      last_update_time_ms = current_time_ms;
   }
}

//+------------------------------------------------------------------+
//| Output performance statistics                                    |
//+------------------------------------------------------------------+
void LogPerformanceStatistics(ulong message_count)
{
   double elapsed_sec = (double)STATS_LOG_INTERVAL;
   double msg_per_sec = (message_count > 0) ? (double)message_count / elapsed_sec : 0;
   CLogger::LogInfo(StringFormat("Performance statistics: Messages processed=%d, Period=%.1f sec, Throughput=%.0f msg/s",
                                 message_count, elapsed_sec, msg_per_sec));
}

//+------------------------------------------------------------------+
//| Configuration reload callback                                    |
//+------------------------------------------------------------------+
void ReloadConfiguration()
{
   CLogger::LogInfo("Reloading configuration...");
   g_Config.Load();
   CLogger::LogWarning("Note: Only some configurations (like default MagicNumber/Slippage) can be hot-reloaded. Core settings like ports/keys require a service restart to take effect.");
   CLogger::LogInfo("Configuration reload complete.");
}

//+------------------------------------------------------------------+
//| Cleanup service                                                  |
//+------------------------------------------------------------------+
void CleanupService()
{
   CLogger::LogInfo("Cleaning up service resources...");
   g_State.Shutdown();
   CLogger::LogInfo("Service resource cleanup complete.");
}