9 Commits

Author SHA1 Message Date
Ding Li 4645013bb2 Fix GlobalHandle bug. Add rebuild method to ZmqMsg. 2017-06-08 18:35:32 +08:00
Ding Li a2c610956c Fix #2 caused by a StringToCharArray bug in MQL 2017-05-27 10:22:18 +08:00
Ding Li 3a231f12b8 Fix send recv method return value check bug 2017-05-26 14:09:57 +08:00
Ding Li b4c30887c9 Fix more format 2017-05-26 13:21:56 +08:00
Ding Li 1720e3c559 Fix format 2017-05-26 13:18:09 +08:00
Ding Li 7d29a7ebe9 update documentation 2017-05-26 13:14:21 +08:00
Ding Li 29b5b9e36e Sharing context between mql programs in the same Terminal; add new examples 2017-05-26 13:02:23 +08:00
Ding Li 1450c16b36 Fix #3: delete deprecated comment of Socket contructor 2017-05-22 10:57:01 +08:00
Ding Li c8ffea4433 Add an usage sample to README 2016-12-28 13:16:59 +08:00
15 changed files with 408 additions and 25 deletions
-2
View File
@@ -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"
+3 -2
View File
@@ -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);
}
+25 -7
View File
@@ -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);}
+122
View File
@@ -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;}
};
//+------------------------------------------------------------------+
+3 -6
View File
@@ -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 |
+22 -5
View File
@@ -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 |
+58 -1
View File
@@ -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.
@@ -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;
Binary file not shown.
@@ -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");
}
//+------------------------------------------------------------------+
@@ -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);
}
}
//+------------------------------------------------------------------+
@@ -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());
}
//+------------------------------------------------------------------+
@@ -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);
}
}
}
//+------------------------------------------------------------------+