Refactored poll support. Add Chapter 2 examples from the official ZMQ guide.

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,22 +3,12 @@
 //|                                          Copyright 2016, Li Ding |
 //|                                            dingmaotu@hotmail.com |
 //+------------------------------------------------------------------+
-#property copyright "Copyright 2016, Li Ding"
-#property link      "dingmaotu@hotmail.com"
 #property strict
 
 #include "Common.mqh"
 #include "Context.mqh"
 #include "SocketOptions.mqh"
 #include "ZmqMsg.mqh"
-//--- fd is SOCKET on Win32, which is defined as UINT_PTR
-struct zmq_pollitem_t
-  {
-   intptr_t          socket;
-   uintptr_t         fd;
-   short             events;
-   short             revents;
-  };
 
 //--- Socket types
 #define ZMQ_PAIR 0
@@ -66,7 +56,17 @@ struct zmq_pollitem_t
 #define ZMQ_POLLPRI 8
 
 #define ZMQ_POLLITEMS_DFLT 16
+//--- fd is SOCKET on Win32, which is defined as UINT_PTR
+struct PollItem
+  {
+   intptr_t          socket;
+   uintptr_t         fd;
+   short             events;
+   short             revents;
 
+   bool              hasInput() const {return(revents&ZMQ_POLLIN)!=0;}
+   bool              hasOutput() const {return(revents&ZMQ_POLLOUT)!=0;}
+  };
 #import "libzmq.dll"
 //+------------------------------------------------------------------+
 //| Sockets                                                          |
@@ -89,7 +89,7 @@ int zmq_msg_recv(zmq_msg_t &msg,intptr_t s,int flags);
 //+------------------------------------------------------------------+
 //| I/O multiplexing                                                 |
 //+------------------------------------------------------------------+
-int zmq_poll(zmq_pollitem_t &items[],int nitems,long timeout);
+int zmq_poll(PollItem &items[],int nitems,long timeout);
 //+------------------------------------------------------------------+
 //| Message proxying                                                 |
 //+------------------------------------------------------------------+
@@ -102,7 +102,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));}}
 
@@ -126,8 +125,8 @@ public:
    bool              send(ZmqMsg &msg,bool nowait=false,bool more=false);
    bool              recv(ZmqMsg &msg,bool nowait=false);
 
-   void              register(zmq_pollitem_t &pollitem,bool read=false,bool write=false);
-   void              register(zmq_pollitem_t &pollitems[],int index,bool read=false,bool write=false);
+   void              register(PollItem &pollitem,bool read=false,bool write=false);
+   void              register(PollItem &pollitems[],int index,bool read=false,bool write=false);
 
    //--- monitor socket events
    bool              monitor(string addr,int events);
@@ -137,7 +136,10 @@ public:
    static bool       proxySteerable(Socket *frontend,Socket *backend,Socket *capture,Socket *control);
 
    //--- poll
-   static int        poll(zmq_pollitem_t &arr[],long timeout);
+   static int        poll(PollItem &arr[],long timeout);
+
+   //--- fill a poll item for this socket
+   void              fillPollItem(PollItem &item,short events);
   };
 //+------------------------------------------------------------------+
 //|                                                                  |
@@ -190,7 +192,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 +202,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 +212,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                                |
@@ -245,7 +247,7 @@ bool Socket::monitor(string addr,int events)
 //+------------------------------------------------------------------+
 //|                                                                  |
 //+------------------------------------------------------------------+
-void Socket::register(zmq_pollitem_t &pollitem,bool read=false,bool write=false)
+void Socket::register(PollItem &pollitem,bool read=false,bool write=false)
   {
    ZeroMemory(pollitem);
    pollitem.socket=m_ref;
@@ -255,7 +257,7 @@ void Socket::register(zmq_pollitem_t &pollitem,bool read=false,bool write=false)
 //+------------------------------------------------------------------+
 //|                                                                  |
 //+------------------------------------------------------------------+
-void Socket::register(zmq_pollitem_t &pollitems[],int index,bool read=false,bool write=false)
+void Socket::register(PollItem &pollitems[],int index,bool read=false,bool write=false)
   {
    ZeroMemory(pollitems[index]);
    pollitems[index].socket=m_ref;
@@ -284,10 +286,21 @@ bool Socket::proxySteerable(Socket *frontend,Socket *backend,Socket *capture,Soc
    return 0==zmq_proxy_steerable(frontend_ref, backend_ref, capture_ref, control_ref);
   }
 //+------------------------------------------------------------------+
-//|                                                                  |
+//| poll for events. timeout is milliseconds (-1 for indefinite wait |
+//| and 0 for immediate return)                                      |
 //+------------------------------------------------------------------+
-int Socket::poll(zmq_pollitem_t &arr[],long timeout)
+int Socket::poll(PollItem &arr[],long timeout)
   {
    return zmq_poll(arr,ArraySize(arr),timeout);
   }
 //+------------------------------------------------------------------+
+//| fill a poll item for this socket                                 |
+//+------------------------------------------------------------------+
+void Socket::fillPollItem(PollItem &item,short events)
+  {
+   item.socket=m_ref;
+   item.fd=0;
+   item.events=events;
+   item.revents=0;
+  }
+//+------------------------------------------------------------------+

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

@@ -7,11 +7,19 @@ ZMQ binding for the MQL language (both 32bit MT4 and 64bit MT5)
 This is a complete binding of the [ZeroMQ](http://zeromq.org/) library
 for the MQL4/5 language provided by MetaTrader4/5.
 
-Traders with programming abilities have always wanted a messaging solution
-like ZeroMQ, simple and powerful, far better than the PIPE trick as 
-suggested by the official articles. However, bindings for MQL were either outdated or not complete (mostly toy projects and only basic features are implemented). This binding is based on latest 4.2 version of the library, and provides all functionalities as specified in the API documentation.
+Traders with programming abilities have always wanted a messaging solution like
+ZeroMQ, simple and powerful, far better than the PIPE trick as suggested by the
+official articles. However, bindings for MQL were either outdated or not
+complete (mostly toy projects and only basic features are implemented). This
+binding is based on latest 4.2 version of the library, and provides all
+functionalities as specified in the API documentation.
 
-This binding tries to remain compatible between MQL4/5. Users of both versions can use this binding, with a single set of headers. MQL4 and MQL5 are basically the same in that they are merged in recent versions. The difference is in the runtime environment (MetaTrader5 is 64bit by default, while MetaTrader4 is 32bit). The trading system is also different, but it is no concern of this binding.
+This binding tries to remain compatible between MQL4/5. Users of both versions
+can use this binding, with a single set of headers. MQL4 and MQL5 are basically
+the same in that they are merged in recent versions. The difference is in the
+runtime environment (MetaTrader5 is 64bit by default, while MetaTrader4 is
+32bit). The trading system is also different, but it is no concern of this
+binding.
 
 ## Files 
 
@@ -19,17 +27,89 @@ This binding contains three sets of files:
 
 1. The binding itself is in the `Include/Zmq` directory.
 
-2. The testing scripts and zmq guide examples are in `Scripts` directory. The script files are mq4 by default, but you can change the extension to mq5 to use them in MetaTrader5.
+2. The testing scripts and zmq guide examples are in `Scripts` directory. The
+   script files are mq4 by default, but you can change the extension to mq5 to
+   use them in MetaTrader5.
 
-3. Precompiled DLLs of both 64bit (`Library/MT5`) and 32bit (`Library/MT4`) ZeroMQ and libsodium are provided. Copy the corresponding DLLs to the `Library` folder of your MetaTrader terminal. If you are using MT5 32bit, use the 32bit version from `Library/MT4`. The DLLs require that you have the latest Visual C++ runtime (2015). *Note* that these DLLs are compiled from official sources, without any modification. You can compile your own if you don't trust these binaries. The `libsodium.dll` is copied from the official binary release. If you want to support security mechanisms other than `curve`, or you want to use transports like OpenPGM, you need to compile your own DLL.
+3. Precompiled DLLs of both 64bit (`Library/MT5`) and 32bit (`Library/MT4`)
+   ZeroMQ and libsodium are provided. Copy the corresponding DLLs to the
+   `Library` folder of your MetaTrader terminal. If you are using MT5 32bit, use
+   the 32bit version from `Library/MT4`. The DLLs require that you have the
+   latest Visual C++ runtime (2015). *Note* that these DLLs are compiled from
+   official sources, without any modification. You can compile your own if you
+   don't trust these binaries. The `libsodium.dll` is copied from the official
+   binary release. If you want to support security mechanisms other than
+   `curve`, or you want to use transports like OpenPGM, you need to compile your
+   own DLL.
 
 ## About string encoding
 
-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.
+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.
+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
 
@@ -38,4 +118,9 @@ You can find a simple test script in `Scripts/Test`, and you can find examples o
 
 ## Changes
 
-2016-12-27: Released 1.0.
+* 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.
+* 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);
+        }
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter2/MSPoller.mq4

@@ -0,0 +1,54 @@
+//+------------------------------------------------------------------+
+//|                                                     MSPoller.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>
+//+------------------------------------------------------------------+
+//| Reading from multiple sockets in MQL (adapted from C++ version)  |
+//| This version uses zmq_poll()                                     |
+//|                                                                  |
+//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+//---
+   Context context;
+
+//  Connect to task ventilator
+   Socket receiver(context,ZMQ_PULL);
+   receiver.connect("tcp://localhost:5557");
+
+//  Connect to weather server
+   Socket subscriber(context,ZMQ_SUB);
+   subscriber.connect("tcp://localhost:5556");
+   subscriber.subscribe("10001 ");
+
+//  Initialize poll set
+   PollItem items[2];
+   receiver.fillPollItem(items[0],ZMQ_POLLIN);
+   subscriber.fillPollItem(items[1],ZMQ_POLLIN);
+//  Process messages from both sockets
+   while(!IsStopped())
+     {
+      ZmqMsg message;
+      //--- MQL Note: To handle Script exit properly, we set a timeout of 500 ms instead of infinite wait
+      Socket::poll(items,500);
+
+      if(items[0].hasInput())
+        {
+         receiver.recv(message);
+         //  Process task
+        }
+      if(items[1].hasInput())
+        {
+         subscriber.recv(message);
+         //  Process weather update
+        }
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter2/RRBroker.mq4

@@ -0,0 +1,63 @@
+//+------------------------------------------------------------------+
+//|                                                     RRBroker.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>
+//+------------------------------------------------------------------+
+//| Simple request-reply broker in MQL (adapted from C++ version)    |
+//|                                                                  |
+//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+//  Prepare our context and sockets
+   Context context;
+   Socket frontend(context,ZMQ_ROUTER);
+   Socket backend(context,ZMQ_DEALER);
+
+   frontend.bind("tcp://*:5559");
+   backend.bind("tcp://*:5560");
+
+//  Initialize poll set
+   PollItem items[2];
+   frontend.fillPollItem(items[0],ZMQ_POLLIN);
+   backend.fillPollItem(items[1],ZMQ_POLLIN);
+
+//  Switch messages between sockets
+   while(!IsStopped())
+     {
+      ZmqMsg message;
+      bool more=false;               //  Multipart detection
+
+      Socket::poll(items,500);
+
+      if(items[0].hasInput())
+        {
+         //  Process all parts of the message
+         do
+           {
+            frontend.recv(message);
+            more=message.more();
+            backend.send(message,false,more);
+           }
+         while(more);
+        }
+      if(items[1].hasInput())
+        {
+         //  Process all parts of the message
+         do
+           {
+            backend.recv(message);
+            more=message.more();
+            frontend.send(message,false,more);
+           }
+         while(more);
+        }
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter2/RRClient.mq4

@@ -0,0 +1,36 @@
+//+------------------------------------------------------------------+
+//|                                                     RRClient.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>
+//+------------------------------------------------------------------+
+//| Request-reply client in MQL (adapted from C++ version)           |
+//| Connects REQ socket to tcp://localhost:5559                      |
+//| Sends "Hello" to server, expects "World" back                    |
+//|                                                                  |
+//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+   Context context;
+
+   Socket requester(context,ZMQ_REQ);
+   requester.connect("tcp://localhost:5559");
+
+   for(int request=0; request<10; request++)
+     {
+      ZmqMsg message("Hello");
+      requester.send(message);
+
+      ZmqMsg reply;
+      requester.recv(reply,true);
+
+      Print("Received reply ",reply.getData());
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter2/RRWorker.mq4

@@ -0,0 +1,41 @@
+//+------------------------------------------------------------------+
+//|                                                     RRWorker.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>
+//+------------------------------------------------------------------+
+//| Request-reply service in MQL (adapted from C++ version)          |
+//| Connects REP socket to tcp://localhost:5560                      |
+//| Expects "Hello" from client, replies with "World"                |
+//|                                                                  |
+//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+   Context context;
+
+   Socket responder(context,ZMQ_REP);
+   responder.connect("tcp://localhost:5560");
+
+   while(!IsStopped())
+     {
+      //  Wait for next request from client
+      ZmqMsg req;
+      responder.recv(req);
+      Print("Received request: ",req.getData());
+
+      // Do some 'work'
+      Sleep(1000);
+
+      ZmqMsg reply("World");
+
+      //  Send reply back to client
+      responder.send(reply);
+     }
+  }
+//+------------------------------------------------------------------+

Scripts/ZeroMQGuideExamples/Chapter2/WUProxy.mq4

@@ -0,0 +1,47 @@
+//+------------------------------------------------------------------+
+//|                                                      WUProxy.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>
+//+------------------------------------------------------------------+
+//| Weather proxy device MQL (adapted from C++)                      |
+//|                                                                  |
+//| Olivier Chamoux <olivier.chamoux@fr.thalesgroup.com>             |
+//+------------------------------------------------------------------+
+void OnStart()
+  {
+//---
+   Context context;
+
+//  This is where the weather server sits
+   Socket frontend(context,ZMQ_XSUB);
+   frontend.connect("tcp://192.168.55.210:5556");
+
+//  This is our public endpoint for subscribers
+   Socket backend(context,ZMQ_XPUB);
+   backend.bind("tcp://10.1.1.0:8100");
+
+//  Subscribe on everything
+   frontend.subscribe("");
+
+//  Shunt messages out to our own subscribers
+   while(!IsStopped())
+     {
+      //  Process all parts of the message
+      ZmqMsg message;
+      bool more;
+      do
+        {
+         frontend.recv(message);
+         more=message.more();
+         backend.send(message,false,more);
+        }
+      while(more);
+     }
+  }
+//+------------------------------------------------------------------+