Fix GlobalHandle bug. Add rebuild method to ZmqMsg.

Include/Zmq/AtomicCounter.mqh

@@ -3,8 +3,6 @@
 //|                                          Copyright 2016, Li Ding |
 //|                                            dingmaotu@hotmail.com |
 //+------------------------------------------------------------------+
-#property copyright "Copyright 2016, Li Ding"
-#property link      "dingmaotu@hotmail.com"
 #property strict
 
 #include "Common.mqh"

Include/Zmq/Common.mqh

@@ -1,10 +1,10 @@
 //+------------------------------------------------------------------+
 //|                                                       Common.mqh |
+//| This file is part of mql4-lib project (Lang/Native.mqh):         |
+//| (github.com/dingmaotu/mql4-lib)                                  |
 //|                                          Copyright 2016, Li Ding |
 //|                                            dingmaotu@hotmail.com |
 //+------------------------------------------------------------------+
-#property copyright "Copyright 2016, Li Ding"
-#property link      "dingmaotu@hotmail.com"
 #property strict
 
 #include "Errno.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

@@ -6,6 +6,7 @@
 
 #include "Common.mqh"
 #include "SocketOptions.mqh"
+#include "GlobalHandle.mqh"
 
 //--- Context options
 #define ZMQ_IO_THREADS 1
@@ -30,19 +31,36 @@ int zmq_ctx_get(intptr_t context,int option);
 #import
 //+------------------------------------------------------------------+
 //| Wraps a 0MZ context                                              |
+//|                                                                  |
+//| Note on context creation:                                        |
+//| In the official guide:                                           |
+//|   You should create and use exactly one context in your process. |
+//|   Technically, the context is the container for all sockets in a |
+//|   single process, and acts as the transport for inproc sockets,  |
+//|   which are the fastest way to connect threads in one process.   |
+//|   If at runtime a process has two contexts, these are like       |
+//|   separate ZeroMQ instances.                                     |
+//| In metatrader Terminal, every Script and Expert Advsior has its  |
+//| own thread, but they all 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 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__"                                               |
 //+------------------------------------------------------------------+
-class Context
+class Context: public GlobalHandle<intptr_t,Context>
   {
-private:
-   intptr_t          m_ref;
 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);}
+
 public:
-                     Context() {m_ref=zmq_ctx_new();}
-                    ~Context() {if(0!=zmq_ctx_term(m_ref)){Debug("failed to terminate context");}}
-   // for better cooperation between objects
-   intptr_t          ref() const {return m_ref;}
+
+   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);}
 
    int               getIoThreads() {return get(ZMQ_IO_THREADS);}

Include/Zmq/GlobalHandle.mqh

@@ -0,0 +1,122 @@
+//+------------------------------------------------------------------+
+//|                                                 GlobalHandle.mqh |
+//| This file is part of mql4-lib project (Lang/GlobalVariable.mqh): |
+//| (github.com/dingmaotu/mql4-lib)                                  |
+//|                                          Copyright 2017, Li Ding |
+//|                                            dingmaotu@hotmail.com |
+//+------------------------------------------------------------------+
+#property strict
+//+------------------------------------------------------------------+
+//| Wraps global variable functions                                  |
+//+------------------------------------------------------------------+
+class GlobalVariable
+  {
+public:
+   static int        total() {return GlobalVariablesTotal();}
+   static string     name(int index) {return GlobalVariableName(index);}
+   static void       flush() {GlobalVariablesFlush();}
+
+   static bool       exists(string name) {return GlobalVariableCheck(name);}
+   static datetime   lastAccess(string name) {return GlobalVariableTime(name);}
+
+   static bool       makeTemp(string name) {return GlobalVariableTemp(name);}
+   static double     get(string name) {return GlobalVariableGet(name);}
+   static bool       get(string name,double &value) {return GlobalVariableGet(name,value);}
+   static datetime   set(string name,double value) {return GlobalVariableSet(name,value);}
+   static bool       setOn(string name,double value,double check) {return GlobalVariableSetOnCondition(name,value,check);}
+
+   static bool       remove(string name) {return GlobalVariableDel(name);}
+   static bool       removeAll(string prefix=NULL,datetime before=0) {return GlobalVariablesDeleteAll(prefix,before);}
+  };
+//+------------------------------------------------------------------+
+//| CriticalSection object for making atomic operations              |
+//|                                                                  |
+//| An exmaple of creating a global context (the creation and destroy|
+//| are both enclosed between the SAME critical section):            |
+//|                                                                  |
+//| enter()                                                          |
+//|   if(refcount==0) create context                                 |
+//|   else refcontext                                                |
+//|   increase refcount                                              |
+//| leave()                                                          |
+//|                                                                  |
+//| enter()                                                          |
+//|   decrease refcount                                              |
+//|   if(refcount==0) context destroy                                |
+//| leave()                                                          |
+//+------------------------------------------------------------------+
+class CriticalSection
+  {
+private:
+   const string      m_name;
+public:
+                     CriticalSection(string name):m_name(name){}
+
+   bool              isValid() const {return m_name!=NULL;}
+   string            getName() const {return m_name;}
+
+   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,typename HandleManager>
+class GlobalHandle
+  {
+private:
+   CriticalSection   m_cs;
+   string            m_refName;
+   string            m_counterName;
+protected:
+   T                 m_ref;
+public:
+                     GlobalHandle(string sharedKey=NULL):m_cs(sharedKey)
+     {
+      m_refName=m_cs.getName()+"_Ref";
+      m_counterName=m_cs.getName()+"_Count";
+      if(!m_cs.isValid()) m_ref=HandleManager::create();
+      else
+        {
+         m_cs.enter();
+         if(!GlobalVariable::exists(m_counterName))
+           {
+            GlobalVariable::makeTemp(m_counterName);
+            GlobalVariable::set(m_counterName,0);
+           }
+         if(long(GlobalVariable::get(m_counterName))==0)
+           {
+            m_ref=HandleManager::create();
+            if(!GlobalVariable::exists(m_refName))
+              {
+               GlobalVariable::makeTemp(m_refName);
+               GlobalVariable::set(m_refName,m_ref);
+              }
+           }
+         else
+           {
+            m_ref=(T)(GlobalVariable::get(m_refName));
+           }
+         GlobalVariable::set(m_counterName,GlobalVariable::get(m_counterName)+1);
+         m_cs.leave();
+        }
+     }
+                    ~GlobalHandle()
+     {
+      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)
+        {
+         HandleManager::destroy(m_ref);
+         GlobalVariable::remove(m_refName);
+         GlobalVariable::remove(m_counterName);
+        }
+      m_cs.leave();
+     }
+
+   T                 ref() const {return m_ref;}
+  };
+//+------------------------------------------------------------------+

Include/Zmq/Socket.mqh

@@ -3,8 +3,6 @@
 //|                                          Copyright 2016, Li Ding |
 //|                                            dingmaotu@hotmail.com |
 //+------------------------------------------------------------------+
-#property copyright "Copyright 2016, Li Ding"
-#property link      "dingmaotu@hotmail.com"
 #property strict
 
 #include "Common.mqh"
@@ -102,7 +100,6 @@ int zmq_proxy_steerable(intptr_t frontend_ref,intptr_t backend_ref,intptr_t capt
 class Socket: public SocketOptions
   {
 public:
-   //--- it is not recommended to use this constructor directly: use Context factory methods instead
                      Socket(const Context &ctx,int type):SocketOptions(zmq_socket(ctx.ref(),type)){}
    virtual          ~Socket() {if(0!=zmq_close(m_ref)){Debug(StringFormat("Failed to close socket 0x%0X",m_ref));}}
 
@@ -190,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);
   }
 //+------------------------------------------------------------------+
 //|                                                                  |
@@ -200,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);
   }
 //+------------------------------------------------------------------+
 //|                                                                  |
@@ -210,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

@@ -27,10 +27,64 @@ This binding contains three sets of files:
 
 MQL strings are Win32 UNICODE strings (basically 2-byte UTF-16). In this binding all strings are converted to utf-8 strings before sending to the dll layer. The ZmqMsg supports a constructor from MQL strings, the default is _NOT_ null-terminated.
 
+## Notes on context creation
+
+In the official guide:
+
+> You should create and use exactly one context in your process. Technically,
+> the context is the container for all sockets in a single process, and acts as
+> the transport for inproc sockets, which are the fastest way to connect threads
+> in one process. If at runtime a process has two contexts, these are like
+> separate ZeroMQ instances.
+
+In MetaTrader, every Script and Expert Advsior has its own thread, but they all
+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__`
+
 ## Usage
 
 You can find a simple test script in `Scripts/Test`, and you can find examples of the official guide in Scripts/ZeroMQGuideExamples. I intend to translate all examples to this binding, but now only the hello world example is provided. I will gradually add those examples. Of course forking this binding if you are interested and welcome to send pull requests.
 
+Here is a sample from `HelloWorldServer.mq4`:
+
+```c++
+#include <Zmq/Zmq.mqh>
+//+------------------------------------------------------------------+
+//| Hello World server in MQL                                        |
+//| Binds REP socket to tcp://*:5555                                 |
+//| Expects "Hello" from client, replies with "World"                |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+   Context context("helloworld");
+   Socket socket(context,ZMQ_REP);
+
+   socket.bind("tcp://*:5555");
+
+   while(true)
+     {
+      ZmqMsg request;
+
+      // Wait for next request from client
+
+      // MetaTrader note: this will block the script thread
+      // and if you try to terminate this script, MetaTrader
+      // will hang (and crash if you force closing it)
+      socket.recv(request);
+      Print("Receive Hello");
+
+      Sleep(1000);
+
+      ZmqMsg reply("World");
+      // Send reply back to client
+      socket.send(reply);
+     }
+  }
+```
+
 ## TODO
 
 1. Write more tests.
@@ -38,4 +92,7 @@ You can find a simple test script in `Scripts/Test`, and you can find examples o
 
 ## Changes
 
-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/HelloWorldServer.mq4

@@ -15,12 +15,12 @@
 //+------------------------------------------------------------------+
 void OnStart()
   {
-   Context context;
+   Context context("helloworld");
    Socket socket(context,ZMQ_REP);
 
    socket.bind("tcp://*:5555");
 
-   while(true)
+   while(!IsStopped())
      {
       ZmqMsg request;
 

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);
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter1/VersionReporting.mq4

@@ -0,0 +1,19 @@
+//+------------------------------------------------------------------+
+//|                                         VersionReporting.mq4.mq4 |
+//|                                          Copyright 2016, Li Ding |
+//|                                            dingmaotu@hotmail.com |
+//+------------------------------------------------------------------+
+#property copyright "Copyright 2016, Li Ding"
+#property link      "dingmaotu@hotmail.com"
+#property version   "1.00"
+#property strict
+
+#include <Zmq/Zmq.mqh>
+//+------------------------------------------------------------------+
+//| Report 0MQ version                                               |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+   Print(Zmq::getVersion());
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter1/WeatherUpdateServer.mq4

@@ -0,0 +1,54 @@
+//+------------------------------------------------------------------+
+//|                                      WeatherUpdateServer.mq4.mq4 |
+//|                                          Copyright 2016, Li Ding |
+//|                                            dingmaotu@hotmail.com |
+//+------------------------------------------------------------------+
+#property copyright "Copyright 2016, Li Ding"
+#property link      "dingmaotu@hotmail.com"
+#property version   "1.00"
+#property strict
+
+#include <Zmq/Zmq.mqh>
+
+#define within(num) (int) ((float) num * MathRand() / (32767 + 1.0))
+//+------------------------------------------------------------------+
+//| Weather update server in MQL                                     |
+//| Binds PUB socket to tcp://*:5556                                 |
+//| Publishes random weather updates                                 |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+//--- Prepare our context and publisher
+   Context context;
+   Socket publisher(context,ZMQ_PUB);
+   publisher.bind("tcp://*:5556");
+
+   long messages_sent=0;
+//--- Initialize random number generator
+   MathSrand(GetTickCount());
+   while(!IsStopped())
+     {
+      int zipcode,temperature,relhumidity;
+
+      // Get values that will fool the boss
+
+      // MetaTrader Note:
+      // if RAND_MAX < 100000, which is the case for MetaTrader, 
+      // you may never get the required value
+      // So 30000 might be a good alternative
+      zipcode=within(30000);
+      temperature=within(215) - 80;
+      relhumidity=within(50) + 10;
+
+      // Send message to all subscribers
+      ZmqMsg message(StringFormat("%05d %d %d",zipcode,temperature,relhumidity));
+      publisher.send(message);
+      messages_sent++;
+
+      if(messages_sent%1000000==0)
+        {
+         PrintFormat("Sent %dM messages now.",messages_sent/1000000);
+        }
+     }
+  }
+//+------------------------------------------------------------------+