Fix GlobalHandle bug. Add rebuild method to ZmqMsg.

Include/Zmq/Common.mqh

@@ -104,6 +104,7 @@ string StringFromUtf8(const uchar &utf8[])
 //+------------------------------------------------------------------+
 void StringToUtf8(const string str,uchar &utf8[],bool ending=true)
   {
+   if(!ending && str=="") return;
    int count=ending ? -1 : StringLen(str);
    StringToCharArray(str,utf8,0,count,CP_UTF8);
   }

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<intptr_t>
+class Context: public GlobalHandle<intptr_t,Context>
   {
 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<intptr_t>(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<intptr_t,Context>(shared) {}
 
    bool              shutdown() {return 0==zmq_ctx_shutdown(m_ref);}
 

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<typename T>
+template<typename T,typename HandleManager>
 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;
   };
 //+------------------------------------------------------------------+

Include/Zmq/Socket.mqh

@@ -187,7 +187,7 @@ bool Socket::recv(uchar &buf[],bool nowait=false)
   {
    int options=0;
    if(nowait) options|=ZMQ_DONTWAIT;
-   return 0==zmq_recv(m_ref,buf,ArraySize(buf),options);
+   return -1!=zmq_recv(m_ref,buf,ArraySize(buf),options);
   }
 //+------------------------------------------------------------------+
 //|                                                                  |
@@ -197,7 +197,7 @@ 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 0==zmq_send(m_ref,buf,ArraySize(buf),options);
+   return -1!=zmq_send(m_ref,buf,ArraySize(buf),options);
   }
 //+------------------------------------------------------------------+
 //|                                                                  |
@@ -207,7 +207,7 @@ 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 0==zmq_send_const(m_ref,buf,ArraySize(buf),options);
+   return -1!=zmq_send_const(m_ref,buf,ArraySize(buf),options);
   }
 //+------------------------------------------------------------------+
 //| Send a zmq_msg_t through a socket                                |

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                                  |

README.md

@@ -42,7 +42,7 @@ share a process, that is the Terminal. So it is advised to use a single global
 context on all your MQL programs. The `shared` parameter of `Context` is used
 for sychronization of context creation and destruction. It is better named
 globally, and in a manner not easily recognized by humans, for example:
-"__3kewducdxhkd__"
+`__3kewducdxhkd__`
 
 ## Usage
 
@@ -91,5 +91,8 @@ void OnStart()
 2. Add more examples from the official ZMQ guide.
 
 ## Changes
-2017-05-26: Released 1.1: add the ability to share a ZMQ context globally in a terminal
-2016-12-27: Released 1.0.
+
+* 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.

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 <Zmq/Zmq.mqh>
+//+------------------------------------------------------------------+
+//|  Task sink in MQL (adapted from C++ version)                     |
+//|  Binds PULL socket to tcp://localhost:5558                       |
+//|  Collects results from workers via that socket                   |
+//|                                                                  |
+//|  Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>            |
+//+------------------------------------------------------------------+
+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");
+  }
+//+------------------------------------------------------------------+

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 <Zmq/Zmq.mqh>
+//+------------------------------------------------------------------+
+//| 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 <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+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);
+     }
+  }
+//+------------------------------------------------------------------+