//+------------------------------------------------------------------+ //| 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