gavindiaz/mql-zmq
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix GlobalHandle bug. Add rebuild method to ZmqMsg.

Browse Source
This commit is contained in:
Ding Li
2017-06-08 18:35:32 +08:00
parent a2c610956c
commit 4645013bb2
7 changed files with 137 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Include/Zmq/Context.mqh
+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);}
Include/Zmq/GlobalHandle.mqh
+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;
};
//+------------------------------------------------------------------+
Include/Zmq/ZmqMsg.mqh
+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 |
README.md
+2
View File
@@ -92,5 +92,7 @@ void OnStart()
## Changes
* 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/TaskEvent.mq4
BIN
View File
Binary file not shown.
Scripts/ZeroMQGuideExamples/Chapter1/TaskSink.mq4
+48
View File
@@ -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
+52
View File
@@ -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);
}
}
//+------------------------------------------------------------------+