diff --git a/Include/Zmq/Context.mqh b/Include/Zmq/Context.mqh index 51d1262..920667d 100644 --- a/Include/Zmq/Context.mqh +++ b/Include/Zmq/Context.mqh @@ -48,16 +48,18 @@ int zmq_ctx_get(intptr_t context,int option); //| and in a manner not easily recognized by humans, for example: | //| "__3kewducdxhkd__" | //+------------------------------------------------------------------+ -class Context: public GlobalHandle +class Context: public GlobalHandle { protected: int get(int option) {return zmq_ctx_get(m_ref,option);} bool set(int option,int optval) {return 0==zmq_ctx_set(m_ref,option,optval);} - intptr_t create() override {return zmq_ctx_new();} - void destroy(intptr_t handle) override {if(0!=zmq_ctx_term(handle)) {Debug("failed to terminate context");}} public: - Context(string shared=NULL):GlobalHandle(shared) {} + + static intptr_t create() {return zmq_ctx_new();} + static void destroy(intptr_t handle) {if(0!=zmq_ctx_term(handle)) {Debug("failed to terminate context");}} + + Context(string shared=NULL):GlobalHandle(shared) {} bool shutdown() {return 0==zmq_ctx_shutdown(m_ref);} diff --git a/Include/Zmq/GlobalHandle.mqh b/Include/Zmq/GlobalHandle.mqh index 73c3065..3de03b5 100644 --- a/Include/Zmq/GlobalHandle.mqh +++ b/Include/Zmq/GlobalHandle.mqh @@ -55,14 +55,15 @@ public: bool isValid() const {return m_name!=NULL;} string getName() const {return m_name;} - void enter() { while(!GlobalVariable::makeTemp(m_name))Sleep(100); } + void enter() { while(!GlobalVariable::makeTemp(m_name) && !IsStopped())Sleep(100); } bool tryEnter() { return GlobalVariable::makeTemp(m_name); } void leave() { GlobalVariable::remove(m_name);} }; //+------------------------------------------------------------------+ //| A reference counted global pointer (or handle) | +//| HandleManager should implement 2 static methods: create & destroy| //+------------------------------------------------------------------+ -template +template class GlobalHandle { private: @@ -76,7 +77,7 @@ public: { m_refName=m_cs.getName()+"_Ref"; m_counterName=m_cs.getName()+"_Count"; - if(!m_cs.isValid()) m_ref=create(); + if(!m_cs.isValid()) m_ref=HandleManager::create(); else { m_cs.enter(); @@ -87,7 +88,7 @@ public: } if(long(GlobalVariable::get(m_counterName))==0) { - m_ref=create(); + m_ref=HandleManager::create(); if(!GlobalVariable::exists(m_refName)) { GlobalVariable::makeTemp(m_refName); @@ -104,12 +105,12 @@ public: } ~GlobalHandle() { - if(!m_cs.isValid()) {destroy(m_ref); return;} + if(!m_cs.isValid()) {HandleManager::destroy(m_ref); return;} m_cs.enter(); GlobalVariable::set(m_counterName,GlobalVariable::get(m_counterName)-1); if(long(GlobalVariable::get(m_counterName))==0) { - destroy(m_ref); + HandleManager::destroy(m_ref); GlobalVariable::remove(m_refName); GlobalVariable::remove(m_counterName); } @@ -117,9 +118,5 @@ public: } T ref() const {return m_ref;} - -protected: - virtual T create()=NULL; - virtual void destroy(T handle)=NULL; }; //+------------------------------------------------------------------+ diff --git a/Include/Zmq/ZmqMsg.mqh b/Include/Zmq/ZmqMsg.mqh index cd63a97..ebd0a21 100644 --- a/Include/Zmq/ZmqMsg.mqh +++ b/Include/Zmq/ZmqMsg.mqh @@ -49,12 +49,29 @@ protected: int get(int property) {return zmq_msg_get(this,property);} bool set(int property,int value) {return 0==zmq_msg_set(this,property,value);} intptr_t data() {return zmq_msg_data(this);} + bool setStringData(string data,bool nullterminated=false); public: ZmqMsg() {zmq_msg_init(this);} ZmqMsg(int size) {if(0!=zmq_msg_init_size(this,size)){Debug("Failed to init size msg: insufficient space");}} - ZmqMsg(string data,bool nullterminated=false); + ZmqMsg(string data,bool nullterminated=false) {setStringData(data,nullterminated);} ~ZmqMsg() {if(0!=zmq_msg_close(this)){Debug("Failed to close msg");}} + bool rebuild() + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return 0==zmq_msg_init(this); + } + bool rebuild(int size) + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return 0==zmq_msg_init_size(this,size); + } + bool rebuild(string data,bool nullterminated=false) + { + if(0!=zmq_msg_close(this)){Debug("Failed to close msg");return false;} + return setStringData(data,nullterminated); + } + size_t size() {return zmq_msg_size(this);} void getData(uchar &data[]); @@ -71,13 +88,13 @@ public: //+------------------------------------------------------------------+ //| Initialize a utf-8 string message | //+------------------------------------------------------------------+ -ZmqMsg::ZmqMsg(string data,bool nullterminated) +bool ZmqMsg::setStringData(string data,bool nullterminated) { uchar array[]; StringToUtf8(data,array,nullterminated); - int size=ArraySize(array); - zmq_msg_init_size(this,size); - setData(array); + bool res=(0==zmq_msg_init_size(this,ArraySize(array))); + if(res)setData(array); + return res; } //+------------------------------------------------------------------+ //| Get message data as bytes array | diff --git a/README.md b/README.md index 4dca96c..e5ab17e 100644 --- a/README.md +++ b/README.md @@ -92,5 +92,7 @@ void OnStart() ## Changes +* 2017-06-08: Released 1.2: Fix GlobalHandle bug. Add rebuild method to ZmqMsg. + Complete all examples in ZMQ Guide Chanpter 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/Chapter1/TaskEvent.mq4 b/Scripts/ZeroMQGuideExamples/Chapter1/TaskEvent.mq4 new file mode 100644 index 0000000..c91224e Binary files /dev/null and b/Scripts/ZeroMQGuideExamples/Chapter1/TaskEvent.mq4 differ diff --git a/Scripts/ZeroMQGuideExamples/Chapter1/TaskSink.mq4 b/Scripts/ZeroMQGuideExamples/Chapter1/TaskSink.mq4 new file mode 100644 index 0000000..021fdd3 --- /dev/null +++ b/Scripts/ZeroMQGuideExamples/Chapter1/TaskSink.mq4 @@ -0,0 +1,48 @@ +//+------------------------------------------------------------------+ +//| TaskSink.mq4 | +//| Copyright 2017, Bear Two Technologies Co., Ltd. | +//| dingmaotu@126.com | +//+------------------------------------------------------------------+ +#property copyright "Copyright 2017, Bear Two Technologies Co., Ltd." +#property link "dingmaotu@126.com" +#property version "1.00" +#property strict + +#include +//+------------------------------------------------------------------+ +//| Task sink in MQL (adapted from C++ version) | +//| Binds PULL socket to tcp://localhost:5558 | +//| Collects results from workers via that socket | +//| | +//| Olivier Chamoux | +//+------------------------------------------------------------------+ +void OnStart() + { +//--- +// Prepare our context and socket + Context context; + Socket receiver(context,ZMQ_PULL); + receiver.bind("tcp://*:5558"); + +// Wait for start of batch + ZmqMsg message; + receiver.recv(message); + +// Start our clock now + uint tstart=GetTickCount(); +// Process 100 confirmations + string progress=""; + for(int i=0; i<100; i++) + { + receiver.recv(message); + if((i/10)*10==i) + progress+=":"; + else + progress+="."; + Comment(progress); + } +// Calculate and report duration of batch + uint tend=GetTickCount(); + Print(">>> Total elapsed time: ",tend-tstart," msec"); + } +//+------------------------------------------------------------------+ diff --git a/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 b/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 new file mode 100644 index 0000000..fa0a298 --- /dev/null +++ b/Scripts/ZeroMQGuideExamples/Chapter1/TaskWorker.mq4 @@ -0,0 +1,52 @@ +//+------------------------------------------------------------------+ +//| TaskWorker.mq4 | +//| Copyright 2017, Bear Two Technologies Co., Ltd. | +//| dingmaotu@126.com | +//+------------------------------------------------------------------+ +#property copyright "Copyright 2017, Bear Two Technologies Co., Ltd." +#property link "dingmaotu@126.com" +#property version "1.00" +#property strict +#include +//+------------------------------------------------------------------+ +//| Task worker in MQL (adapted from C++ version) | +//| Connects PULL socket to tcp://localhost:5557 | +//| Collects workloads from ventilator via that socket | +//| Connects PUSH socket to tcp://localhost:5558 | +//| Sends results to sink via that socket | +//| | +//| Olivier Chamoux | +//+------------------------------------------------------------------+ +void OnStart() + { +//--- Share a single context in the terminal by the key "work" + Context context("work"); + +//--- Socket to receive messages on + Socket receiver(context,ZMQ_PULL); + receiver.connect("tcp://localhost:5557"); + +//--- Socket to send messages to + Socket sender(context,ZMQ_PUSH); + sender.connect("tcp://localhost:5558"); + +//--- Process tasks forever + string progress=""; + while(!IsStopped()) + { + ZmqMsg message; + receiver.recv(message); + //--- Workload in msecs + int workload=(int)StringToInteger(message.getData()); + //--- Do the work + Sleep(workload); + //--- Send results to sink + message.rebuild(); + sender.send(message); + + // Simple progress indicator for the viewer + progress+="."; + Comment(progress); + } + } +//+------------------------------------------------------------------+