Centaur_Quant_Architecture/MQL5/Include/Network/CSocketClient.mqh

//+------------------------------------------------------------------+
//|                                                     CSocketClient.mqh |
//|                 Centaur Quant Architecture — Network Module        |
//|                 Non-Blocking TCP/IP Socket Transport (SDP)         |
//+------------------------------------------------------------------+
//|  PURPOSE                                                          |
//|  Thin, timeout-safe wrapper around native MQL5 sockets for the    |
//|  Universal Communication Bridge. Sends SDP JSON frames (CRLF      |
//|  terminated) to the Python router and reads AI advisory frames    |
//|  without blocking the MT5 event loop beyond explicit timeouts.    |
//+------------------------------------------------------------------+
#property strict

#ifndef CSOCKETCLIENT_MQH
#define CSOCKETCLIENT_MQH

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
class CSocketClient
  {
private:
   int               m_socket;          // native socket handle; INVALID_HANDLE when closed
   string            m_recv_buffer;     // partial inbound bytes awaiting a complete frame
   uint              m_send_timeout_ms; // total time budget for one Send() operation

   //--- compile-time constants for the socket loops ---
   enum
     {
      MAX_CHUNK_SIZE  = 4096,             // bytes per SocketRead() call
      MAX_RECV_BUFFER = 1048576,          // 1 MB guard against unframed floods
      READ_SLICE_MS   = 100               // max wait per SocketRead() slice
     };

   //--- release the handle and reset all internal state ---
   void              Cleanup();

   //--- append raw bytes (as chars) to the receive buffer ---
   void              AppendBytes(const uchar &bytes[], const int count);

   //--- pull the next "\n"-terminated frame; strips a trailing "\r" ---
   bool              ExtractFrame(string &out_frame);

public:
                     CSocketClient();
                    ~CSocketClient();

   //--- establish TCP/IP connection to the Python router ---
   bool              Connect(const string host, const int port, const uint timeout_ms = 3000);

   //--- close the socket and reset state ---
   void              Disconnect();

   //--- live connection status via SocketIsConnected() ---
   bool              IsConnected();

   //--- send one SDP JSON frame (CRLF framing appended if missing) ---
   bool              Send(const string json_payload);

   //--- read one complete inbound frame within the timeout budget ---
   bool              Read(string &out_response, const uint timeout_ms = 1000);
  };
//+------------------------------------------------------------------+
//| Constructor — start fully disconnected with default send budget.  |
//+------------------------------------------------------------------+
CSocketClient::CSocketClient()
   : m_socket(INVALID_HANDLE),
     m_recv_buffer(""),
     m_send_timeout_ms(1000)
  {
  }
//+------------------------------------------------------------------+
//| Destructor — always release the native handle on destruction.     |
//+------------------------------------------------------------------+
CSocketClient::~CSocketClient()
  {
   Disconnect();
  }
//+------------------------------------------------------------------+
//| Cleanup — close the handle and reset state. Safe to call anytime. |
//+------------------------------------------------------------------+
void CSocketClient::Cleanup()
  {
   if(m_socket != INVALID_HANDLE)
      SocketClose(m_socket);
   m_socket = INVALID_HANDLE;
   m_recv_buffer = "";
  }
//+------------------------------------------------------------------+
//| Connect — resolve + connect with an explicit deadline.            |
//| SocketConnect() blocks at most timeout_ms; a final                |
//| SocketIsConnected() sanity check guards against half-open links.  |
//+------------------------------------------------------------------+
bool CSocketClient::Connect(const string host, const int port, const uint timeout_ms)
  {
   Disconnect();   // always start from a clean state
   if(StringLen(host) == 0)
     {
      PrintFormat("[CSocketClient] ERROR: empty host. Connection aborted.");
      return false;
     }
   if(port <= 0 || port > 65535)
     {
      PrintFormat("[CSocketClient] ERROR: invalid port %d (1-65535). Connection aborted.", port);
      return false;
     }
   m_socket = SocketCreate();
   if(m_socket == INVALID_HANDLE || m_socket == 0)
     {
      PrintFormat("[CSocketClient] ERROR: SocketCreate() failed. GetLastError=%d.", GetLastError());
      m_socket = INVALID_HANDLE;
      return false;
     }
   ResetLastError();
   if(!SocketConnect(m_socket, host, port, timeout_ms))
     {
      PrintFormat("[CSocketClient] ERROR: SocketConnect(%s:%d) failed within %u ms. GetLastError=%d.",
                  host, port, timeout_ms, GetLastError());
      Cleanup();
      return false;
     }
   if(!SocketIsConnected(m_socket))
     {
      PrintFormat("[CSocketClient] ERROR: socket not reported connected after SocketConnect(%s:%d).",
                  host, port);
      Cleanup();
      return false;
     }
   PrintFormat("[CSocketClient] INFO: connected to %s:%d (socket handle %d).", host, port, m_socket);
   return true;
  }
//+------------------------------------------------------------------+
//| Disconnect — close an active link and reset buffered state.       |
//+------------------------------------------------------------------+
void CSocketClient::Disconnect()
  {
   if(m_socket != INVALID_HANDLE)
     {
      if(SocketIsConnected(m_socket))
         PrintFormat("[CSocketClient] INFO: closing active connection (socket %d).", m_socket);
      Cleanup();
     }
  }
//+------------------------------------------------------------------+
//| IsConnected — live status; the native call is the source of truth.|
//+------------------------------------------------------------------+
bool CSocketClient::IsConnected()
  {
   return (m_socket != INVALID_HANDLE && SocketIsConnected(m_socket));
  }
//+------------------------------------------------------------------+
//| Send — UTF-8 encode one SDP frame and push it over the socket.    |
//| CRLF framing is appended when missing. Partial sends are looped   |
//| with a bounded total budget (m_send_timeout_ms); any failure      |
//| triggers full internal cleanup.                                   |
//+------------------------------------------------------------------+
bool CSocketClient::Send(const string json_payload)
  {
   if(StringLen(json_payload) == 0)
     {
      PrintFormat("[CSocketClient] ERROR: empty payload. Send aborted.");
      return false;
     }
   if(!IsConnected())
     {
      PrintFormat("[CSocketClient] ERROR: Send() called while not connected. Payload dropped.");
      return false;
     }
//--- CRLF frame termination (receiver splits on "\n", tolerates "\r") ---
   string frame = json_payload;
   const int flen = StringLen(frame);
   if(StringSubstr(frame, flen - 2, 2) != "\r\n")
     {
      if(StringSubstr(frame, flen - 1, 1) == "\n")
         frame = StringSubstr(frame, 0, flen - 1) + "\r\n";
      else
         frame += "\r\n";
     }
//--- UTF-8 encode; the returned count includes the trailing '\0' ---
   uchar buffer[];
   const int total = StringToCharArray(frame, buffer, 0, WHOLE_ARRAY, CP_UTF8) - 1;
   if(total <= 0)
     {
      PrintFormat("[CSocketClient] ERROR: payload encoding produced 0 bytes. Send aborted.");
      return false;
     }
//--- bounded partial-send loop; total wait never exceeds the budget ---
   int offset = 0;
   const ulong deadline = GetTickCount64() + m_send_timeout_ms;
   while(offset < total)
     {
      if(GetTickCount64() > deadline)
        {
         PrintFormat("[CSocketClient] ERROR: Send() timed out after %u ms (%d/%d bytes sent).",
                     m_send_timeout_ms, offset, total);
         Cleanup();
         return false;
        }
      if(!IsConnected())
        {
         PrintFormat("[CSocketClient] ERROR: connection lost during Send(). Payload dropped.");
         Cleanup();
         return false;
        }
      //--- transmit only the unsent tail ---
      uchar tail[];
      ArrayCopy(tail, buffer, 0, offset, total - offset);
      ResetLastError();
      const int sent = SocketSend(m_socket, tail, (uint)ArraySize(tail));
      if(sent < 0)
        {
         PrintFormat("[CSocketClient] ERROR: SocketSend() failed. GetLastError=%d. Connection closed.",
                     GetLastError());
         Cleanup();
         return false;
        }
      if(sent == 0)
        {
         PrintFormat("[CSocketClient] ERROR: SocketSend() returned 0 — peer closed the connection.");
         Cleanup();
         return false;
        }
      offset += sent;
      Sleep(1);   // keep the event loop breathing between slices
     }
   return true;
  }
//+------------------------------------------------------------------+
//| Read — return the next complete "\n"-framed payload.              |
//| Serves frames already buffered first, then polls the socket in    |
//| small slices; the whole wait never exceeds timeout_ms. Returns    |
//| false (with out_response="") when no complete frame arrived.      |
//+------------------------------------------------------------------+
bool CSocketClient::Read(string &out_response, const uint timeout_ms)
  {
   out_response = "";
   if(!IsConnected())
     {
      PrintFormat("[CSocketClient] ERROR: Read() called while not connected.");
      return false;
     }
//--- 1) serve complete frames already buffered from previous reads ---
   if(ExtractFrame(out_response))
      return true;
//--- 2) poll the socket for up to timeout_ms ---
   const ulong deadline = GetTickCount64() + timeout_ms;
   uchar chunk[MAX_CHUNK_SIZE];
   while(true)
     {
      const ulong now = GetTickCount64();
      if(now >= deadline)
         break;
      if(!IsConnected())
        {
         PrintFormat("[CSocketClient] ERROR: connection lost during Read().");
         Cleanup();
         return false;
        }
      //--- guard: an unframed flood must not grow the buffer unbounded ---
      if(StringLen(m_recv_buffer) > MAX_RECV_BUFFER)
        {
         PrintFormat("[CSocketClient] WARNING: receive buffer exceeded %d bytes without a frame terminator. Buffer cleared.",
                     MAX_RECV_BUFFER);
         m_recv_buffer = "";
        }
      const uint slice = (uint)MathMin((ulong)READ_SLICE_MS, deadline - now);
      ResetLastError();
      const int got = SocketRead(m_socket, chunk, MAX_CHUNK_SIZE, slice);
      if(got > 0)
        {
         AppendBytes(chunk, got);
         if(ExtractFrame(out_response))
            return true;
        }
      else
         if(got < 0)
           {
            PrintFormat("[CSocketClient] ERROR: SocketRead() failed. GetLastError=%d. Connection closed.",
                        GetLastError());
            Cleanup();
            return false;
           }
         else
           {
            // got == 0: nothing arrived in this slice — detect a silent peer close
            if(!SocketIsConnected(m_socket))
              {
               PrintFormat("[CSocketClient] WARNING: peer closed the connection (read returned 0).");
               Cleanup();
               return false;
              }
           }
      Sleep(1);
     }
   return false;   // timeout: no complete frame available
  }
//+------------------------------------------------------------------+
//| AppendBytes — copy raw bytes into the receive buffer as chars.    |
//| SDP frames are ASCII/UTF-8 JSON, so byte-to-char is lossless for  |
//| the control characters and separators we scan for.                |
//+------------------------------------------------------------------+
void CSocketClient::AppendBytes(const uchar &bytes[], const int count)
  {
   string chunk;
   StringInit(chunk, count, 0);
   for(int i = 0; i < count; i++)
      StringSetCharacter(chunk, i, (ushort)bytes[i]);
   m_recv_buffer += chunk;
  }
//+------------------------------------------------------------------+
//| ExtractFrame — pull the next "\n"-terminated frame out of the     |
//| receive buffer; strips a trailing "\r" (CRLF tolerance).          |
//+------------------------------------------------------------------+
bool CSocketClient::ExtractFrame(string &out_frame)
  {
   const int pos = StringFind(m_recv_buffer, "\n");
   if(pos < 0)
      return false;
   out_frame = StringSubstr(m_recv_buffer, 0, pos);
   const int olen = StringLen(out_frame);
   if(olen > 0 && StringSubstr(out_frame, olen - 1, 1) == "\r")
      out_frame = StringSubstr(out_frame, 0, olen - 1);
//--- consume the frame including its terminator ---
   m_recv_buffer = StringSubstr(m_recv_buffer, pos + 1);
   return true;
  }

#endif // CSOCKETCLIENT_MQH