11 Commits

Author SHA1 Message Date
Ding Li afab5f93cc Released 1.5
This release contains important API changes that may break existing
code.

1. Incompatible change: `Socket.send` series methods are splitted
to `Socket.send` and `Socket.sendMore` and have more overloads for
common message types like string or empty
2. Remove PollItem duplicate API (#11)
3. Fix compiler warning (#10) and compile failure (#12)
4. Add RTReq example from ZMQ Guide Chapter 3
2017-10-28 22:45:23 +08:00
Ding Li 795acc13c8 Fix #12 (#1, #4, #9): SocketOptions curve key argument causes compile error 2017-10-23 22:13:23 +08:00
Ding Li 184b9f6d8b Merge branch 'master' of github.com:dingmaotu/mql-zmq 2017-10-23 10:22:36 +08:00
Ding Li 96fd26ad85 Fix #11: remove duplicate and deprecated API Socket::register 2017-10-23 10:21:36 +08:00
Ding Li 1d6b1aa2d5 Fix #11: remove duplicate and deprecated API Socket::register 2017-10-23 09:50:25 +08:00
Ding Li 07414d6300 Fix #10: fix size parameter type on 64bit Terminal 2017-10-17 09:29:06 +08:00
Ding Li 8475ecc03a Fix #8 by importing changes from mql4-lib (eliminating WinUser32.mqh dependency) 2017-09-13 11:52:21 +08:00
Ding Li 1d127c66b7 Merge pull request #7 from yerden/patch-1
zmq: fix typo in SendTimeout method
2017-09-08 08:31:13 +08:00
yerden 101807d69e zmq: fix typo in SendTimeout method 2017-09-08 00:27:47 +06:00
Ding Li cd7e5f0e47 Only resize target array when the array size is smaller than ZmqMsg 2017-08-30 10:15:46 +08:00
Ding Li bab1c9753a Fix documentation reference to a mql4-lib file 2017-08-18 10:34:30 +08:00
9 changed files with 202 additions and 132 deletions
-43
View File
@@ -45,28 +45,6 @@
#define size_t int
#endif
//--- _WIN32_WINNT version constants
#define _WIN32_WINNT_NT4 0x0400 // Windows NT 4.0
#define _WIN32_WINNT_WIN2K 0x0500 // Windows 2000
#define _WIN32_WINNT_WINXP 0x0501 // Windows XP
#define _WIN32_WINNT_WS03 0x0502 // Windows Server 2003
#define _WIN32_WINNT_WIN6 0x0600 // Windows Vista
#define _WIN32_WINNT_VISTA 0x0600 // Windows Vista
#define _WIN32_WINNT_WS08 0x0600 // Windows Server 2008
#define _WIN32_WINNT_LONGHORN 0x0600 // Windows Vista
#define _WIN32_WINNT_WIN7 0x0601 // Windows 7
#define _WIN32_WINNT_WIN8 0x0602 // Windows 8
#define _WIN32_WINNT_WINBLUE 0x0603 // Windows 8.1
#define _WIN32_WINNT_WINTHRESHOLD 0x0A00 // Windows 10
#define _WIN32_WINNT_WIN10 0x0A00 // Windows 10
//--- define you own for your target platform
#define WINVER 0x0A00
#define _WIN32_WINNT 0x0A00
#define FORMAT_MESSAGE_FROM_SYSTEM 0x00001000
#define FORMAT_MESSAGE_IGNORE_INSERTS 0x00000200
#import "kernel32.dll"
void RtlMoveMemory(intptr_t dest,const uchar &array[],size_t length);
void RtlMoveMemory(uchar &array[],intptr_t src,size_t length);
@@ -83,18 +61,7 @@ int MultiByteToWideChar(uint codePage,
string &str,
int length
);
uint FormatMessageW(uint dwFlags,
intptr_t lpSource,
uint dwMessageId,
uint dwLanguageId,
ushort &buffer[],
uint nSize,
intptr_t Arguments
);
#import
//--- This is a standard header of the official MetaTrader distribution
#include <WinUser32.mqh>
//+------------------------------------------------------------------+
//| Copy the memory contents pointed by src to array |
//| array parameter should be initialized to the desired size |
@@ -188,13 +155,3 @@ void StringToUtf8(const string str,uchar &utf8[],bool ending=true)
StringToCharArray(str,utf8,0,count,CP_UTF8);
}
//+------------------------------------------------------------------+
//| Get system defined error code message |
//+------------------------------------------------------------------+
string GetErrorMessage(int errorCode)
{
static ushort buffer[64*1024];
FormatMessageW(FORMAT_MESSAGE_FROM_SYSTEM|FORMAT_MESSAGE_IGNORE_INSERTS,
0,errorCode,0,buffer,ArraySize(buffer),0);
return ShortArrayToString(buffer);
}
//+------------------------------------------------------------------+
+24 -76
View File
@@ -133,16 +133,32 @@ public:
bool connect(string addr);
bool disconnect(string addr);
//--- send and receive packets
bool recv(uchar &buf[],bool nowait=false);
bool send(const uchar &buf[],bool nowait=false,bool more=false);
bool sendConst(const uchar &buf[],bool nowait=false,bool more=false);
//--- send raw bytes
bool send(const uchar &buf[],bool nowait=false) {return -1!=zmq_send(m_ref,buf,ArraySize(buf),nowait?ZMQ_DONTWAIT:0);}
bool sendConst(const uchar &buf[],bool nowait=false) {return -1!=zmq_send_const(m_ref,buf,ArraySize(buf),nowait?ZMQ_DONTWAIT:0);}
bool send(ZmqMsg &msg,bool nowait=false,bool more=false);
bool recv(ZmqMsg &msg,bool nowait=false);
//--- send ZmqMsg
bool send(ZmqMsg &msg,bool nowait=false) {return -1!=zmq_msg_send(msg,m_ref,nowait?ZMQ_DONTWAIT:0);}
//--- send string
bool send(string msg,bool nowait=false) {ZmqMsg m(msg); return send(m,nowait);}
//--- send empty
bool send(bool nowait=false) {ZmqMsg m; return send(m,nowait);}
void register(PollItem &pollitem,bool read=false,bool write=false);
void register(PollItem &pollitems[],int index,bool read=false,bool write=false);
//--- same as above 5 but for multipart messages
bool sendMore(const uchar &buf[],bool nowait=false);
bool sendConstMore(const uchar &buf[],bool nowait=false);
bool sendMore(ZmqMsg &msg,bool nowait=false)
{
int flags=ZMQ_SNDMORE;
if(nowait) flags|=ZMQ_DONTWAIT;
return -1!=zmq_msg_send(msg,m_ref,flags);
}
bool sendMore(string msg,bool nowait=false) {ZmqMsg m(msg); return sendMore(m,nowait);}
bool sendMore(bool nowait=false) {ZmqMsg m; return sendMore(m,nowait);}
//--- receive packets
bool recv(uchar &buf[],bool nowait=false) {return -1!=zmq_recv(m_ref,buf,ArraySize(buf),nowait?ZMQ_DONTWAIT:0);}
bool recv(ZmqMsg &msg,bool nowait=false) {return -1!=zmq_msg_recv(msg,m_ref,nowait?ZMQ_DONTWAIT:0);}
//--- monitor socket events
bool monitor(string addr,int events);
@@ -204,54 +220,6 @@ bool Socket::disconnect(string addr)
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
bool Socket::recv(uchar &buf[],bool nowait=false)
{
int options=0;
if(nowait) options|=ZMQ_DONTWAIT;
return -1!=zmq_recv(m_ref,buf,ArraySize(buf),options);
}
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
bool Socket::send(const uchar &buf[],bool nowait=false,bool more=false)
{
int options=0;
if(nowait) options|=ZMQ_DONTWAIT;
if(more) options|=ZMQ_SNDMORE;
return -1!=zmq_send(m_ref,buf,ArraySize(buf),options);
}
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
bool Socket::sendConst(const uchar &buf[],bool nowait=false,bool more=false)
{
int options=0;
if(nowait) options|=ZMQ_DONTWAIT;
if(more) options|=ZMQ_SNDMORE;
return -1!=zmq_send_const(m_ref,buf,ArraySize(buf),options);
}
//+------------------------------------------------------------------+
//| Send a zmq_msg_t through a socket |
//+------------------------------------------------------------------+
bool Socket::send(ZmqMsg &msg,bool nowait=false,bool more=false)
{
int flags=0;
if(nowait) flags|=ZMQ_DONTWAIT;
if(more) flags|=ZMQ_SNDMORE;
return -1!=zmq_msg_send(msg,m_ref,flags);
}
//+------------------------------------------------------------------+
//| Receive a zmq_msg_t from a socket |
//+------------------------------------------------------------------+
bool Socket::recv(ZmqMsg &msg,bool nowait=false)
{
int flags=0;
if(nowait) flags|=ZMQ_DONTWAIT;
return -1!=zmq_msg_recv(msg,m_ref,flags);
}
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
bool Socket::monitor(string addr,int events)
{
uchar str[];
@@ -263,26 +231,6 @@ bool Socket::monitor(string addr,int events)
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
void Socket::register(PollItem &pollitem,bool read=false,bool write=false)
{
ZeroMemory(pollitem);
pollitem.socket=m_ref;
if(read) pollitem.events|=ZMQ_POLLIN;
if(write) pollitem.events|=ZMQ_POLLOUT;
}
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
void Socket::register(PollItem &pollitems[],int index,bool read=false,bool write=false)
{
ZeroMemory(pollitems[index]);
pollitems[index].socket=m_ref;
if(read) pollitems[index].events|=ZMQ_POLLIN;
if(write) pollitems[index].events|=ZMQ_POLLOUT;
}
//+------------------------------------------------------------------+
//| |
//+------------------------------------------------------------------+
bool Socket::proxy(Socket *frontend,Socket *backend,Socket *capture)
{
intptr_t frontend_ref= CheckPointer(frontend)==POINTER_DYNAMIC?frontend.ref():0;
+6 -3
View File
@@ -184,10 +184,13 @@ public:
SOCKOPT_GET_BYTES(OptionName,Macro,InitSize)
//--- for curve key
//--- NOTE that the length of the key array must be 32.
//--- since some versions of mt4 and newer versions of mt5 will report error
//--- when the length of the array is specified, so the length specification is removed
#define SOCKOPT_CURVE_KEY(KeyType,Macro) \
bool getCurve##KeyType##Key(uchar &key[32]) {size_t len=32; return getOption(Macro,key,len);} \
bool getCurve##KeyType##Key(uchar &key[]) {size_t len=32; return getOption(Macro,key,len);} \
bool getCurve##KeyType##Key(string &key) {return getStringOption(Macro,key,41);} \
bool setCurve##KeyType##Key(const uchar &key[32]) {return setOption(Macro,key,32);} \
bool setCurve##KeyType##Key(const uchar &key[]) {return setOption(Macro,key,32);} \
bool setCurve##KeyType##Key(string key) {return setStringOption(Macro,key);}
SOCKOPT_GET(int,Type,ZMQ_TYPE)
@@ -251,7 +254,7 @@ public:
SOCKOPT(int,ReceiveTimeout,ZMQ_RCVTIMEO) // milliseconds
SOCKOPT(int,SendBuffer,ZMQ_SNDBUF) // bytes
SOCKOPT(int,SendHighWaterMark,ZMQ_SNDHWM) // messages
SOCKOPT(int,SendTimout,ZMQ_SNDTIMEO)
SOCKOPT(int,SendTimeout,ZMQ_SNDTIMEO)
SOCKOPT_GET_BOOL(ReceiveMore,ZMQ_RCVMORE)
+2 -2
View File
@@ -117,7 +117,7 @@ void ZmqMsg::getData(uchar &data[])
{
size_t size=size();
intptr_t src=data();
ArrayResize(data,(int)size);
if(ArraySize(data)<size) ArrayResize(data,(int)size);
ArrayFromPointer(data,src);
}
//+------------------------------------------------------------------+
@@ -136,7 +136,7 @@ void ZmqMsg::setData(const uchar &data[])
{
intptr_t dest=data();
size_t size=size();
ArrayToPointer(data,dest,size);
ArrayToPointer(data,dest,(int)size);
}
//+------------------------------------------------------------------+
//| Wraps zmq_msg_gets: get metadata associated with the msg |
+7 -2
View File
@@ -45,7 +45,7 @@ This binding contains three sets of files:
latest Visual C++ runtime (2015)**.
*Note* that if you are using **MT5 32bit**, you need to comment out the
`__X64__` macro definition at the top of the `Include/Zmq/Common.mqh`. I
`__X64__` macro definition at the top of the `Include/Mql/Lang/Native.mqh`. I
assume MT5 is 64 bit, since their is no way to detect 32 bit by native
macros, and to define pointer related values a macro like this is required.
@@ -128,14 +128,19 @@ void OnStart()
1. Write more tests.
2. Add more examples from the official ZMQ guide.
3. More documentation
4. High level API for common patterns
## Changes
* 2017-10-28: Released 1.5: Important: API change for `Socket.send`; Remove
PollItem duplicate API (#11); Fix compiler warning (#10) and compile failure
(#12); Add RTReq example from ZMQ Guide Chapter 3.
* 2017-08-18: Released 1.4: Fix ZmqMsg setData bug; Change License to Apache
2.0; Inlcude mql4-lib dependencies directly.
* 2017-07-18: Released 1.3: Refactored poll support; Add Chapter 2 examples from
the official ZMQ guide.
* 2017-06-08: Released 1.2: Fix GlobalHandle bug; Add rebuild method to ZmqMsg;
Complete all examples in ZMQ Guide Chanpter 1.
Complete all examples in ZMQ Guide Chapter 1.
* 2017-05-26: Released 1.1: add the ability to share a ZMQ context globally in a terminal
* 2016-12-27: Released 1.0.
@@ -42,8 +42,8 @@ void OnStart()
do
{
frontend.recv(message);
more=message.more();
backend.send(message,false,more);
if(message.more()) backend.sendMore(message);
else backend.send(message);
}
while(more);
}
@@ -53,8 +53,8 @@ void OnStart()
do
{
backend.recv(message);
more=message.more();
frontend.send(message,false,more);
if(message.more()) frontend.sendMore(message);
else frontend.send(message);
}
while(more);
}
@@ -38,8 +38,8 @@ void OnStart()
do
{
frontend.recv(message);
more=message.more();
backend.send(message,false,more);
if(message.more()) backend.sendMore(message);
else backend.send(message);
}
while(more);
}
@@ -0,0 +1,85 @@
//+------------------------------------------------------------------+
//| RTReqBroker.mq4 |
//| Copyright 2017, Li Ding |
//| dingmaotu@126.com |
//+------------------------------------------------------------------+
#property copyright "Copyright 2017, Li Ding"
#property link "dingmaotu@126.com"
#property version "1.00"
#property strict
#property show_inputs
//+------------------------------------------------------------------+
//| This example comes from the "Load Balancing Pattern" |
//| The original example creates threads for workers and spawn them |
//| in the broker main thread. MetaTrader Terminal does not support |
//| thread creation. So we split the broker and worker code to two |
//| scripts. Since we splitted the code, we need to wait all workers |
//| connect. |
//| This is the broker part. |
//+------------------------------------------------------------------+
#include <Zmq/Zmq.mqh>
input int InpNumberWorkers=5;
//+------------------------------------------------------------------+
//| Custom routing Router to Mama (ROUTER to REQ) (adapted from C++) |
//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com> |
//+------------------------------------------------------------------+
void OnStart()
{
Context context("rtreq");
Socket broker(context,ZMQ_ROUTER);
broker.bind("inproc://rtreq");
string identities[];
ArrayResize(identities,InpNumberWorkers);
// Wait until InpNumberWorkers workers connected
for(int i=0; i<InpNumberWorkers; i++)
{
ZmqMsg msg;
broker.recv(msg);
string identity=msg.getData();
broker.recv(msg); // Envelope delimiter
broker.recv(msg); // Response from worker
identities[i]=identity;
Print("Broker: Worker ",identity," connected.");
}
Print("Broker: All workers connected!");
// Notify all workers that it is ready to dispatch work
for(int i=0; i<InpNumberWorkers; i++)
{
broker.sendMore(identities[i]);
broker.sendMore();
broker.send("Go!");
}
// Run for five seconds and then tell workers to end
long endTime=TimeLocal()+5;
int workersFired=0;
while(!IsStopped())
{
ZmqMsg msg;
// Next message gives us least recently used worker
broker.recv(msg);
string identity=msg.getData();
Print("Broker: Get available worker [",identity,"]");
broker.recv(msg); // Envelope delimiter
broker.recv(msg); // Response from worker
Print("Broker: And he says ",msg.getData());
if(!broker.sendMore(identity)) {Print("Error sending identity.");}
if(!broker.sendMore("")) {Print("Error sending delimeter.");}
// Encourage workers until it's time to fire them
if(TimeLocal()<endTime)
{
Print("Send work!");
if(!broker.send("Work harder")) {Print("Error sending work.");}
}
else
{
Print("Send fire!");
if(!broker.send("Fired!")) {Print("Error sending fired.");}
if(++workersFired==InpNumberWorkers)
break;
}
}
}
//+------------------------------------------------------------------+
@@ -0,0 +1,72 @@
//+------------------------------------------------------------------+
//| RTReqWorker.mq4 |
//| Copyright 2017, Li Ding |
//| dingmaotu@126.com |
//+------------------------------------------------------------------+
#property copyright "Copyright 2017, Li Ding"
#property link "dingmaotu@126.com"
#property version "1.00"
#property strict
#include <Zmq/Zmq.mqh>
#property show_inputs
//+------------------------------------------------------------------+
//| This example comes from the "Load Balancing Pattern" |
//| The original example creates threads for workers and spawn them |
//| in the broker main thread. MetaTrader Terminal does not support |
//| thread creation. So we split the broker and worker code to two |
//| scripts. Since we splitted the code, we need to wait all workers |
//| connect. |
//| This is the worker part. |
//| For the worker, we can use either ZMQ_REQ or ZMQ_DEALER, the |
//| difference is minimal. When using a dealer socket, remember to |
//| send an empty frame to emulate the REQ behavior. |
//+------------------------------------------------------------------+
#define within(num) (int) ((float) num * MathRand() / (32767 + 1.0))
input string InpWorkerIdentity="worker1";
//+------------------------------------------------------------------+
//| Custom routing Router to Mama (ROUTER to REQ) (adapted from C++) |
//| The worker |
//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com> |
//+------------------------------------------------------------------+
void OnStart()
{
//--- use inproc
Context context("rtreq");
Socket worker(context,ZMQ_REQ);
//--- We use a string identity for ease here
worker.setIdentity(InpWorkerIdentity);
worker.connect("inproc://rtreq");
ZmqMsg msg("Connect!");
worker.send(msg);
worker.recv(msg);
Print(InpWorkerIdentity," connect: ",msg.getData());
int total=0;
while(!IsStopped())
{
// Tell the broker we're ready for work
worker.send("Hi Boss");
// Get workload from broker, until finished
worker.recv(msg);
string workload=msg.getData();
if("Fired!"==workload)
{
Print(InpWorkerIdentity," is fired!");
Print(InpWorkerIdentity," processed: ",total," tasks");
break;
}
else
{
Print(InpWorkerIdentity," received work!");
}
total++;
// Do some random work
Sleep(within(500)+1);
}
}
//+------------------------------------------------------------------+