gavindiaz/mql-zmq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Released 1.5

Browse Source 
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
This commit is contained in:
Ding Li
2017-10-28 22:45:23 +08:00
parent 795acc13c8
commit afab5f93cc
6 changed files with 194 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Include/Zmq/Socket.mqh
+25 -54
View File
@@ -133,13 +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);}
//--- 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);
@@ -201,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[];
README.md
+6 -1
View File
@@ -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.
Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4
+4 -4
View File
@@ -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);
}
Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4
+2 -2
View File
@@ -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);
}
Scripts/ZeroMQGuideExamples/Chapter3/RTReqBroker.mq4
+85
View File
@@ -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;
}
}
}
//+------------------------------------------------------------------+
Scripts/ZeroMQGuideExamples/Chapter3/RTReqWorker.mq4
+72
View File
@@ -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);
}
}
//+------------------------------------------------------------------+