809 lines
27 KiB
MQL5
809 lines
27 KiB
MQL5
 | //+------------------------------------------------------------------+
| ||
//| Zmtp.mqh |
| |||
//| Copyright 2025, MetaQuotes Ltd. |
| |||
//| https://www.mql5.com |
| |||
//+------------------------------------------------------------------+
| |||
#property copyright "Copyright 2025, MetaQuotes Ltd."
| |||
#property link "https://www.mql5.com"
| |||
//+------------------------------------------------------------------+
| |||
//| Zmtp.mqh |
| |||
//| Minimal ZMTP 3.0 client transport for MetaTrader 5. |
| |||
//| |
| |||
//| Implements: |
| |||
//| - ZMTP 3.0 greeting + framing (RFC 23) |
| |||
//| - NULL and PLAIN security mechanisms (RFC 24) |
| |||
//| - CZmqReqSocket : REQ/REP pattern |
| |||
//| |
| |||
//| CURVE security and the other socket-pattern classes are NOT |
| |||
//| implemented yet |
| |||
//+------------------------------------------------------------------+
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| Socket-Type property this library presents to the ZMTP peer |
| |||
//+------------------------------------------------------------------+
| |||
enum ENUM_ZMQ_SOCKET_TYPE
| |||
{
| |||
ZMQ_TYPE_REQ,
| |||
ZMQ_TYPE_REP,
| |||
ZMQ_TYPE_DEALER,
| |||
ZMQ_TYPE_ROUTER,
| |||
ZMQ_TYPE_PUB,
| |||
ZMQ_TYPE_SUB,
| |||
ZMQ_TYPE_PUSH,
| |||
ZMQ_TYPE_PULL,
| |||
ZMQ_TYPE_PAIR
| |||
};
| |||
| |||
enum ENUM_ZMTP_SECURITY
| |||
{
| |||
ZMTP_SEC_NULL, // no authentication, no encryption
| |||
ZMTP_SEC_PLAIN, // username/password, sent in clear text
| |||
ZMTP_SEC_CURVE // NOT IMPLEMENTED - see note at bottom of file
| |||
};
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
string ZmqSocketTypeName(ENUM_ZMQ_SOCKET_TYPE t)
| |||
{
| |||
switch(t)
| |||
{
| |||
case ZMQ_TYPE_REQ:
| |||
return "REQ";
| |||
case ZMQ_TYPE_REP:
| |||
return "REP";
| |||
case ZMQ_TYPE_DEALER:
| |||
return "DEALER";
| |||
case ZMQ_TYPE_ROUTER:
| |||
return "ROUTER";
| |||
case ZMQ_TYPE_PUB:
| |||
return "PUB";
| |||
case ZMQ_TYPE_SUB:
| |||
return "SUB";
| |||
case ZMQ_TYPE_PUSH:
| |||
return "PUSH";
| |||
case ZMQ_TYPE_PULL:
| |||
return "PULL";
| |||
case ZMQ_TYPE_PAIR:
| |||
return "PAIR";
| |||
}
| |||
return "";
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| One ZMTP frame as read off / written to a node |
| |||
//+------------------------------------------------------------------+
| |||
struct ZmtpFrame
| |||
{
| |||
bool more; // MORE flag - another frame follows in this message
| |||
bool command; // COMMAND flag - control frame (handshake), not user data
| |||
uchar data[];
| |||
};
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| Appends a ZMTP command-property (name/value pair) to a buffer. |
| |||
//| Wire format: name-len(1) name value-len(4,BE) value |
| |||
//+------------------------------------------------------------------+
| |||
void ZmtpAppendProperty(uchar &buf[], string name, const uchar &value[])
| |||
{
| |||
uchar name_bytes[];
| |||
StringToCharArray(name, name_bytes, 0, StringLen(name));
| |||
int nlen = ArraySize(name_bytes);
| |||
int vlen = ArraySize(value);
| |||
int old = ArraySize(buf);
| |||
ArrayResize(buf, old + 1 + nlen + 4 + vlen);
| |||
buf[old] = (uchar)nlen;
| |||
for(int i=0; i<nlen; i++)
| |||
buf[old+1+i] = name_bytes[i];
| |||
int off = old + 1 + nlen;
| |||
buf[off+0] = (uchar)((vlen>>24)&0xFF);
| |||
buf[off+1] = (uchar)((vlen>>16)&0xFF);
| |||
buf[off+2] = (uchar)((vlen>>8)&0xFF);
| |||
buf[off+3] = (uchar)(vlen&0xFF);
| |||
for(int i=0; i<vlen; i++)
| |||
buf[off+4+i] = value[i];
| |||
}
| |||
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| Appends src onto the end of dst - used to reassemble a multipart |
| |||
//| ZMTP message into one flat byte buffer. |
| |||
//+------------------------------------------------------------------+
| |||
void ZmtpAppendBytes(uchar &dst[], const uchar &src[])
| |||
{
| |||
int old = ArraySize(dst);
| |||
int add = ArraySize(src);
| |||
ArrayResize(dst, old+add);
| |||
ArrayCopy(dst, src, old, 0, add);
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| CZmtpTransport |
| |||
//| Low-level ZMTP 3.0 connection: raw socket, receive buffering, |
| |||
//| greeting exchange, security handshake, frame read/write. |
| |||
//| Every socket-pattern class (REQ, and later DEALER/PUB/SUB/etc.) |
| |||
//| is built on top of one of these. |
| |||
//+------------------------------------------------------------------+
| |||
class CZmtpTransport
| |||
{
| |||
private:
| |||
int m_socket;
| |||
uchar m_rxbuf[];
| |||
int m_rxlen; // valid bytes currently in m_rxbuf
| |||
int m_rxpos; // read cursor into m_rxbuf
| |||
int m_timeout_ms;
| |||
ENUM_ZMTP_SECURITY m_security;
| |||
string m_username;
| |||
string m_password;
| |||
| |||
bool EnsureBytes(int n);
| |||
void PopBytes(int n, uchar &out[]);
| |||
bool RawSend(const uchar &buf[], int len);
| |||
| |||
bool SendGreeting();
| |||
bool RecvGreeting(string &peer_mechanism);
| |||
bool DoPlainHandshake();
| |||
bool SendReady(ENUM_ZMQ_SOCKET_TYPE stype, string identity);
| |||
bool RecvReady();
| |||
| |||
public:
| |||
CZmtpTransport();
| |||
~CZmtpTransport();
| |||
| |||
bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec, int timeout_ms=5000);
| |||
void Disconnect();
| |||
bool IsConnected();
| |||
| |||
void SetPlainCredentials(string user, string pass) { m_username=user; m_password=pass; }
| |||
| |||
// Full ZMTP handshake: greeting -> (security mechanism) -> READY/READY.
| |||
// Call once, immediately after Connect().
| |||
bool Handshake(ENUM_ZMQ_SOCKET_TYPE stype, string identity="");
| |||
| |||
bool SendFrame(const uchar &data[], bool more, bool command=false);
| |||
bool SendCommand(string name, const uchar &body[]);
| |||
bool RecvFrame(ZmtpFrame &frame);
| |||
};
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
CZmtpTransport::CZmtpTransport()
| |||
{
| |||
m_socket = INVALID_HANDLE;
| |||
m_rxlen = 0;
| |||
m_rxpos = 0;
| |||
m_timeout_ms = 5000;
| |||
m_security = ZMTP_SEC_NULL;
| |||
m_username = "";
| |||
m_password = "";
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
CZmtpTransport::~CZmtpTransport()
| |||
{
| |||
Disconnect();
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::Connect(string host, int port, ENUM_ZMTP_SECURITY sec, int timeout_ms=5000)
| |||
{
| |||
m_security = sec;
| |||
m_timeout_ms = timeout_ms;
| |||
m_rxlen = 0;
| |||
m_rxpos = 0;
| |||
ArrayResize(m_rxbuf, 0);
| |||
| |||
m_socket = SocketCreate();
| |||
if(m_socket == INVALID_HANDLE)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: SocketCreate failed, error ", GetLastError());
| |||
return false;
| |||
}
| |||
if(!SocketConnect(m_socket, host, (uint)port, (uint)timeout_ms))
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: SocketConnect to ", host, ":", port, " failed, error ", GetLastError());
| |||
SocketClose(m_socket);
| |||
m_socket = INVALID_HANDLE;
| |||
return false;
| |||
}
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
void CZmtpTransport::Disconnect()
| |||
{
| |||
if(m_socket != INVALID_HANDLE)
| |||
{
| |||
SocketClose(m_socket);
| |||
m_socket = INVALID_HANDLE;
| |||
}
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::IsConnected()
| |||
{
| |||
return (m_socket != INVALID_HANDLE) && SocketIsConnected(m_socket);
| |||
}
| |||
| |||
//--- send every byte, looping past partial sends -------------------
| |||
bool CZmtpTransport::RawSend(const uchar &buf[], int len)
| |||
{
| |||
int sent_total = 0;
| |||
uchar chunk[];
| |||
uint start = GetTickCount();
| |||
while(sent_total < len)
| |||
{
| |||
int remaining = len - sent_total;
| |||
ArrayResize(chunk, remaining);
| |||
ArrayCopy(chunk, buf, 0, sent_total, remaining);
| |||
int sent = SocketSend(m_socket, chunk, remaining);
| |||
if(sent <= 0)
| |||
{
| |||
if(!SocketIsConnected(m_socket))
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: connection dropped mid-send");
| |||
return false;
| |||
}
| |||
if((int)(GetTickCount()-start) > m_timeout_ms)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: send timed out");
| |||
return false;
| |||
}
| |||
Sleep(1);
| |||
continue;
| |||
}
| |||
sent_total += sent;
| |||
start = GetTickCount(); // reset timeout on forward progress
| |||
}
| |||
return true;
| |||
}
| |||
| |||
//--- block (up to m_timeout_ms) until n bytes are buffered ---------
| |||
bool CZmtpTransport::EnsureBytes(int n)
| |||
{
| |||
uint start = GetTickCount();
| |||
while((m_rxlen - m_rxpos) < n)
| |||
{
| |||
if(!SocketIsConnected(m_socket))
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: connection dropped mid-receive");
| |||
return false;
| |||
}
| |||
uint avail = SocketIsReadable(m_socket);
| |||
if(avail > 0)
| |||
{
| |||
uchar tmp[];
| |||
ArrayResize(tmp, avail);
| |||
int got = SocketRead(m_socket, tmp, avail, 100);
| |||
if(got > 0)
| |||
{
| |||
int need = m_rxlen + got;
| |||
if(need > ArraySize(m_rxbuf))
| |||
ArrayResize(m_rxbuf, need);
| |||
for(int i=0; i<got; i++)
| |||
m_rxbuf[m_rxlen+i] = tmp[i];
| |||
m_rxlen += got;
| |||
start = GetTickCount(); // reset timeout on forward progress
| |||
}
| |||
}
| |||
else
| |||
Sleep(1);
| |||
| |||
/*if((int)(GetTickCount()-start) > m_timeout_ms)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: receive timed out waiting for ", n, " bytes");
| |||
return false;
| |||
}*/
| |||
}
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
void CZmtpTransport::PopBytes(int n, uchar &out[])
| |||
{
| |||
ArrayResize(out, n);
| |||
for(int i=0; i<n; i++)
| |||
out[i] = m_rxbuf[m_rxpos+i];
| |||
m_rxpos += n;
| |||
// compact the buffer once we've drained a meaningful chunk of it
| |||
if(m_rxpos > 4096)
| |||
{
| |||
int remain = m_rxlen - m_rxpos;
| |||
for(int i=0; i<remain; i++)
| |||
m_rxbuf[i] = m_rxbuf[m_rxpos+i];
| |||
m_rxlen = remain;
| |||
m_rxpos = 0;
| |||
}
| |||
}
| |||
| |||
//--- 64-byte ZMTP greeting -------------------------------------------
| |||
bool CZmtpTransport::SendGreeting()
| |||
{
| |||
uchar g[];
| |||
ArrayResize(g, 64);
| |||
ArrayFill(g, 0, 64, 0);
| |||
g[0] = 0xFF; // signature start
| |||
g[9] = 0x7F; // signature end
| |||
g[10] = 3; // version-major = 3 (ZMTP 3.x)
| |||
g[11] = 0; // version-minor
| |||
| |||
string mech = "NULL";
| |||
if(m_security == ZMTP_SEC_PLAIN)
| |||
mech = "PLAIN";
| |||
else
| |||
if(m_security == ZMTP_SEC_CURVE)
| |||
mech = "CURVE";
| |||
uchar mech_bytes[];
| |||
StringToCharArray(mech, mech_bytes, 0, StringLen(mech));
| |||
for(int i=0; i<ArraySize(mech_bytes) && i<20; i++)
| |||
g[12+i] = mech_bytes[i];
| |||
| |||
g[32] = 0; // as-server: this library only ever connects as a client
| |||
// g[33..63] filler, already zero
| |||
return RawSend(g, 64);
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::RecvGreeting(string &peer_mechanism)
| |||
{
| |||
if(!EnsureBytes(64))
| |||
return false;
| |||
uchar g[];
| |||
PopBytes(64, g);
| |||
if(g[0] != 0xFF || g[9] != 0x7F)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: peer sent an invalid greeting signature - is this really a ZMTP 3.x endpoint?");
| |||
return false;
| |||
}
| |||
int major = g[10];
| |||
if(major < 3)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: peer speaks ZMTP ", major, ".x - this library only supports ZMTP 3.x");
| |||
return false;
| |||
}
| |||
uchar mech_bytes[20];
| |||
for(int i=0; i<20; i++)
| |||
mech_bytes[i] = g[12+i];
| |||
int mlen = 0;
| |||
while(mlen < 20 && mech_bytes[mlen] != 0)
| |||
mlen++;
| |||
peer_mechanism = CharArrayToString(mech_bytes, 0, mlen);
| |||
return true;
| |||
}
| |||
| |||
//--- frame I/O ---------------------------------------------------------
| |||
bool CZmtpTransport::SendFrame(const uchar &data[], bool more, bool command=false)
| |||
{
| |||
int len = ArraySize(data);
| |||
uchar flags = 0;
| |||
if(more)
| |||
flags |= 0x01;
| |||
bool long_form = (len > 255);
| |||
if(long_form)
| |||
flags |= 0x02;
| |||
if(command)
| |||
flags |= 0x04;
| |||
| |||
uchar header[];
| |||
if(!long_form)
| |||
{
| |||
ArrayResize(header, 2);
| |||
header[0] = flags;
| |||
header[1] = (uchar)len;
| |||
}
| |||
else
| |||
{
| |||
ArrayResize(header, 9);
| |||
header[0] = flags;
| |||
ulong ulen = (ulong)len;
| |||
for(int i=0; i<8; i++)
| |||
header[1+i] = (uchar)((ulen >> (8*(7-i))) & 0xFF);
| |||
}
| |||
| |||
int hlen = ArraySize(header);
| |||
uchar packet[];
| |||
ArrayResize(packet, hlen+len);
| |||
ArrayCopy(packet, header, 0, 0, hlen);
| |||
if(len > 0)
| |||
ArrayCopy(packet, data, hlen, 0, len);
| |||
| |||
return RawSend(packet, ArraySize(packet));
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::RecvFrame(ZmtpFrame &frame)
| |||
{
| |||
if(!EnsureBytes(1))
| |||
return false;
| |||
uchar hdr1[];
| |||
PopBytes(1, hdr1);
| |||
uchar flags = hdr1[0];
| |||
frame.more = (flags & 0x01) != 0;
| |||
frame.command = (flags & 0x04) != 0;
| |||
bool long_form = (flags & 0x02) != 0;
| |||
| |||
ulong len = 0;
| |||
if(!long_form)
| |||
{
| |||
uchar l[];
| |||
if(!EnsureBytes(1))
| |||
return false;
| |||
PopBytes(1, l);
| |||
len = l[0];
| |||
}
| |||
else
| |||
{
| |||
uchar l[];
| |||
if(!EnsureBytes(8))
| |||
return false;
| |||
PopBytes(8, l);
| |||
for(int i=0; i<8; i++)
| |||
len = (len << 8) | l[i];
| |||
}
| |||
| |||
if(len > 0)
| |||
{
| |||
if(!EnsureBytes((int)len))
| |||
return false;
| |||
PopBytes((int)len, frame.data);
| |||
}
| |||
else
| |||
ArrayResize(frame.data, 0);
| |||
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::SendCommand(string name, const uchar &body[])
| |||
{
| |||
uchar name_bytes[];
| |||
StringToCharArray(name, name_bytes, 0, StringLen(name));
| |||
int nlen = ArraySize(name_bytes);
| |||
uchar frame[];
| |||
ArrayResize(frame, 1+nlen+ArraySize(body));
| |||
frame[0] = (uchar)nlen;
| |||
for(int i=0; i<nlen; i++)
| |||
frame[1+i] = name_bytes[i];
| |||
for(int i=0; i<ArraySize(body); i++)
| |||
frame[1+nlen+i] = body[i];
| |||
return SendFrame(frame, false, true);
| |||
}
| |||
| |||
//--- PLAIN mechanism: HELLO(user,pass) -> WELCOME, then READY/READY --
| |||
bool CZmtpTransport::DoPlainHandshake()
| |||
{
| |||
uchar body[];
| |||
uchar uv[];
| |||
StringToCharArray(m_username, uv, 0, StringLen(m_username));
| |||
uchar pv[];
| |||
StringToCharArray(m_password, pv, 0, StringLen(m_password));
| |||
ZmtpAppendProperty(body, "Username", uv);
| |||
ZmtpAppendProperty(body, "Password", pv);
| |||
if(!SendCommand("HELLO", body))
| |||
return false;
| |||
| |||
ZmtpFrame f;
| |||
if(!RecvFrame(f))
| |||
return false;
| |||
if(!f.command)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp/PLAIN: expected a WELCOME command, got a data frame");
| |||
return false;
| |||
}
| |||
int nlen = f.data[0];
| |||
string cname = CharArrayToString(f.data, 1, nlen);
| |||
if(cname == "ERROR")
| |||
{
| |||
Print(__FUNCTION__," : Zmtp/PLAIN: server rejected the supplied username/password");
| |||
return false;
| |||
}
| |||
if(cname != "WELCOME")
| |||
{
| |||
Print(__FUNCTION__," : Zmtp/PLAIN: expected WELCOME, got '", cname, "'");
| |||
return false;
| |||
}
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::SendReady(ENUM_ZMQ_SOCKET_TYPE stype, string identity)
| |||
{
| |||
uchar body[];
| |||
ArrayResize(body, 0);
| |||
uchar stv[];
| |||
string sname = ZmqSocketTypeName(stype);
| |||
StringToCharArray(sname, stv, 0, StringLen(sname));
| |||
ZmtpAppendProperty(body, "Socket-Type", stv);
| |||
if(identity != "")
| |||
{
| |||
uchar idv[];
| |||
StringToCharArray(identity, idv, 0, StringLen(identity));
| |||
ZmtpAppendProperty(body, "Identity", idv);
| |||
}
| |||
return SendCommand("READY", body);
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::RecvReady()
| |||
{
| |||
ZmtpFrame f;
| |||
if(!RecvFrame(f))
| |||
return false;
| |||
if(!f.command)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: expected a READY command, got a data frame");
| |||
return false;
| |||
}
| |||
int nlen = f.data[0];
| |||
string cname = CharArrayToString(f.data, 1, nlen);
| |||
if(cname == "ERROR")
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: peer sent ERROR during handshake");
| |||
return false;
| |||
}
| |||
if(cname != "READY")
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: expected READY, got '", cname, "'");
| |||
return false;
| |||
}
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| |
| |||
//+------------------------------------------------------------------+
| |||
bool CZmtpTransport::Handshake(ENUM_ZMQ_SOCKET_TYPE stype, string identity="")
| |||
{
| |||
if(m_security == ZMTP_SEC_CURVE)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: ZMTP_SEC_CURVE is not implemented yet - see the note at the bottom of Zmtp.mqh");
| |||
return false;
| |||
}
| |||
| |||
if(!SendGreeting())
| |||
return false;
| |||
string peer_mech;
| |||
if(!RecvGreeting(peer_mech))
| |||
return false;
| |||
| |||
string want_mech = (m_security == ZMTP_SEC_PLAIN) ? "PLAIN" : "NULL";
| |||
if(peer_mech != want_mech)
| |||
{
| |||
Print(__FUNCTION__," : Zmtp: security mismatch - this socket requested ", want_mech,
| |||
" but the peer offered '", peer_mech, "'");
| |||
return false;
| |||
}
| |||
| |||
if(m_security == ZMTP_SEC_PLAIN)
| |||
if(!DoPlainHandshake())
| |||
return false;
| |||
| |||
if(!SendReady(stype, identity))
| |||
return false;
| |||
if(!RecvReady())
| |||
return false;
| |||
| |||
return true;
| |||
}
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| CZmqReqSocket - ZMQ REQ pattern over ZMTP. |
| |||
//| |
| |||
//| Enforces the strict send/recv/send/recv alternation that REQ |
| |||
//| requires, and handles the empty delimiter-frame envelope that |
| |||
//| REQ/REP sockets use automatically in real ZeroMQ. |
| |||
//+------------------------------------------------------------------+
| |||
class CZmqReqSocket
| |||
{
| |||
private:
| |||
CZmtpTransport m_t;
| |||
bool m_awaiting_reply;
| |||
| |||
public:
| |||
CZmqReqSocket() { m_awaiting_reply = false; }
| |||
| |||
bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec=ZMTP_SEC_NULL,
| |||
int timeout_ms=5000, string identity="", string plain_user="", string plain_pass="")
| |||
{
| |||
if(sec == ZMTP_SEC_PLAIN)
| |||
m_t.SetPlainCredentials(plain_user, plain_pass);
| |||
if(!m_t.Connect(host, port, sec, timeout_ms))
| |||
return false;
| |||
return m_t.Handshake(ZMQ_TYPE_REQ, identity);
| |||
}
| |||
| |||
void Disconnect() { m_t.Disconnect(); }
| |||
bool IsConnected() { return m_t.IsConnected(); }
| |||
| |||
// Single-part string request.
| |||
bool Send(const string &request)
| |||
{
| |||
uchar body[];
| |||
StringToCharArray(request, body, 0, StringLen(request));
| |||
return SendRaw(body);
| |||
}
| |||
| |||
// Single-part binary request (e.g. a packed struct of doubles).
| |||
bool SendRaw(const uchar &request[])
| |||
{
| |||
if(m_awaiting_reply)
| |||
{
| |||
Print(__FUNCTION__," : Zmq REQ: Send() called out of order - a reply is still pending");
| |||
return false;
| |||
}
| |||
uchar empty[];
| |||
ArrayResize(empty, 0);
| |||
// REQ envelope: empty delimiter frame, then the request body
| |||
if(!m_t.SendFrame(empty, true))
| |||
return false;
| |||
if(!m_t.SendFrame(request, false))
| |||
return false;
| |||
m_awaiting_reply = true;
| |||
return true;
| |||
}
| |||
| |||
// Blocks for the matching reply (up to the connect-time timeout).
| |||
bool Recv(string &reply)
| |||
{
| |||
uchar raw[];
| |||
if(!RecvRaw(raw))
| |||
return false;
| |||
reply = CharArrayToString(raw, 0, ArraySize(raw));
| |||
return true;
| |||
}
| |||
| |||
bool RecvRaw(uchar &reply[])
| |||
{
| |||
if(!m_awaiting_reply)
| |||
{
| |||
Print(__FUNCTION__," : Zmq REQ: Recv() called before Send()");
| |||
return false;
| |||
}
| |||
ZmtpFrame f;
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
// If a ROUTER sits between us and the REP, extra identity/envelope
| |||
// frames may precede the empty delimiter - skip anything non-empty.
| |||
while(ArraySize(f.data) > 0 && f.more)
| |||
{
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
}
| |||
ArrayResize(reply, 0);
| |||
while(f.more)
| |||
{
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
int old = ArraySize(reply);
| |||
int add = ArraySize(f.data);
| |||
ArrayResize(reply, old+add);
| |||
ArrayCopy(reply, f.data, old, 0, add);
| |||
}
| |||
m_awaiting_reply = false;
| |||
return true;
| |||
}
| |||
};
| |||
| |||
//+------------------------------------------------------------------+
| |||
//| CZmqSubSocket - ZMQ SUB pattern over ZMTP. |
| |||
//| Receive-only, pairs with a PUB peer. Per ZMTP, subscriptions are |
| |||
//| sent to the peer as ordinary data frames (not COMMAND frames): |
| |||
//| a leading 0x01 (SUBSCRIBE) or 0x00 (UNSUBSCRIBE) byte followed by |
| |||
//| the topic-prefix bytes - filtering happens on the PUB side. |
| |||
//+------------------------------------------------------------------+
| |||
class CZmqSubSocket
| |||
{
| |||
private:
| |||
CZmtpTransport m_t;
| |||
| |||
bool SendSubscription(const string &prefix, uchar marker)
| |||
{
| |||
uchar p[];
| |||
StringToCharArray(prefix, p, 0, StringLen(prefix));
| |||
uchar frame[];
| |||
ArrayResize(frame, 1+ArraySize(p));
| |||
frame[0] = marker;
| |||
for(int i=0; i<ArraySize(p); i++)
| |||
frame[1+i] = p[i];
| |||
return m_t.SendFrame(frame, false);
| |||
}
| |||
| |||
public:
| |||
CZmqSubSocket(void) { }
| |||
bool Connect(string host, int port, ENUM_ZMTP_SECURITY sec=ZMTP_SEC_NULL,
| |||
int timeout_ms=5000, string identity="", string plain_user="", string plain_pass="")
| |||
{
| |||
if(sec == ZMTP_SEC_PLAIN)
| |||
m_t.SetPlainCredentials(plain_user, plain_pass);
| |||
if(!m_t.Connect(host, port, sec, timeout_ms))
| |||
return false;
| |||
return m_t.Handshake(ZMQ_TYPE_SUB, identity);
| |||
}
| |||
| |||
void Disconnect() { m_t.Disconnect(); }
| |||
bool IsConnected() { return m_t.IsConnected(); }
| |||
| |||
// Subscribe to any topic starting with `prefix` - pass "" to receive
| |||
// everything the publisher sends. Call before/after connecting; must
| |||
// be called at least once or a compliant PUB peer will send nothing.
| |||
bool Subscribe(const string &prefix) { return SendSubscription(prefix, 0x01); }
| |||
bool Unsubscribe(const string &prefix) { return SendSubscription(prefix, 0x00); }
| |||
| |||
// Blocks for the next published message, matching CZmqPubSocket::Publish's
| |||
// two-frame [topic][payload] convention.
| |||
bool Recv(string &topic, string &message)
| |||
{
| |||
ZmtpFrame f;
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
topic = CharArrayToString(f.data, 0, ArraySize(f.data));
| |||
uchar raw[];
| |||
while(f.more)
| |||
{
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
ZmtpAppendBytes(raw, f.data);
| |||
}
| |||
message = CharArrayToString(raw, 0, ArraySize(raw));
| |||
return true;
| |||
}
| |||
| |||
bool RecvRaw(uchar &topic[], uchar &message[])
| |||
{
| |||
ZmtpFrame f;
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
ArrayResize(topic, ArraySize(f.data));
| |||
ArrayCopy(topic, f.data, 0, 0, ArraySize(f.data));
| |||
ArrayResize(message, 0);
| |||
while(f.more)
| |||
{
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
ZmtpAppendBytes(message, f.data);
| |||
}
| |||
return true;
| |||
}
| |||
| |||
// For peers using the single-frame (topic-prefixed) publish convention
| |||
// instead of the two-frame one - returns the frame's raw content
| |||
// unsplit, so you can peel the prefix off yourself.
| |||
bool RecvSingleFrame(string &data)
| |||
{
| |||
uchar raw[];
| |||
ZmtpFrame f;
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
ZmtpAppendBytes(raw, f.data);
| |||
while(f.more)
| |||
{
| |||
if(!m_t.RecvFrame(f))
| |||
return false;
| |||
ZmtpAppendBytes(raw, f.data);
| |||
}
| |||
data = CharArrayToString(raw, 0, ArraySize(raw));
| |||
return true;
| |||
}
| |||
};
| |||
//+------------------------------------------------------------------+
|