6 Commits

Author SHA1 Message Date
Ding Li 01afabc408 Refactored poll support. Add Chapter 2 examples from the official ZMQ guide. 2017-07-18 16:40:22 +08:00
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
14 changed files with 454 additions and 49 deletions
+1
View File
@@ -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);
}
+6 -4
View File
@@ -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);}
+7 -10
View File
@@ -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;
};
//+------------------------------------------------------------------+
+35 -19
View File
@@ -9,14 +9,6 @@
#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
@@ -64,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 |
@@ -87,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 |
//+------------------------------------------------------------------+
@@ -123,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);
@@ -134,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);
};
//+------------------------------------------------------------------+
//| |
@@ -187,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);
}
//+------------------------------------------------------------------+
//| |
@@ -197,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);
}
//+------------------------------------------------------------------+
//| |
@@ -207,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 |
@@ -242,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;
@@ -252,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;
@@ -281,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;
}
//+------------------------------------------------------------------+
+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 |
+42 -11
View File
@@ -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,13 +27,27 @@ 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
@@ -42,11 +64,15 @@ 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
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`:
@@ -91,5 +117,10 @@ 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-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.
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,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
}
}
}
//+------------------------------------------------------------------+
@@ -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);
}
}
}
//+------------------------------------------------------------------+
@@ -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());
}
}
//+------------------------------------------------------------------+
@@ -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);
}
}
//+------------------------------------------------------------------+
@@ -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);
}
}
//+------------------------------------------------------------------+