From afab5f93cc8c55dad6ad7348be7f442b78898c16 Mon Sep 17 00:00:00 2001 From: Ding Li Date: Sat, 28 Oct 2017 22:45:23 +0800 Subject: [PATCH] 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 --- Include/Zmq/Socket.mqh | 79 ++++++----------- README.md | 7 +- .../ZeroMQGuideExamples/Chapter2/RRBroker.mq4 | 8 +- .../ZeroMQGuideExamples/Chapter2/WUProxy.mq4 | 4 +- .../Chapter3/RTReqBroker.mq4 | 85 +++++++++++++++++++ .../Chapter3/RTReqWorker.mq4 | 72 ++++++++++++++++ 6 files changed, 194 insertions(+), 61 deletions(-) create mode 100644 Scripts/ZeroMQGuideExamples/Chapter3/RTReqBroker.mq4 create mode 100644 Scripts/ZeroMQGuideExamples/Chapter3/RTReqWorker.mq4 diff --git a/Include/Zmq/Socket.mqh b/Include/Zmq/Socket.mqh index cb8df99..f56c127 100644 --- a/Include/Zmq/Socket.mqh +++ b/Include/Zmq/Socket.mqh @@ -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[]; diff --git a/README.md b/README.md index bd19160..81b6c9a 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4 b/Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4 index 406d080..3c8b901 100644 --- a/Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4 +++ b/Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4 @@ -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); } diff --git a/Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4 b/Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4 index 62e54e4..7493823 100644 --- a/Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4 +++ b/Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4 @@ -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); } diff --git a/Scripts/ZeroMQGuideExamples/Chapter3/RTReqBroker.mq4 b/Scripts/ZeroMQGuideExamples/Chapter3/RTReqBroker.mq4 new file mode 100644 index 0000000..f6b882e --- /dev/null +++ b/Scripts/ZeroMQGuideExamples/Chapter3/RTReqBroker.mq4 @@ -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 +input int InpNumberWorkers=5; +//+------------------------------------------------------------------+ +//| Custom routing Router to Mama (ROUTER to REQ) (adapted from C++) | +//| Olivier Chamoux | +//+------------------------------------------------------------------+ +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 +#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 | +//+------------------------------------------------------------------+ +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); + } + } +//+------------------------------------------------------------------+