Designing a Multi-EA Communication Bus Using Named Pipes in MQL5

Introduction

If you run multiple Expert Advisors in one terminal, they do not share state by default. Each keeps its own view of the account, and none can see what the others have already worked out. MetaTrader effectively shares only the GlobalVariables namespace. It becomes hard to manage quickly: values are untyped (double), keys are strings with no schema, and stale values are indistinguishable from current ones.

Named pipes solve a narrower, better-defined problem. A named pipe is an OS-level channel that two processes use to exchange bytes, with message framing built in. Instead of encoding meaning into a key, the meaning lives in code: a typed message with an explicit type field and a fixed layout both ends agree on.

This article builds a message bus on that idea. A broker EA owns a named pipe server and a message registry. Three slave EAs connect as clients, each reporting its own position state and receiving back a centrally computed portfolio risk figure. A dashboard shows every sender's live status, and a verification script proves the core logic before any of it touches a real pipe. Ten files make up the system: the message type enum, the typed message and its serialization, the server and client wrappers, the registry, the risk aggregator, the dashboard, the two EAs, and the test script.

The architecture of the MQL5 multi-EA system, showing how the Broker EA coordinates core services and communicates with multiple Slave EAs through a named-pipe message bus. The Broker EA displays the state of the Slave EAs on a chart dashboard.

Section 1 — How Named Pipes Work in MQL5

A named pipe is addressed by a path such as \\.\pipe\SomeName. One process creates it as a server and waits for a connection; another opens the same name as a client. The dot stands for the local machine, which is the only host this design targets.

MQL5 has no native pipe API, so every call comes from kernel32.dll through #import. The server side needs CreateNamedPipeW(), ConnectNamedPipe(), ReadFile(), WriteFile(), SetNamedPipeHandleState(), PeekNamedPipe(), FlushFileBuffers(), DisconnectNamedPipe(), CloseHandle(), and GetLastError(). The client side only needs CreateFileW() plus the same read, write, and close calls.

The pipe runs in message mode on both ends, so one WriteFile() call is always read back by exactly one ReadFile() call, with no risk of messages splitting or merging. Accepting a new client is a different matter, though: a blocking ConnectNamedPipe() waits forever if nobody connects, and a broker driving that call from a timer with zero slaves running would simply stop responding. This design temporarily switches the pipe to non-blocking mode for a single connection attempt. It then restores blocking mode and confirms the connection state with PeekNamedPipe(). Section 3 explains why.

Section 2 — The Message Vocabulary: MessageType.mqh and Message.mqh

Every message on the bus carries an explicit type from a plain enum, rather than a bare integer with a meaning only the original author remembers.

//+------------------------------------------------------------------+
//|                                                  MessageType.mqh |
//+------------------------------------------------------------------+
#ifndef MESSAGETYPE_MQH
#define MESSAGETYPE_MQH

//+------------------------------------------------------------------+
//| ENUM_MESSAGE_TYPE                                                |
//| The complete set of typed message categories that may travel     |
//| across the message bus. Every CMessage carries exactly one of    |
//| these in its message_type field, so both ends agree on meaning.  |
//+------------------------------------------------------------------+
enum ENUM_MESSAGE_TYPE
  {
   MSG_POSITION_UPDATE = 1,  // slave reports its local position state
   MSG_RISK_QUERY      = 2,  // slave explicitly asks for aggregate risk
   MSG_RISK_RESPONSE   = 3,  // broker returns the aggregate risk figure
   MSG_HEARTBEAT       = 4,  // liveness ping with no payload meaning
   MSG_ACK             = 5   // generic acknowledgment of receipt
  };

#endif // MESSAGETYPE_MQH
//+------------------------------------------------------------------+

The CMessage struct carries the type, the sender's magic number, the instrument as a fixed 12-byte array (a dynamic string would break the fixed-size layout), two general payload fields whose meaning depends on the type, a per-sender sequence number, and a build timestamp.

//+------------------------------------------------------------------+
//|                                                      Message.mqh |
//+------------------------------------------------------------------+
#ifndef MESSAGE_MQH
#define MESSAGE_MQH

#include "MessageType.mqh"

#define MESSAGE_BYTE_SIZE 52
#define MESSAGE_SYMBOL_LEN 12

//--- raw byte copy used only for the exact double serialization
#import "kernel32.dll"
void RtlMoveMemory(uchar &dst[], const double &src, int len);
void RtlMoveMemory(double &dst, const uchar &src[], int len);
#import

//+------------------------------------------------------------------+
//| CMessage                                                         |
//| One typed message on the bus. Holds the message type, the        |
//| sending EA's magic number, the instrument as a fixed byte array, |
//| two general payload fields, a per-sender sequence id, and a      |
//| build timestamp. Serializes to a fixed 52-byte layout.           |
//+------------------------------------------------------------------+
struct CMessage
  {
   ENUM_MESSAGE_TYPE message_type;                   // what this message is
   ulong             sender_magic;                   // magic of the sending EA
   uchar             symbol[MESSAGE_SYMBOL_LEN];     // instrument, null padded
   double            payload_double;                 // type-specific double
   int               payload_int;                    // type-specific int
   ulong             sequence_id;                    // per-sender counter
   datetime          timestamp;                      // when the message was built

   void              SetSymbol(const string sym);
   string            GetSymbol(void) const;
   void              Serialize(uchar &buf[]) const;
   bool              Deserialize(const uchar &buf[]);
  };

SetSymbol() and GetSymbol() move an instrument name between a string and the fixed byte array, clearing the field first so trailing bytes are always zero.

//+------------------------------------------------------------------+
//| SetSymbol                                                        |
//+------------------------------------------------------------------+
void CMessage::SetSymbol(const string sym)
  {
//--- clears the fixed field so trailing bytes are always zero
   for(int i = 0; i < MESSAGE_SYMBOL_LEN; i++)
      symbol[i] = 0;
//--- converts the string to bytes and copies what fits, leaving a null
   uchar tmp[];
   int   n = ::StringToCharArray(sym, tmp);
   int   copy = ::MathMin(n, MESSAGE_SYMBOL_LEN - 1);
   for(int i = 0; i < copy; i++)
      symbol[i] = tmp[i];
  }

//+------------------------------------------------------------------+
//| GetSymbol                                                        |
//+------------------------------------------------------------------+
string CMessage::GetSymbol(void) const
  {
//--- read bytes until the null terminator or the field end
   string result = "";
   for(int i = 0; i < MESSAGE_SYMBOL_LEN; i++)
     {
      if(symbol[i] == 0)
         break;
      result += ::CharToString(symbol[i]);
     }
   return(result);
  }

Serialize() packs every field into a fixed 52-byte little-endian layout. Integers are byte-shifted; the one field that resists that treatment is the double, so RtlMoveMemory() copies its eight raw bytes bit-exact instead.

//+------------------------------------------------------------------+
//| Serialize                                                        |
//+------------------------------------------------------------------+
void CMessage::Serialize(uchar &buf[]) const
  {
//--- size and zero the output buffer
   ::ArrayResize(buf, MESSAGE_BYTE_SIZE);
   ::ArrayInitialize(buf, 0);
   int off = 0;
//--- message_type as a four-byte little-endian integer
   int type_value = (int)message_type;
   for(int i = 0; i < 4; i++)
      buf[off + i] = (uchar)((type_value >> (i * 8)) & 0xFF);
   off += 4;
//--- sender_magic as an eight-byte little-endian integer
   for(int i = 0; i < 8; i++)
      buf[off + i] = (uchar)((sender_magic >> (i * 8)) & 0xFF);
   off += 8;
//--- symbol as a raw twelve-byte block
   for(int i = 0; i < MESSAGE_SYMBOL_LEN; i++)
      buf[off + i] = symbol[i];
   off += MESSAGE_SYMBOL_LEN;
//--- payload_double as bit-exact bytes via RtlMoveMemory
   uchar dbytes[8];
   RtlMoveMemory(dbytes, payload_double, 8);
   for(int i = 0; i < 8; i++)
      buf[off + i] = dbytes[i];
   off += 8;
//--- payload_int as a four-byte little-endian integer
   for(int i = 0; i < 4; i++)
      buf[off + i] = (uchar)((payload_int >> (i * 8)) & 0xFF);
   off += 4;
//--- sequence_id as an eight-byte little-endian integer
   for(int i = 0; i < 8; i++)
      buf[off + i] = (uchar)((sequence_id >> (i * 8)) & 0xFF);
   off += 8;
//--- timestamp as an eight-byte little-endian integer
   long ts_value = (long)timestamp;
   for(int i = 0; i < 8; i++)
      buf[off + i] = (uchar)((ts_value >> (i * 8)) & 0xFF);
   off += 8;
  }

Deserialize() reverses the process field by field, and rejects a buffer that is too short to hold a full message, so a partial read never corrupts the struct.

//+------------------------------------------------------------------+
//| Deserialize                                                      |
//+------------------------------------------------------------------+
bool CMessage::Deserialize(const uchar &buf[])
  {
//--- rejects a buffer that is too short to hold a full message
   if(::ArraySize(buf) < MESSAGE_BYTE_SIZE)
      return(false);
   int off = 0;
//--- message_type from four little-endian bytes
   int type_value = 0;
   for(int i = 0; i < 4; i++)
      type_value |= ((int)buf[off + i]) << (i * 8);
   message_type = (ENUM_MESSAGE_TYPE)type_value;
   off += 4;
//--- sender_magic from eight little-endian bytes
   sender_magic = 0;
   for(int i = 0; i < 8; i++)
      sender_magic |= ((ulong)buf[off + i]) << (i * 8);
   off += 8;
//--- symbol from the raw twelve-byte block
   for(int i = 0; i < MESSAGE_SYMBOL_LEN; i++)
      symbol[i] = buf[off + i];
   off += MESSAGE_SYMBOL_LEN;
//--- payload_double from bit-exact bytes via RtlMoveMemory
   uchar dbytes[8];
   for(int i = 0; i < 8; i++)
      dbytes[i] = buf[off + i];
   double recovered = 0.0;
   RtlMoveMemory(recovered, dbytes, 8);
   payload_double = recovered;
   off += 8;
//--- payload_int from four little-endian bytes
   payload_int = 0;
   for(int i = 0; i < 4; i++)
      payload_int |= ((int)buf[off + i]) << (i * 8);
   off += 4;
//--- sequence_id from eight little-endian bytes
   sequence_id = 0;
   for(int i = 0; i < 8; i++)
      sequence_id |= ((ulong)buf[off + i]) << (i * 8);
   off += 8;
//--- timestamp from eight little-endian bytes
   long ts_value = 0;
   for(int i = 0; i < 8; i++)
      ts_value |= ((long)buf[off + i]) << (i * 8);
   timestamp = (datetime)ts_value;
   off += 8;
   return(true);
  }

Section 3 — CPipeServer: the Broker's Pipe Endpoint

Handles are declared as long, since a HANDLE is pointer-sized on 64-bit terminals. The import block sticks to scalars, scalar references, and byte arrays throughout; a custom struct passed by reference does not marshal reliably through MQL5's DLL-import layer.

//+------------------------------------------------------------------+
//|                                                   PipeServer.mqh |
//+------------------------------------------------------------------+
#ifndef PIPESERVER_MQH
#define PIPESERVER_MQH

#include "Message.mqh"

//--- named pipe creation and access constants
#define PIPE_ACCESS_DUPLEX        0x00000003
#define PIPE_TYPE_MESSAGE         0x00000004
#define PIPE_READMODE_MESSAGE     0x00000002
#define PIPE_WAIT                 0x00000000
#define PIPE_NOWAIT               0x00000001
#define PIPE_UNLIMITED_INSTANCES  255
#define PIPE_INVALID_HANDLE       (-1)
#define PIPE_OUT_BUFFER           4096
#define PIPE_IN_BUFFER            4096

//--- Windows error codes used by the accept poll below
#define ERR_PIPE_CONNECTED   535   // a client got there first, already attached
#define ERR_PIPE_LISTENING   536   // in non-blocking mode: nobody trying to connect yet
#define ERR_PIPE_BUSY        231   // seen transiently on some systems instead of 536

//--- Windows named pipe server API. Every function here uses only
//--- primitive scalar / array parameters -- the same kind the
//--- original, demonstrably working implementation used -- rather
//--- than a custom struct passed by reference. Custom-struct
//--- marshaling (as required for true overlapped/asynchronous I/O)
//--- proved unreliable when tested against a real MT5 terminal, so
//--- this version deliberately avoids it.
#import "kernel32.dll"
long CreateNamedPipeW(string name, uint open_mode, uint pipe_mode, uint max_instances,
                      uint out_buffer, uint in_buffer, uint default_timeout, long security);
int  SetNamedPipeHandleState(long handle, uint &mode, long max_collection_count, long collect_data_timeout);
int  ConnectNamedPipe(long handle, long overlapped);
int  PeekNamedPipe(long handle, uchar &buffer[], uint buffer_size, uint &bytes_read,
                   uint &total_bytes_avail, uint &bytes_left_this_message);
int  ReadFile(long handle, uchar &buffer[], uint bytes_to_read, uint &bytes_read, long overlapped);
int  WriteFile(long handle, uchar &buffer[], uint bytes_to_write, uint &bytes_written, long overlapped);
int  FlushFileBuffers(long handle);
int  DisconnectNamedPipe(long handle);
int  CloseHandle(long handle);
uint GetLastError(void);
#import

//+------------------------------------------------------------------+
//| CPipeServer                                                      |
//| Wraps the Windows named pipe server API for the broker EA.       |
//+------------------------------------------------------------------+
class CPipeServer
  {
private:
   string            m_pipe_name;
   long              m_handle;
   bool              m_connected;
   uint              m_last_error;

public:
                     CPipeServer(void);
                    ~CPipeServer(void);

   bool              Create(const string pipe_name);
   bool              TryAcceptClient(void);
   bool              ReadMessage(CMessage &msg);
   bool              WriteMessage(const CMessage &msg);
   void              Flush(void);
   void              Disconnect(void);
   void              Close(void);
   bool              IsConnected(void) const { return(m_connected); }
   uint              LastError(void)   const { return(m_last_error); }
  };

//+------------------------------------------------------------------+
//| Constructor                                                      |
//+------------------------------------------------------------------+
CPipeServer::CPipeServer(void)
  {
   m_pipe_name  = "";
   m_handle     = PIPE_INVALID_HANDLE;
   m_connected  = false;
   m_last_error = 0;
  }

//+------------------------------------------------------------------+
//| Destructor                                                       |
//+------------------------------------------------------------------+
CPipeServer::~CPipeServer(void)
  {
   Close();
  }

Create() builds the endpoint in ordinary blocking PIPE_WAIT mode — the same mode a slave expects when it opens the pipe as a client.

//+------------------------------------------------------------------+
//| Create                                                           |
//+------------------------------------------------------------------+
bool CPipeServer::Create(const string pipe_name)
  {
   m_pipe_name = pipe_name;
   uint open_mode = PIPE_ACCESS_DUPLEX;
   uint pipe_mode = PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT;
   m_handle = ::CreateNamedPipeW(m_pipe_name, open_mode, pipe_mode, PIPE_UNLIMITED_INSTANCES,
                                 PIPE_OUT_BUFFER, PIPE_IN_BUFFER, 0, 0);
   if(m_handle == PIPE_INVALID_HANDLE)
     {
      m_last_error = ::GetLastError();
      ::PrintFormat("CPipeServer: CreateNamedPipeW failed on '%s', error %d", m_pipe_name, m_last_error);
      return(false);
     }
   m_connected = false;
   return(true);
  }

TryAcceptClient() solves the blocking-accept problem. It flips the handle to PIPE_NOWAIT for one ConnectNamedPipe() attempt, then flips it straight back to PIPE_WAIT so every read and write afterward behaves like an ordinary blocking call. A failed attempt is ambiguous — it can mean nobody is connecting, or that a client attached in the same instant, reported as ERROR_PIPE_CONNECTED. Calling GetLastError() immediately after ConnectNamedPipe() proved unreliable. MQL5 may reset the last-error value between calls. A follow-up PeekNamedPipe() call, trusted only for its own direct return value, confirms the real state instead.

//+------------------------------------------------------------------+
//| TryAcceptClient                                                  |
//| Polls for a connecting client without blocking.                  |
//+------------------------------------------------------------------+
bool CPipeServer::TryAcceptClient(void)
  {
   if(m_handle == PIPE_INVALID_HANDLE)
      return(false);
   if(m_connected)
      return(true);
   uint nowait_mode = PIPE_READMODE_MESSAGE | PIPE_NOWAIT;
   ::SetNamedPipeHandleState(m_handle, nowait_mode, 0, 0);
   int  ok = ::ConnectNamedPipe(m_handle, 0);
//--- unconditionally restore blocking mode before any read/write happens
   uint wait_mode = PIPE_READMODE_MESSAGE | PIPE_WAIT;
   ::SetNamedPipeHandleState(m_handle, wait_mode, 0, 0);
   if(ok != 0)
     {
      m_connected = true;
      return(true);
     }
//--- ConnectNamedPipe's own return was false, but that's ambiguous: it
//--- could mean nobody is there yet, or that a client actually attached
//--- (the ERROR_PIPE_CONNECTED race). Verify independently rather than
//--- trusting GetLastError() for the distinction.
   uchar dummy[1];
   uint  bytes_read = 0, avail = 0, left = 0;
   int   peek_ok = ::PeekNamedPipe(m_handle, dummy, 0, bytes_read, avail, left);
   if(peek_ok != 0)
     {
      m_connected = true;                    // a client is genuinely attached
      return(true);
     }
   return(false);                            // confirmed: nobody waiting this tick
  }

ReadMessage() and WriteMessage() move one message per call. The client that just connected is already about to write, so the blocking read returns almost immediately in practice.

//+------------------------------------------------------------------+
//| ReadMessage                                                      |
//+------------------------------------------------------------------+
bool CPipeServer::ReadMessage(CMessage &msg)
  {
   if(m_handle == PIPE_INVALID_HANDLE || !m_connected)
      return(false);
   uchar buf[];
   ::ArrayResize(buf, MESSAGE_BYTE_SIZE);
   uint bytes_read = 0;
   int  ok = ::ReadFile(m_handle, buf, MESSAGE_BYTE_SIZE, bytes_read, 0);
   if(ok == 0 || bytes_read != (uint)MESSAGE_BYTE_SIZE)
     {
      m_last_error = ::GetLastError();
      return(false);
     }
   return(msg.Deserialize(buf));
  }

//+------------------------------------------------------------------+
//| WriteMessage                                                     |
//+------------------------------------------------------------------+
bool CPipeServer::WriteMessage(const CMessage &msg)
  {
   if(m_handle == PIPE_INVALID_HANDLE || !m_connected)
      return(false);
   uchar buf[];
   msg.Serialize(buf);
   uint bytes_written = 0;
   int  ok = ::WriteFile(m_handle, buf, MESSAGE_BYTE_SIZE, bytes_written, 0);
   if(ok == 0 || bytes_written != (uint)MESSAGE_BYTE_SIZE)
     {
      m_last_error = ::GetLastError();
      return(false);
     }
   return(true);
  }

Flush() closes a timing gap: DisconnectNamedPipe() can discard a response the client has not yet read. Calling FlushFileBuffers() first forces the bytes through before the connection drops.

//+------------------------------------------------------------------+
//| Flush                                                            |
//+------------------------------------------------------------------+
void CPipeServer::Flush(void)
  {
   if(m_handle != PIPE_INVALID_HANDLE && m_connected)
      ::FlushFileBuffers(m_handle);
  }

Disconnect() releases the current client while keeping the endpoint alive for the next one; Close() tears the endpoint down entirely.

//+------------------------------------------------------------------+
//| Disconnect                                                       |
//+------------------------------------------------------------------+
void CPipeServer::Disconnect(void)
  {
   if(m_handle != PIPE_INVALID_HANDLE && m_connected)
      ::DisconnectNamedPipe(m_handle);
   m_connected = false;
  }

//+------------------------------------------------------------------+
//| Close                                                            |
//+------------------------------------------------------------------+
void CPipeServer::Close(void)
  {
   if(m_handle != PIPE_INVALID_HANDLE)
     {
      ::CloseHandle(m_handle);
      m_handle = PIPE_INVALID_HANDLE;
     }
   m_connected = false;
  }

Section 4 — CPipeClient: the Slave's Connection

The client class is simpler than the server because it only initiates connections rather than accepting them. It uses CreateFileW() to open the existing named pipe by name with simultaneous read and write access, and CloseHandle() to close the handle when finished.

//+------------------------------------------------------------------+
//|                                                   PipeClient.mqh |
//+------------------------------------------------------------------+
#ifndef PIPECLIENT_MQH
#define PIPECLIENT_MQH

#include "Message.mqh"

//--- client access and open constants
#define GENERIC_READ            0x80000000
#define GENERIC_WRITE           0x40000000
#define OPEN_EXISTING           3
#define CLIENT_INVALID_HANDLE   (-1)

//--- Windows named pipe client API
#import "kernel32.dll"
long CreateFileW(string name, uint desired_access, uint share_mode, long security,
                 uint creation, uint flags, long template_file);
int  ReadFile(long handle, uchar &buffer[], uint bytes_to_read, uint &bytes_read, long overlapped);
int  WriteFile(long handle, uchar &buffer[], uint bytes_to_write, uint &bytes_written, long overlapped);
int  CloseHandle(long handle);
uint GetLastError(void);
#import

//+------------------------------------------------------------------+
//| CPipeClient                                                      |
//| Wraps the Windows named pipe client API for a slave EA. Opens an |
//| existing pipe by name and exchanges typed messages with the      |
//| broker EA using the CMessage serialization.                      |
//+------------------------------------------------------------------+
class CPipeClient
  {
private:
   string            m_pipe_name;
   long              m_handle;
   bool              m_connected;
   uint              m_last_error;

public:
                     CPipeClient(void);
                    ~CPipeClient(void);

   bool              Connect(const string pipe_name);
   bool              SendMessage(const CMessage &msg);
   bool              ReceiveMessage(CMessage &msg);
   void              Close(void);
   bool              IsConnected(void) const { return(m_connected); }
   uint              LastError(void)   const { return(m_last_error); }
  };

//+------------------------------------------------------------------+
//| Constructor: Marks the handle invalid and the client             |
//| unconnected until Connect succeeds.                              |
//+------------------------------------------------------------------+
CPipeClient::CPipeClient(void)
  {
//--- start from a known idle state
   m_pipe_name  = "";
   m_handle     = CLIENT_INVALID_HANDLE;
   m_connected  = false;
   m_last_error = 0;
  }

//+------------------------------------------------------------------+
//| Destructor: Closes the pipe handle if it is still open.          |
//+------------------------------------------------------------------+
CPipeClient::~CPipeClient(void)
  {
//--- release the handle on destruction
   Close();
  }

Connect() opens the pipe. An invalid handle usually just means the broker is not ready yet, so the method returns false rather than treating it as fatal; the slave retries on its next tick.

//+------------------------------------------------------------------+
//| Connect                                                          |
//+------------------------------------------------------------------+
bool CPipeClient::Connect(const string pipe_name)
  {
//--- open the existing pipe for duplex access
   m_pipe_name = pipe_name;
   uint access = GENERIC_READ | GENERIC_WRITE;
   m_handle = ::CreateFileW(m_pipe_name, access, 0, 0, OPEN_EXISTING, 0, 0);
//--- an invalid handle means the server is not ready
   if(m_handle == CLIENT_INVALID_HANDLE)
     {
      m_last_error = ::GetLastError();
      m_connected  = false;
      return(false);
     }
//--- the client is now attached to the endpoint
   m_connected = true;
   return(true);
  }

SendMessage() and ReceiveMessage() are the deliberate mirror of the server's write and read, so a message written by one side is always readable by the other.

//+------------------------------------------------------------------+
//| SendMessage                                                      |
//| Serializes msg to a byte array and writes it to the pipe.        |
//| Returns true when WriteFile confirms all bytes were written.     |
//+------------------------------------------------------------------+
bool CPipeClient::SendMessage(const CMessage &msg)
  {
//--- a connected pipe is required
   if(m_handle == CLIENT_INVALID_HANDLE || !m_connected)
      return(false);
//--- serialize the message to a flat byte array before writing
   uchar buf[];
   msg.Serialize(buf);
//--- write the full buffer to the named pipe in one call
   uint bytes_written = 0;
   int  ok = ::WriteFile(m_handle, buf, MESSAGE_BYTE_SIZE, bytes_written, 0);
   if(ok == 0 || bytes_written != (uint)MESSAGE_BYTE_SIZE)
     {
      m_last_error = ::GetLastError();
      ::PrintFormat("CPipeClient: SendMessage failed, wrote %d of %d bytes, error %d",
                    bytes_written, MESSAGE_BYTE_SIZE, m_last_error);
      return(false);
     }
   return(true);
  }

//+------------------------------------------------------------------+
//| ReceiveMessage                                                   |
//| Reads one serialized message from the pipe and deserializes it   |
//| into msg. Returns true only on a full, successfully decoded      |
//| message.                                                         |
//+------------------------------------------------------------------+
bool CPipeClient::ReceiveMessage(CMessage &msg)
  {
//--- a connected pipe is required
   if(m_handle == CLIENT_INVALID_HANDLE || !m_connected)
      return(false);
//--- read a full message-sized buffer from the pipe
   uchar buf[];
   ::ArrayResize(buf, MESSAGE_BYTE_SIZE);
   uint bytes_read = 0;
   int  ok = ::ReadFile(m_handle, buf, MESSAGE_BYTE_SIZE, bytes_read, 0);
//--- a short or failed read is not a valid message
   if(ok == 0 || bytes_read != (uint)MESSAGE_BYTE_SIZE)
     {
      m_last_error = ::GetLastError();
      ::PrintFormat("CPipeClient: ReceiveMessage failed, read %d of %d bytes, error %d",
                    bytes_read, MESSAGE_BYTE_SIZE, m_last_error);
      return(false);
     }
//--- decode the bytes into the typed message
   return(msg.Deserialize(buf));
  }

//+------------------------------------------------------------------+
//| Close                                                            |
//| Closes the client handle and marks the client unconnected.       |
//+------------------------------------------------------------------+
void CPipeClient::Close(void)
  {
//--- release the handle if it is open
   if(m_handle != CLIENT_INVALID_HANDLE)
     {
      ::CloseHandle(m_handle);
      m_handle = CLIENT_INVALID_HANDLE;
     }
   m_connected = false;
  }

Section 5 — CMessageRegistry: Tracking Sender State and Detecting Silence

Every incoming message is stored twice: once as the latest of its type, once as the latest from its sender. The type index is a small fixed array; the sender index is a set of parallel dynamic arrays that grow as new senders appear.

//+------------------------------------------------------------------+
//|                                              MessageRegistry.mqh |
//+------------------------------------------------------------------+
#ifndef MESSAGEREGISTRY_MQH
#define MESSAGEREGISTRY_MQH

#include "Message.mqh"

#define REGISTRY_TYPE_SLOTS 8

//+------------------------------------------------------------------+
//| CMessageRegistry                                                 |
//| Stores the most recent typed message per message type and per    |
//| sender magic. Records a GetTickCount64 receipt timestamp for     |
//| every sender so the dashboard can detect when one goes silent.   |
//+------------------------------------------------------------------+
class CMessageRegistry
  {
private:
   CMessage          m_by_type[REGISTRY_TYPE_SLOTS];
   bool              m_type_seen[REGISTRY_TYPE_SLOTS];
   ulong             m_sender_magic[];
   CMessage          m_sender_msg[];
   ulong             m_sender_seen_ms[];
   int               m_sender_count;

public:
                     CMessageRegistry(void);
                    ~CMessageRegistry(void);

   void              Store(const CMessage &msg);
   bool              GetLatest(ENUM_MESSAGE_TYPE type, CMessage &msg);
   bool              GetBySender(ulong magic, CMessage &msg);
   bool              GetSenderStatus(const int index, ulong &magic_out, string &symbol_out,
                                     ulong &last_seen_ms_out, datetime &last_seen_time_out);
   int               Count(void)       const { return(m_sender_count); }
   int               SenderCount(void) const { return(m_sender_count); }
  };

//+------------------------------------------------------------------+
//| Constructor: Clears the per-type seen flags and the sender       |
//| count so the registry starts empty.                              |
//+------------------------------------------------------------------+
CMessageRegistry::CMessageRegistry(void)
  {
//--- mark every type slot as not yet seen
   for(int i = 0; i < REGISTRY_TYPE_SLOTS; i++)
      m_type_seen[i] = false;
//--- no senders have reported yet
   m_sender_count = 0;
   ::ArrayResize(m_sender_magic,   0);
   ::ArrayResize(m_sender_msg,     0);
   ::ArrayResize(m_sender_seen_ms, 0);
  }

//+------------------------------------------------------------------+
//| Destructor: No owned resources beyond the dynamic arrays, which  |
//| the runtime releases automatically.                              |
//+------------------------------------------------------------------+
CMessageRegistry::~CMessageRegistry(void)
  {
  }

Store() writes the message into its type slot, then searches the sender arrays for a matching magic number. A match overwrites the entry and refreshes the receipt time; no match appends a new sender. That receipt time, from GetTickCount64(), is what later tells the dashboard whether a sender is active or has gone quiet.

//+------------------------------------------------------------------+
//| Store                                                            |
//| Records msg as the latest message of its type and as the latest  |
//| message from its sender. Stamps the wall clock receipt time in   |
//| milliseconds for every write so silence detection works.         |
//+------------------------------------------------------------------+
void CMessageRegistry::Store(const CMessage &msg)
  {
//--- index by message type when the type is in range
   int slot = (int)msg.message_type;
   if(slot >= 0 && slot < REGISTRY_TYPE_SLOTS)
     {
      m_by_type[slot]   = msg;
      m_type_seen[slot] = true;
     }
//--- search for an existing entry from this sender
   for(int i = 0; i < m_sender_count; i++)
     {
      if(m_sender_magic[i] == msg.sender_magic)
        {
         //--- overwrite the message and refresh the receipt timestamp
         m_sender_msg[i]     = msg;
         m_sender_seen_ms[i] = ::GetTickCount64();
         return;
        }
     }
//--- a new sender: append magic, message, and receipt timestamp
   ::ArrayResize(m_sender_magic,   m_sender_count + 1);
   ::ArrayResize(m_sender_msg,     m_sender_count + 1);
   ::ArrayResize(m_sender_seen_ms, m_sender_count + 1);
   m_sender_magic[m_sender_count]   = msg.sender_magic;
   m_sender_msg[m_sender_count]     = msg;
   m_sender_seen_ms[m_sender_count] = ::GetTickCount64();
   m_sender_count++;
  }

GetLatest() and GetBySender() are the two direct lookups against what Store() recorded.

//+------------------------------------------------------------------+
//| GetLatest                                                        |
//| Returns the most recent stored message of the given type. Yields |
//| false when no message of that type has been received.            |
//+------------------------------------------------------------------+
bool CMessageRegistry::GetLatest(ENUM_MESSAGE_TYPE type, CMessage &msg)
  {
//--- reject an out of range type slot
   int slot = (int)type;
   if(slot < 0 || slot >= REGISTRY_TYPE_SLOTS)
      return(false);
//--- return the stored message only if one was seen
   if(!m_type_seen[slot])
      return(false);
   msg = m_by_type[slot];
   return(true);
  }

//+------------------------------------------------------------------+
//| GetBySender                                                      |
//| Returns the most recent message from the sender with the given   |
//| magic. Yields false when that sender has never reported.         |
//+------------------------------------------------------------------+
bool CMessageRegistry::GetBySender(ulong magic, CMessage &msg)
  {
//--- scans the sender entries for a matching magic
   for(int i = 0; i < m_sender_count; i++)
     {
      if(m_sender_magic[i] == magic)
        {
         msg = m_sender_msg[i];
         return(true);
        }
     }
//--- no entry for that sender
   return(false);
  }

GetSenderStatus() feeds the dashboard directly, returning one sender's magic, symbol, receipt time, and message timestamp by index.

//+------------------------------------------------------------------+
//| GetSenderStatus                                                  |
//| Returns identity and timing fields for the sender at the given   |
//| index. Returns false when the index is out of range.             |
//+------------------------------------------------------------------+
bool CMessageRegistry::GetSenderStatus(const int index, ulong &magic_out,
                                       string &symbol_out, ulong &last_seen_ms_out,
                                       datetime &last_seen_time_out)
  {
//--- reject an out of range index
   if(index < 0 || index >= m_sender_count)
      return(false);
//--- fill the output fields from the stored entry
   magic_out          = m_sender_magic[index];
   symbol_out         = m_sender_msg[index].GetSymbol();
   last_seen_ms_out   = m_sender_seen_ms[index];
   last_seen_time_out = m_sender_msg[index].timestamp;
   return(true);
  }

Section 6 — CRiskAggregator: Computing and Attributing Portfolio Risk

For every open position, the aggregator measures the distance from entry to stop loss, converts it to money using the instrument's tick value, scales by volume, and adds it to a running total. A position with no stop loss contributes nothing.

//+------------------------------------------------------------------+
//|                                               RiskAggregator.mqh |
//+------------------------------------------------------------------+
#ifndef RISKAGGREGATOR_MQH
#define RISKAGGREGATOR_MQH

//+------------------------------------------------------------------+
//| CRiskAggregator                                                  |
//| Computes the portfolio aggregate risk by summing the money at    |
//| risk across every open position, using each position's stop loss |
//| distance converted to account currency. ComputeBreakdown reports |
//| the same total split out per symbol, which lets the broker EA    |
//| identify which symbol's risk actually changed between ticks,     |
//| regardless of which slave EA's magic number a position carries.  |
//+------------------------------------------------------------------+
class CRiskAggregator
  {
private:
   double            PointValuePerLot(const string symbol);

public:
                     CRiskAggregator(void);
                    ~CRiskAggregator(void);

   double            RiskForPosition(const double volume, const double stop_points,
                                     const double value_per_point_per_lot);
   double            Compute(void);
   void              ComputeBreakdown(string &symbols_out[], double &risks_out[]);
  };

//+------------------------------------------------------------------+
//| Constructor: The class holds no state between calls.             |
//+------------------------------------------------------------------+
CRiskAggregator::CRiskAggregator(void)
  {
  }

//+------------------------------------------------------------------+
//| Destructor: Nothing to release.                                  |
//+------------------------------------------------------------------+
CRiskAggregator::~CRiskAggregator(void)
  {
  }

PointValuePerLot() returns what one point of movement is worth, in account currency, for one lot.

//+------------------------------------------------------------------+
//| PointValuePerLot                                                 |
//| Returns the account currency value of one point of price move    |
//| for one lot of the symbol, derived from the tick value and tick  |
//| size. Returns zero when the symbol data is unavailable.          |
//+------------------------------------------------------------------+
double CRiskAggregator::PointValuePerLot(const string symbol)
  {
//--- read the tick geometry for the instrument
   double point     = ::SymbolInfoDouble(symbol, SYMBOL_POINT);
   double tick_size = ::SymbolInfoDouble(symbol, SYMBOL_TRADE_TICK_SIZE);
   double tick_val  = ::SymbolInfoDouble(symbol, SYMBOL_TRADE_TICK_VALUE);
//--- guard against missing or zero instrument data
   if(tick_size <= 0.0 || point <= 0.0)
      return(0.0);
//--- value of one point equals the tick value scaled by point over tick size
   return(tick_val * point / tick_size);
  }

RiskForPosition() is the pure formula underneath everything else, which is what makes it directly testable with hand-picked numbers.

//+------------------------------------------------------------------+
//| RiskForPosition                                                  |
//| Pure risk formula for a single position: volume multiplied by    |
//| the stop distance in points multiplied by the money value of one |
//| point per lot. Kept separate so the logic is testable.           |
//+------------------------------------------------------------------+
double CRiskAggregator::RiskForPosition(const double volume, const double stop_points,
                                        const double value_per_point_per_lot)
  {
//--- money at risk is volume times point distance times point value
   return(volume * stop_points * value_per_point_per_lot);
  }

ComputeBreakdown() iterates over all open positions in the account and accumulates per-symbol risk into parallel arrays. That split is what lets the broker later say which instrument actually caused a change in the aggregate figure.

//+------------------------------------------------------------------+
//| ComputeBreakdown                                                 |
//| Iterates every open position and accumulates money at risk per   |
//| symbol, filling parallel output arrays. This is the authoritative|
//| account-wide view: it does not depend on any slave EA's magic    |
//| number, so it correctly reflects risk from positions opened      |
//| manually or by any EA regardless of how it identifies itself.    |
//+------------------------------------------------------------------+
void CRiskAggregator::ComputeBreakdown(string &symbols_out[], double &risks_out[])
  {
//--- start from an empty breakdown
   ::ArrayResize(symbols_out, 0);
   ::ArrayResize(risks_out, 0);
   int count = ::PositionsTotal();
   for(int i = 0; i < count; i++)
     {
      //--- select the position by its ticket
      ulong ticket = ::PositionGetTicket(i);
      if(ticket == 0)
         continue;
      //--- read the fields needed for the risk calculation
      string symbol = ::PositionGetString(POSITION_SYMBOL);
      double volume = ::PositionGetDouble(POSITION_VOLUME);
      double open   = ::PositionGetDouble(POSITION_PRICE_OPEN);
      double sl     = ::PositionGetDouble(POSITION_SL);
      //--- a position without a stop loss has no defined risk
      if(sl <= 0.0)
         continue;
      //--- convert the stop distance to points
      double point = ::SymbolInfoDouble(symbol, SYMBOL_POINT);
      if(point <= 0.0)
         continue;
      double stop_points = ::MathAbs(open - sl) / point;
      double vpp          = PointValuePerLot(symbol);
      double risk          = RiskForPosition(volume, stop_points, vpp);
      //--- accumulate into the existing entry for this symbol, or add a new one
      int idx = -1;
      int n   = ::ArraySize(symbols_out);
      for(int j = 0; j < n; j++)
        {
         if(symbols_out[j] == symbol)
           {
            idx = j;
            break;
           }
        }
      if(idx == -1)
        {
         ::ArrayResize(symbols_out, n + 1);
         ::ArrayResize(risks_out,   n + 1);
         symbols_out[n] = symbol;
         risks_out[n]   = risk;
        }
      else
        {
         risks_out[idx] += risk;
        }
     }
  }

Compute() folds that breakdown into a single scalar total for callers who only need the aggregate figure.

//+------------------------------------------------------------------+
//| Compute                                                          |
//| Returns the portfolio aggregate risk as a single scalar, by      |
//| summing the per-symbol breakdown. Behavior is unchanged from     |
//| earlier versions of this class.                                  |
//+------------------------------------------------------------------+
double CRiskAggregator::Compute(void)
  {
//--- sum the per-symbol breakdown into a single total
   string symbols[];
   double risks[];
   ComputeBreakdown(symbols, risks);
   double total = 0.0;
   int    n     = ::ArraySize(risks);
   for(int i = 0; i < n; i++)
      total += risks[i];
   return(total);
  }

Section 7 — CPipeBrokerDashboard: the Broker's Control Panel

The dashboard renders a chart comment: portfolio-level figures at the top, then one status row per sender.

//+------------------------------------------------------------------+
//|                                          PipeBrokerDashboard.mqh |
//+------------------------------------------------------------------+
#ifndef PIPEBROKERDASHBOARD_MQH
#define PIPEBROKERDASHBOARD_MQH

//+------------------------------------------------------------------+
//| CPipeBrokerDashboard                                             |
//| Renders the broker EA control panel as a chart comment. Shows    |
//| aggregate risk and message rate at the top, then one status row  |
//| per known sender with an ACTIVE or SILENT badge and the          |
//| timestamp of the last received message from that sender.         |
//+------------------------------------------------------------------+
class CPipeBrokerDashboard
  {
private:
   string            m_title;

public:
                     CPipeBrokerDashboard(void);
                    ~CPipeBrokerDashboard(void);

   void              Update(const double   aggregate_risk,
                            const int      sender_count,
                            const ulong    &sender_magic[],
                            const string   &sender_symbol[],
                            const ulong    &sender_seen_ms[],
                            const datetime &sender_seen_time[],
                            const ulong    silence_threshold_ms,
                            const int      msgs_last_second);
   void              Clear(void);
  };

//+------------------------------------------------------------------+
//| Constructor: Fixes the panel title text.                         |
//+------------------------------------------------------------------+
CPipeBrokerDashboard::CPipeBrokerDashboard(void)
  {
//--- the panel heading shown at the top of the comment
   m_title = "MQL5 Message Bus - Broker EA";
  }

//+------------------------------------------------------------------+
//| Destructor: Clears the chart comment on teardown.                |
//+------------------------------------------------------------------+
CPipeBrokerDashboard::~CPipeBrokerDashboard(void)
  {
//--- leave the chart clean
   Clear();
  }

Update() compares elapsed milliseconds since each sender's last receipt against a configurable threshold to mark it ACTIVE or SILENT. Because the broker calls this every tick regardless of whether a client happens to be connected, a badge flips to SILENT on schedule even after every slave has been removed.

//+------------------------------------------------------------------+
//| Update                                                           |
//+------------------------------------------------------------------+
void CPipeBrokerDashboard::Update(const double   aggregate_risk,
                                  const int      sender_count,
                                  const ulong    &sender_magic[],
                                  const string   &sender_symbol[],
                                  const ulong    &sender_seen_ms[],
                                  const datetime &sender_seen_time[],
                                  const ulong    silence_threshold_ms,
                                  const int      msgs_last_second)
  {
//--- header and portfolio-level figures
   string panel = m_title + "\n";
   panel += "-------------------------------\n";
   panel += ::StringFormat("Aggregate risk    : %.2f\n", aggregate_risk);
   panel += ::StringFormat("Messages / second : %d\n",   msgs_last_second);
   panel += "-------------------------------\n";
//--- one status row per known sender
   ulong now_ms = ::GetTickCount64();
   for(int i = 0; i < sender_count; i++)
     {
      //--- derive status from elapsed milliseconds since last receipt
      ulong  elapsed_ms = now_ms - sender_seen_ms[i];
      string status     = (elapsed_ms <= silence_threshold_ms) ? "ACTIVE" : "SILENT";
      string last_seen  = ::TimeToString(sender_seen_time[i], TIME_DATE | TIME_SECONDS);
      panel += ::StringFormat("  [%I64u] %-6s  %s  last:%s\n",
                              sender_magic[i], sender_symbol[i], status, last_seen);
     }
//--- render the assembled block to the chart
   ::Comment(panel);
  }

//+------------------------------------------------------------------+
//| Clear                                                            |
//| Removes the panel by writing an empty chart comment.             |
//+------------------------------------------------------------------+
void CPipeBrokerDashboard::Clear(void)
  {
//--- erase the comment text
   ::Comment("");
  }

Section 8 — CPipeBrokerEA.mq5: Assembling the Broker

The broker Expert Advisor is where every earlier piece finally comes together into one running loop. DetectCauseSymbol() compares the current per-symbol breakdown against the previous tick's snapshot, looking for a symbol that is new, one whose risk moved by at least a cent, or one that disappeared entirely. That threshold filters out ordinary exchange-rate noise on pairs like USDJPY, which would otherwise drift every tick with no real position change behind it.

//+------------------------------------------------------------------+
//|                                                CPipeBrokerEA.mq5 |
//+------------------------------------------------------------------+

#include <MessageBusEngine/MessageType.mqh>
#include <MessageBusEngine/Message.mqh>
#include <MessageBusEngine/PipeServer.mqh>
#include <MessageBusEngine/MessageRegistry.mqh>
#include <MessageBusEngine/RiskAggregator.mqh>
#include <MessageBusEngine/PipeBrokerDashboard.mqh>

//--- Inputs
input string InpPipeName           = "MQL5MessageBus";  // named pipe identifier
input int    InpTimerMs            = 100;               // message loop interval, milliseconds
input ulong  InpBrokerMagic        = 90000;             // broker EA magic for its own responses
input ulong  InpSilenceThresholdMs = 2000;              // ms without a message before SILENT badge

//--- Components
CPipeServer          g_server;
CMessageRegistry     g_registry;
CRiskAggregator      g_aggregator;
CPipeBrokerDashboard g_dashboard;

//--- Runtime state
string   g_pipe_path         = "";
ulong    g_response_seq      = 0;
double   g_aggregate_risk    = 0.0;
datetime g_last_msg_time     = 0;
int      g_msgs_this_second  = 0;
int      g_msgs_last_second  = 0;
datetime g_current_second    = 0;
string   g_prev_symbols[];
double   g_prev_risks[];
string   g_last_cause_symbol = "";

//+------------------------------------------------------------------+
//| DetectCauseSymbol                                                |
//| Compares the newly computed per-symbol risk breakdown against    |
//| the previous tick's snapshot and returns the symbol responsible  |
//| for the change: a symbol that is new, one whose risk value       |
//| differs, or one that disappeared entirely (fully closed).        |
//+------------------------------------------------------------------+
string DetectCauseSymbol(const string &new_symbols[], const double &new_risks[])
  {
   string cause  = "";
   int    new_n  = ::ArraySize(new_symbols);
   int    old_n  = ::ArraySize(g_prev_symbols);
   for(int i = 0; i < new_n; i++)
     {
      bool found = false;
      for(int j = 0; j < old_n; j++)
        {
         if(g_prev_symbols[j] == new_symbols[i])
           {
            found = true;
            if(::MathAbs(g_prev_risks[j] - new_risks[i]) >= 0.01)
               cause = new_symbols[i];
            break;
           }
        }
      if(!found)
         cause = new_symbols[i];
     }
   for(int j = 0; j < old_n; j++)
     {
      bool still_present = false;
      for(int i = 0; i < new_n; i++)
        {
         if(new_symbols[i] == g_prev_symbols[j])
           {
            still_present = true;
            break;
           }
        }
      if(!still_present)
         cause = g_prev_symbols[j];
     }
   ::ArrayResize(g_prev_symbols, new_n);
   ::ArrayResize(g_prev_risks,   new_n);
   for(int i = 0; i < new_n; i++)
     {
      g_prev_symbols[i] = new_symbols[i];
      g_prev_risks[i]   = new_risks[i];
     }
   return(cause);
  }

OnInit() builds the pipe path, creates the server, and starts the timer that drives the message loop.

//+------------------------------------------------------------------+
//| OnInit                                                           |
//+------------------------------------------------------------------+
int OnInit(void)
  {
   g_pipe_path = "\\\\.\\pipe\\" + InpPipeName;
   if(!g_server.Create(g_pipe_path))
     {
      Print("CPipeBrokerEA: failed to create pipe server, EA will not start");
      return(INIT_FAILED);
     }
   if(!EventSetMillisecondTimer(InpTimerMs))
     {
      Print("CPipeBrokerEA: failed to set timer");
      return(INIT_FAILED);
     }
   PrintFormat("CPipeBrokerEA: pipe server '%s' ready, waiting for slave EAs", g_pipe_path);
   return(INIT_SUCCEEDED);
  }

OnDeinit() stops the timer and releases the pipe. Because TryAcceptClient() never blocks, OnTimer() always returns quickly, so removing the EA does not stall the terminal, even with zero slaves connected.

//+------------------------------------------------------------------+
//| OnDeinit                                                         |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
   EventKillTimer();
   g_server.Disconnect();
   g_server.Close();
   g_dashboard.Clear();
  }

RollSecondCounter() is a small piece of bookkeeping that keeps the dashboard's messages-per-second figure honest. It watches for the wall clock second to change, and when it does, it rolls the running tally over into the "last second" figure and starts counting the new second from zero.

//+------------------------------------------------------------------+
//| RollSecondCounter                                                |
//+------------------------------------------------------------------+
void RollSecondCounter(void)
  {
   datetime now = TimeCurrent();
   if(now != g_current_second)
     {
      g_msgs_last_second = g_msgs_this_second;
      g_msgs_this_second = 0;
      g_current_second   = now;
     }
  }

RefreshDashboard() gathers every known sender from the registry and hands the snapshot to the dashboard. It runs on every tick, connected or not, which is exactly what lets a badge flip to SILENT on schedule.

//+------------------------------------------------------------------+
//| RefreshDashboard                                                 |
//| Called every timer tick unconditionally — with zero clients      |
//| connected as much as with one — which is what lets the last      |
//| remaining slave's badge flip from ACTIVE to SILENT once it stops |
//| reporting.                                                       |
//+------------------------------------------------------------------+
void RefreshDashboard(void)
  {
   int      count = g_registry.SenderCount();
   ulong    magic_arr[];
   string   symbol_arr[];
   ulong    seen_ms_arr[];
   datetime seen_time_arr[];
   ::ArrayResize(magic_arr,     count);
   ::ArrayResize(symbol_arr,    count);
   ::ArrayResize(seen_ms_arr,   count);
   ::ArrayResize(seen_time_arr, count);
   for(int i = 0; i < count; i++)
     {
      ulong    m  = 0;
      string   s  = "";
      ulong    ms = 0;
      datetime dt = 0;
      g_registry.GetSenderStatus(i, m, s, ms, dt);
      magic_arr[i]     = m;
      symbol_arr[i]    = s;
      seen_ms_arr[i]   = ms;
      seen_time_arr[i] = dt;
     }
   g_dashboard.Update(g_aggregate_risk, count, magic_arr, symbol_arr,
                      seen_ms_arr, seen_time_arr, InpSilenceThresholdMs,
                      g_msgs_last_second);
  }

OnTimer() ties everything together: accept a client, read one message, recompute the risk breakdown, respond, and disconnect so the next slave gets a turn.

//+------------------------------------------------------------------+
//| OnTimer                                                          |
//+------------------------------------------------------------------+
void OnTimer(void)
  {
   RollSecondCounter();
   if(!g_server.TryAcceptClient())
     {
      //--- no client this tick; refresh the panel and return
      RefreshDashboard();
      return;
     }
   CMessage incoming;
   if(g_server.ReadMessage(incoming))
     {
      g_registry.Store(incoming);
      g_last_msg_time = TimeCurrent();
      g_msgs_this_second++;
      string bd_symbols[];
      double bd_risks[];
      g_aggregator.ComputeBreakdown(bd_symbols, bd_risks);
      double new_total = 0.0;
      int    bd_n      = ::ArraySize(bd_risks);
      for(int i = 0; i < bd_n; i++)
         new_total += bd_risks[i];
      g_aggregate_risk = new_total;
      string cause = DetectCauseSymbol(bd_symbols, bd_risks);
      if(cause != "")
         g_last_cause_symbol = cause;
      CMessage response;
      response.message_type   = MSG_RISK_RESPONSE;
      response.sender_magic   = InpBrokerMagic;
      response.SetSymbol(g_last_cause_symbol != "" ? g_last_cause_symbol : _Symbol);
      response.payload_double = g_aggregate_risk;
      response.payload_int    = g_registry.Count();
      response.sequence_id    = ++g_response_seq;
      response.timestamp      = TimeCurrent();
      g_server.WriteMessage(response);
      g_server.Flush();
     }
//--- release this client so the next slave can connect next tick
   g_server.Disconnect();
//--- refresh the control panel every tick, connected or not
   RefreshDashboard();
  }

The broker EA's dashboard, rendered as a chart comment. The header shows aggregate risk and message rate. Below it, each sender gets a row with its symbol, ACTIVE/SILENT badge, and last-seen time.

Section 9 — CSlaveEA.mq5: the Slave Template

The slave Expert Advisor is a single template parameterized by a magic number input, so three copies attached to three different charts become three entirely independent clients on the same bus. It connects to the pipe during OnInit() and starts its own periodic timer, and because the broker disconnects after serving every single request, the slave treats each connection as a one-shot transaction rather than something to hold open indefinitely.

//+------------------------------------------------------------------+
//|                                                     CSlaveEA.mq5 |
//+------------------------------------------------------------------+

#include <MessageBusEngine/MessageType.mqh>
#include <MessageBusEngine/Message.mqh>
#include <MessageBusEngine/PipeClient.mqh>

//--- Inputs
input string InpPipeName   = "MQL5MessageBus";  // must match the broker EA
input int    InpTimerMs    = 250;               // report interval, milliseconds
input ulong  InpSlaveMagic = 1;                 // this slave EA's identity

//--- Components
CPipeClient g_client;

//--- runtime state
string g_pipe_path   = "";
ulong  g_send_seq    = 0;

//+------------------------------------------------------------------+
//| OnInit                                                           |
//| Builds the pipe path, attempts an initial connection, and starts |
//| the report timer. A failed initial connect is not fatal; the     |
//| slave retries on each timer tick.                                |
//+------------------------------------------------------------------+
int OnInit(void)
  {
//--- compose the shared local pipe path
   g_pipe_path = "\\\\.\\pipe\\" + InpPipeName;
//--- an initial connection attempt; retried later if it fails
   if(g_client.Connect(g_pipe_path))
      PrintFormat("CSlaveEA[%I64u]: connected to '%s'", InpSlaveMagic, g_pipe_path);
   else
      PrintFormat("CSlaveEA[%I64u]: broker not ready yet, will retry", InpSlaveMagic);
//--- start the periodic report timer
   if(!EventSetMillisecondTimer(InpTimerMs))
     {
      Print("CSlaveEA: failed to set timer");
      return(INIT_FAILED);
     }
   return(INIT_SUCCEEDED);
  }

//+------------------------------------------------------------------+
//| OnDeinit                                                         |
//| Closes the pipe client on teardown.                              |
//+------------------------------------------------------------------+
void OnDeinit(const int reason)
  {
//--- release the client handle
   g_client.Close();
  }

LocalPositionSnapshot() counts only positions whose magic matches the slave's own. Without that filter, PositionsTotal() would reflect the whole account, and every slave would see the same count change whenever any EA opened or closed a trade.

//+------------------------------------------------------------------+
//| LocalPositionSnapshot                                            |
//+------------------------------------------------------------------+
void LocalPositionSnapshot(int &position_count, double &largest_volume)
  {
//--- scan only positions belonging to this slave's magic
   position_count = 0;
   largest_volume = 0.0;
   int total = PositionsTotal();
   for(int i = 0; i < total; i++)
     {
      ulong ticket = PositionGetTicket(i);
      if(ticket == 0)
         continue;
      //--- skip positions that belong to a different EA
      if(PositionGetInteger(POSITION_MAGIC) != (long)InpSlaveMagic)
         continue;
      position_count++;
      double vol = PositionGetDouble(POSITION_VOLUME);
      if(vol > largest_volume)
         largest_volume = vol;
     }
  }

OnTimer() sends a position update, waits for the aggregate risk response, and logs only when that figure actually changes by at least a cent, naming the symbol the broker identified as the cause.

//+------------------------------------------------------------------+
//| OnTimer                                                          |
//+------------------------------------------------------------------+
void OnTimer(void)
  {
//--- ensure the slave is connected before sending
   if(!g_client.IsConnected())
     {
      if(!g_client.Connect(g_pipe_path))
         return;
     }
//--- build the position update message from this slave's own positions only
   int    position_count = 0;
   double largest_volume = 0.0;
   LocalPositionSnapshot(position_count, largest_volume);
   CMessage update;
   update.message_type   = MSG_POSITION_UPDATE;
   update.sender_magic   = InpSlaveMagic;
   update.SetSymbol(_Symbol);
   update.payload_double = largest_volume;
   update.payload_int    = position_count;
   update.sequence_id    = ++g_send_seq;
   update.timestamp      = TimeCurrent();
//--- stamp the send time for the round-trip measurement
   ulong t_send = GetMicrosecondCount();
//--- send the update and abandon the tick on failure
   if(!g_client.SendMessage(update))
     {
      g_client.Close();
      return;
     }
//--- wait for the aggregate risk response from the broker
   CMessage response;
   if(!g_client.ReceiveMessage(response))
     {
      g_client.Close();
      return;
     }
//--- measure the round-trip latency in milliseconds
   ulong  t_recv        = GetMicrosecondCount();
   double round_trip_ms = (double)(t_recv - t_send) / 1000.0;
//--- log whenever the aggregate risk changes by at least a cent, naming the
//--- symbol the broker identified as the actual cause. The cent threshold
//--- filters out currency-conversion noise: for pairs like USDJPY or
//--- USDCHF, the tick value used to compute risk is derived from the live
//--- exchange rate, so the raw risk figure drifts by fractions of a cent
//--- on every tick even with no position change at all.
   static double s_last_risk = -1.0;
   double rounded_risk = ::MathRound(response.payload_double * 100.0) / 100.0;
   if(s_last_risk < 0.0 || ::MathAbs(rounded_risk - s_last_risk) >= 0.01)
     {
      s_last_risk = rounded_risk;
      string cause_symbol = response.GetSymbol();
      PrintFormat("CSlaveEA[%I64u] (%s): aggregate risk changed to %.2f (change originated on %s), round-trip latency = %.3f ms",
                  InpSlaveMagic, _Symbol, response.payload_double, cause_symbol, round_trip_ms);
     }
//--- the broker disconnects after serving, so drop our side too
   g_client.Close();
  }

Section 10 — TestMessageBus.mq5: Verifying the Logic Without a Live Pipe

The verification script exercises the pieces of logic that can be checked without ever opening a real pipe: serialization, registry behavior, and the risk formula. It leans on a small ASSERT macro that logs a pass or a fail as it goes, and each of the three test groups targets one specific guarantee the bus depends on.

//+------------------------------------------------------------------+
//|                                               TestMessageBus.mq5 |
//+------------------------------------------------------------------+
#property script_show_inputs false

#include <MessageBusEngine/MessageType.mqh>
#include <MessageBusEngine/Message.mqh>
#include <MessageBusEngine/MessageRegistry.mqh>
#include <MessageBusEngine/RiskAggregator.mqh>

//--- test counters
int g_pass = 0;
int g_fail = 0;

//--- assertion macro: logs the outcome and updates the counters
#define ASSERT(cond, msg) if(cond) { g_pass++; Print("PASS: ", msg); } else { g_fail++; Print("FAIL: ", msg); }

TestSerialization() checks that every field, including the exact double payload, survives a round trip through Serialize() and Deserialize().

//+------------------------------------------------------------------+
//| TestSerialization                                                |
//| Builds a message, serializes and deserializes it, and asserts    |
//| every field round-trips exactly, including the double payload.   |
//+------------------------------------------------------------------+
void TestSerialization(void)
  {
//--- build a message with known field values
   CMessage original;
   original.message_type   = MSG_RISK_RESPONSE;
   original.sender_magic   = 99001;
   original.SetSymbol("EURUSD");
   original.payload_double = 142.75;
   original.payload_int    = 0;
   original.sequence_id    = 7;
   original.timestamp      = (datetime)1752566400;
//--- serialize to a flat byte array
   uchar buf[];
   original.Serialize(buf);
   ASSERT(ArraySize(buf) == MESSAGE_BYTE_SIZE, "serialized buffer is 52 bytes");
//--- deserialize into a second message
   CMessage decoded;
   bool ok = decoded.Deserialize(buf);
   ASSERT(ok, "deserialize reports success");
//--- assert every field survived the round trip
   ASSERT(decoded.message_type == MSG_RISK_RESPONSE, "message_type round-trips");
   ASSERT(decoded.sender_magic == 99001, "sender_magic round-trips");
   ASSERT(decoded.GetSymbol() == "EURUSD", "symbol round-trips");
   ASSERT(decoded.payload_double == 142.75, "payload_double round-trips exactly");
   ASSERT(decoded.sequence_id == 7, "sequence_id round-trips");
   ASSERT(decoded.timestamp == (datetime)1752566400, "timestamp round-trips");
  }

TestRegistry() stores two responses from two senders and checks that GetLatest(), Count(), and GetBySender() all report correctly.

//+------------------------------------------------------------------+
//| TestRegistry                                                     |
//| Stores two messages of one type with different sequence ids and  |
//| asserts the later one wins, then asserts the unique sender count.|
//+------------------------------------------------------------------+
void TestRegistry(void)
  {
   CMessageRegistry registry;
//--- store an earlier risk response from sender 501
   CMessage first;
   first.message_type = MSG_RISK_RESPONSE;
   first.sender_magic = 501;
   first.sequence_id  = 10;
   first.payload_double = 100.0;
   registry.Store(first);
//--- store a later risk response from sender 502
   CMessage second;
   second.message_type = MSG_RISK_RESPONSE;
   second.sender_magic = 502;
   second.sequence_id  = 20;
   second.payload_double = 200.0;
   registry.Store(second);
//--- the latest of that type must be the later sequence id
   CMessage latest;
   bool got = registry.GetLatest(MSG_RISK_RESPONSE, latest);
   ASSERT(got, "GetLatest finds a stored risk response");
   ASSERT(latest.sequence_id == 20, "GetLatest returns the later message");
//--- two distinct senders must be counted
   ASSERT(registry.Count() == 2, "registry counts two unique senders");
//--- retrieval by sender must return that sender's message
   CMessage by_sender;
   bool found = registry.GetBySender(501, by_sender);
   ASSERT(found && by_sender.sequence_id == 10, "GetBySender returns sender 501 message");
  }

TestAggregation() checks the pure risk formula against three hand-picked positions, whose sum should equal sixteen dollars.

//+------------------------------------------------------------------+
//| TestAggregation                                                  |
//| Calls the pure risk formula with the three validated positions   |
//| and asserts the aggregate equals sixteen.                        |
//+------------------------------------------------------------------+
void TestAggregation(void)
  {
   CRiskAggregator aggregator;
//--- three positions: (0.10,50,1), (0.20,30,1), (0.05,100,1)
   double r1 = aggregator.RiskForPosition(0.10, 50, 1.0);
   double r2 = aggregator.RiskForPosition(0.20, 30, 1.0);
   double r3 = aggregator.RiskForPosition(0.05, 100, 1.0);
//--- each position risk must match the hand calculation
   ASSERT(MathAbs(r1 - 5.0) < 0.0000001, "position 1 risk is 5.00");
   ASSERT(MathAbs(r2 - 6.0) < 0.0000001, "position 2 risk is 6.00");
   ASSERT(MathAbs(r3 - 5.0) < 0.0000001, "position 3 risk is 5.00");
//--- the aggregate must total sixteen
   double aggregate = r1 + r2 + r3;
   ASSERT(MathAbs(aggregate - 16.0) < 0.0000001, "aggregate risk is 16.00");
  }

OnStart() runs the three groups above in sequence and prints a final pass-or-fail summary.

//+------------------------------------------------------------------+
//| OnStart                                                          |
//| Runs every test group and prints the pass/fail summary.          |
//+------------------------------------------------------------------+
void OnStart(void)
  {
//--- run all groups
   Print("=== TestMessageBus: starting ===");
   TestSerialization();
   TestRegistry();
   TestAggregation();
//--- print the summary
   PrintFormat("=== TestMessageBus: %d passed, %d failed ===", g_pass, g_fail);
   if(g_fail == 0)
      Print("=== ALL TESTS PASSED ===");
  }

Section 11 — Extending the Bus

The schema-in-code approach leaves several natural extensions open, and each one fits the existing structure without demanding a rewrite.

A message expiry time would let the broker ignore stale reports from a slave that has gone quiet without formally disconnecting. Since every message already carries its own timestamp, the registry could simply refuse to hand back an entry older than some configurable age, which would stop a removed slave from silently propping up the aggregate risk figure after its last message ages out.

Message acknowledgment with retry would harden delivery further, and the MSG_ACK type already sits in the enum ready for exactly this purpose. A slave could hold onto a sent message until it receives an acknowledgment carrying the matching sequence number, resending after a timeout if none arrives, which would turn the current fire-and-forget exchange into something closer to an at-least-once delivery guarantee.

Directed responses to a specific slave, rather than the same broadcast figure sent to everyone, would be a small change built entirely on information the broker already has. Since the registry knows every sender's magic number, a response could carry a routing field naming its intended recipient, with slaves simply ignoring anything not addressed to them.

Adding new message types is the safest extension of all, precisely because the enum and the fixed struct layout were built with exactly that in mind. A new entry in ENUM_MESSAGE_TYPE does not disturb any existing slave, since older code simply never sends or expects it, so long as the struct's overall size stays fixed and every participant on the bus still agrees on the same byte layout.

Section 12 — Limitations

The design here is deliberately simple, and its boundaries are worth stating just as plainly as its guarantees.

Reading and writing a message, once a client is actually attached, still happens through ordinary blocking calls. That is a reasonable trade, since the client on the other end is already about to send the moment it connects, but it does mean a genuinely misbehaving client that connects and then never writes anything could, in principle, hold up one timer tick longer than expected. In this implementation the risk is bounded in practice, because a slave that fails to send simply closes its own connection and retries later, rather than sitting on the pipe indefinitely.

The pipe name resolves only to the local machine. The \\.\pipe\ prefix names an endpoint on the host running the terminal, and reaching a broker on a separate machine would call for a full network path together with the security configuration that comes with it, neither of which this implementation attempts.

There is no reconnection logic on the broker's side of the relationship. If the broker Expert Advisor restarts, its pipe endpoint is rebuilt from scratch, and while the slaves keep retrying their own connections tick after tick and recover on their own, any transaction genuinely in flight at the moment of a restart is simply lost rather than replayed.

Finally, the CMessage layout is a shared contract for all Expert Advisors on the bus. The struct's size and field order have to stay fixed, and if the schema ever changes, every participant that speaks the bus needs recompiling together. A broker expecting fifty-two-byte messages and a slave sending fifty-six-byte messages will simply fail to interoperate, so schema changes remain an all-at-once operation rather than a gradual one.

Conclusion

This bus gives multiple Expert Advisors a structured way to share computed state, replacing loosely named global variables with typed messages whose schema lives in code. A broker owns the server, registry, aggregator, and dashboard; three slaves connect as clients; and every exchange returns a portfolio risk figure correctly attributed to the symbol that caused it. What it guarantees is concrete: an explicit type on every message, a fixed layout both ends agree on, and freshness derived directly from the wire data. What it does not cover is just as clear: no broker-side reconnection, no cross-machine reach, and a schema that binds every participant at once. It is a clean, honest local bus, not a general-purpose messaging fabric.

Programs used in the article:

#	Name	Type	Description
1	MessageType.mqh	Include File	The ENUM_MESSAGE_TYPE enum naming every valid message category
2	Message.mqh	Include File	The CMessage struct with fixed 52-byte Serialize() and Deserialize()
3	PipeServer.mqh	Include File	CPipeServer, with a non-blocking accept poll
4	PipeClient.mqh	Include File	CPipeClient wrapping the pipe client API for a slave EA
5	MessageRegistry.mqh	Include File	CMessageRegistry storing state by type and by sender, with receipt timing
6	RiskAggregator.mqh	Include File	CRiskAggregator computing risk and its per-symbol breakdown
7	PipeBrokerDashboard.mqh	Include File	CPipeBrokerDashboard rendering ACTIVE/SILENT sender status
8	CPipeBrokerEA.mq5	Demo EA	The broker EA owning the server, registry, aggregator, and dashboard
9	CSlaveEA.mq5	Demo EA	The slave EA template connecting as a pipe client, parameterized by magic number
10	TestMessageBus.mq5	Script	Verification script covering serialization, registry, and aggregation
11	MessageBusEngine.zip	Zip Archive	Zip archive containing all the attached files and their paths relative to the terminal's root folder.