Compare commits
11 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| afab5f93cc | |||
| 795acc13c8 | |||
| 184b9f6d8b | |||
| 96fd26ad85 | |||
| 1d6b1aa2d5 | |||
| 07414d6300 | |||
| 8475ecc03a | |||
| 1d127c66b7 | |||
| 101807d69e | |||
| cd7e5f0e47 | |||
| bab1c9753a |
@@ -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
@@ -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;
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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 |
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
//+------------------------------------------------------------------+
|
||||
Reference in New Issue
Block a user