hce-node application  1.4.3
HCE Hierarchical Cluster Engine node application
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros
DataClientData.cpp
Go to the documentation of this file.
1 #include "DataClientData.hpp"
2 #include "DRCEResourceExtractor.hpp"
3 #include "ServiceMessages.hpp"
4 
5 namespace HCE {
6  namespace handlers {
7 
8  DataClientData::DataClientData(std::string name, zmq::context_t& ctx, const std::string& clientIdentity, std::string connectionString, unsigned char dataClientMode, int64_t startedAt,
9  unsigned char logPriority, const std::string& iniFile) :
10  Handler(name, ctx, clientIdentity, connectionString, startedAt, logPriority, iniFile), DataNodeOptions(Handler::handlerProperties),
11  _shiftCounter(){
12 
13  setIniFile(iniFile);
14  setDataClientSockMode(dataClientMode);
16  initialize();
17  }
18 
19  DataClientData::~DataClientData(){
20  deinitialize();
21  }
22 
23  int DataClientData::initialize(void){
24  logger.log(NODE_LOG_MODE_INFORMATION) << "Initialize client [" << getClientIdentity() << "] with connection [" << getConnectionString() << "]" << flush;
25 
27  _inprocDataSock = NULL;
28  _dataClientSock = NULL;
29  _dataClientSockConnected = false;
30 
32  initNode(getIniFile(), logger);
33 
35  setHeartbeatAt(s_clock() + getHeartbeatDelay());
37  setHeartbeatedAt(s_clock());
38 
40  setPstatsAt(s_clock() + STAT_INTERVAL);
41 
43  checkInit();
44 
46  rebuildAdminClientSocket();
47 
49  rebuildClientConnection(_inprocDataSock, NODE_CS_INPROC_DATA, ZMQ_DEALER, getClientIdentity());
50 
52  rebuildDataClientSocket();
53 
54  return true;
55  }
56 
57  void DataClientData::deinitialize(void){
58  if(_inprocDataSock){
59  delete _inprocDataSock;
60  }
61  _inprocDataSock = NULL;
62 
63  if(_dataClientSock){
64  sendByeMessage(_dataClientSock, getClientIdentity());
65  _dataClientSockConnected = false;
66  delete _dataClientSock;
67  }
68  _dataClientSock = NULL;
69 
70  HCE::drce::ResourceUsageExtractor::shutdown();
71  }
72 
73  void DataClientData::setDataClientSockMode(unsigned char dataClientSockMode){
74  handlerProperties.set(HCE::handlers::properties::DATA_CLIENT_SOCK_MODE, dataClientSockMode);
75  }
76 
77  unsigned char DataClientData::getDataClientSockMode(void){
78  return handlerProperties.get<unsigned char>(HCE::handlers::properties::DATA_CLIENT_SOCK_MODE);
79  }
80 
81  void DataClientData::setStatRequests(unsigned int statRequests){
82  statProperties.set(HCE::handlers::properties::STAT_REQUESTS, statRequests);
83  }
84 
85  unsigned int DataClientData::getStatRequests(void){
86  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_REQUESTS);
87  }
88 
89  void DataClientData::setStatResponses(unsigned int statResponses){
90  statProperties.set(HCE::handlers::properties::STAT_RESPONSES, statResponses);
91  }
92 
93  unsigned int DataClientData::getStatResponses(void){
94  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_RESPONSES);
95  }
96 
97  void DataClientData::setStatRequestsSize(unsigned int statRequestsSize){
98  statProperties.set(HCE::handlers::properties::STAT_REQUESTS_SIZE, statRequestsSize);
99  }
100 
101  unsigned int DataClientData::getStatRequestsSize(void){
102  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_REQUESTS_SIZE);
103  }
104 
105  void DataClientData::setStatResponsesSize(unsigned int statResponsesSize){
106  statProperties.set(HCE::handlers::properties::STAT_RESPONSES_SIZE, statResponsesSize);
107  }
108 
109  unsigned int DataClientData::getStatResponsesSize(void){
110  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_RESPONSES_SIZE);
111  }
112 
113  void DataClientData::setStatRequestsProcessed(unsigned int statRequestsProcessed){
114  statProperties.set(HCE::handlers::properties::STAT_REQUESTS_PROCESSED, statRequestsProcessed);
115  }
116 
117  unsigned int DataClientData::getStatRequestsProcessed(void){
118  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_REQUESTS_PROCESSED);
119  }
120 
121  void DataClientData::setStatResponsesProcessed(unsigned int statResponsesProcessed){
122  statProperties.set(HCE::handlers::properties::STAT_RESPONSES_PROCESSED, statResponsesProcessed);
123  }
124 
125  unsigned int DataClientData::getStatResponsesProcessed(void){
126  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_RESPONSES_PROCESSED);
127  }
128 
129  void DataClientData::setStatRequestsProcessedSize(unsigned int statRequestsProcessedSize){
130  statProperties.set(HCE::handlers::properties::STAT_REQUESTS_PROCESSED_SIZE, statRequestsProcessedSize);
131  }
132 
133  unsigned int DataClientData::getStatRequestsProcessedSize(void){
134  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_REQUESTS_PROCESSED_SIZE);
135  }
136 
137  void DataClientData::setStatResponsesProcessedSize(unsigned int statResponsesProcessedSize){
138  statProperties.set(HCE::handlers::properties::STAT_RESPONSES_PROCESSED_SIZE, statResponsesProcessedSize);
139  }
140 
141  unsigned int DataClientData::getStatResponsesProcessedSize(void){
142  return statProperties.get<unsigned int>(HCE::handlers::properties::STAT_RESPONSES_PROCESSED_SIZE);
143  }
144 
145  void DataClientData::setStatQPS(float statQPS){
146  statProperties.set(HCE::handlers::properties::STAT_QPS, statQPS);
147  }
148 
149  float DataClientData::getStatQPS(void){
150  return statProperties.get<float>(HCE::handlers::properties::STAT_QPS);
151  }
152 
153  void DataClientData::setStatRPS(float statRPS){
154  statProperties.set(HCE::handlers::properties::STAT_RPS, statRPS);
155  }
156 
157  float DataClientData::getStatRPS(void){
158  return statProperties.get<float>(HCE::handlers::properties::STAT_RPS);
159  }
160 
161  void DataClientData::setStatRequestsBandwidth(float statRequestsBandwidth){
162  statProperties.set(HCE::handlers::properties::STAT_REQUESTS_BANDWIDTH, statRequestsBandwidth);
163  }
164 
165  float DataClientData::getStatRequestsBandwidth(void){
166  return statProperties.get<float>(HCE::handlers::properties::STAT_REQUESTS_BANDWIDTH);
167  }
168 
169  void DataClientData::setStatResponsesBandwidth(float statResponsesBandwidth){
170  statProperties.set(HCE::handlers::properties::STAT_RESPONSES_BANDWIDTH, statResponsesBandwidth);
171  }
172 
173  float DataClientData::getStatResponsesBandwidth(void){
174  return statProperties.get<float>(HCE::handlers::properties::STAT_RESPONSES_BANDWIDTH);
175  }
176 
177  void DataClientData::setPstatsAt(int64_t pstatsAt){
178  handlerProperties.set(HCE::handlers::properties::PSTATS_AT, pstatsAt);
179  }
180 
181  int64_t DataClientData::getPstatsAt(void){
182  return handlerProperties.get<int64_t>(HCE::handlers::properties::PSTATS_AT);
183  }
184 
185  void DataClientData::setHeartbeatAt(int64_t heartbeatAt){
186  handlerProperties.set(HCE::handlers::properties::HEARTBEAT_AT, heartbeatAt);
187  }
188 
189  int64_t DataClientData::getHeartbeatAt(void){
190  return handlerProperties.get<int64_t>(HCE::handlers::properties::HEARTBEAT_AT);
191  }
192 
193  void DataClientData::setHeartbeatedAt(int64_t heartbeatedAt){
194  handlerProperties.set(HCE::handlers::properties::HEARTBEATED_AT, heartbeatedAt);
195  }
196 
197  int64_t DataClientData::getHeartbeatedAt(void){
198  return handlerProperties.get<int64_t>(HCE::handlers::properties::HEARTBEATED_AT);
199  }
200 
201  void DataClientData::checkInit(void){
203  if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS))
204  setStatRequests(0);
205 
206  if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES))
207  setStatResponses(0);
208 
209  if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS_SIZE))
210  setStatRequestsSize(0);
211 
212  if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES_SIZE))
213  setStatResponsesSize(0);
214 
215  if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS_PROCESSED))
216  setStatRequestsProcessed(0);
217 
218  if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES_PROCESSED))
219  setStatResponsesProcessed(0);
220 
221  if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS_PROCESSED_SIZE))
222  setStatRequestsProcessedSize(0);
223 
224  if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES_PROCESSED_SIZE))
225  setStatResponsesProcessedSize(0);
226 
227  if (!statProperties.isExist(HCE::handlers::properties::STAT_QPS))
228  setStatQPS(0);
229 
230  if (!statProperties.isExist(HCE::handlers::properties::STAT_RPS))
231  setStatRPS(0);
232 
233  if (!statProperties.isExist(HCE::handlers::properties::STAT_REQUESTS_BANDWIDTH))
234  setStatRequestsBandwidth(0);
235 
236  if (!statProperties.isExist(HCE::handlers::properties::STAT_RESPONSES_BANDWIDTH))
237  setStatResponsesBandwidth(0);
238  }
239 
240  void DataClientData::rebuildDataClientSocket(void){
241  const std::string connectionString = getConnectionString();
242  if(connectionString != "" && connectionString != "0"){
243  _dataClientSockConnected = rebuildClientConnection(_dataClientSock, connectionString, ZMQ_DEALER, getClientIdentity(), 0, NODE_MSG_READY);
244  }else{
245  logger.log(NODE_LOG_MODE_INFORMATION) << "Client connection was not rebuilt due connection string empty value [" << connectionString << "]" << flush;
246  }
247  }
248 
249  void DataClientData::handleExternalMessage(void){
251  zmsg clientMsg(*_dataClientSock);
252 
253  std::string addr((clientMsg.address())?clientMsg.address():"");
254 
255  logger.log(NODE_LOG_MODE_TRACE) << "DataClientData::handleExternalMessage addr: " << addr << flush;
256 
258  if(clientMsg.parts() == NODE_MSG_CMD_MIN_FIELDS){
260  if(addr.compare(0, NODE_MSG_HEARTBEAT.length(), NODE_MSG_HEARTBEAT)==0){
261  logger.log(NODE_LOG_MODE_DEBUG) << "HEARTBEAT message from server" << flush;
262 
264  setHeartbeatedAt(s_clock());
265 
266  if (getHeartbeatMode() == HEARBEAT_MODE_IMMEDIATE_AFTER_HB ||
267  getHeartbeatMode() == HEARBEAT_MODE_MIXED){
269  sendHeartbeat();
270 
271  logger.log(NODE_LOG_MODE_TRACE) << "HEARBEAT_MODE_IMMEDIATE_AFTER_HB || HEARBEAT_MODE_MIXED" << flush;
272  }
273 
274  }else{
275  logger.log(NODE_LOG_MODE_TRACE) << "ERROR control message" << endl << clientMsg.dump() << flush;
276  }
277  }else{
279  logger.log(NODE_LOG_MODE_TRACE) << "MSG FROM dataclient" << endl << clientMsg.dump() << flush;
280 
282  std::string clientIdentity((const char*)clientMsg.pop_front().c_str());
283  std::string messageId((const char*)clientMsg.pop_front().c_str());
284  std::string messageBody((const char*)clientMsg.pop_front().c_str());
285  std::string messageRoute((clientMsg.parts())?(const char*)clientMsg.pop_front().c_str():"");
286 
287  logger.log(NODE_LOG_MODE_TRACE) << "DataClientData::handleExternalMessage Recieved body: \"" << messageBody << "\"" << flush;
288  logger.log(NODE_LOG_MODE_TRACE) << "DataClientData::handleExternalMessage Recieved route: \"" << messageRoute << "\"" << flush;
289 
290  unsigned int statRequests = getStatRequests();
291  setStatRequests(++statRequests);
292  setStatRequestsSize(getStatRequestsSize() + messageBody.size());
293 
294  logger.log(NODE_LOG_MODE_DEBUG) << "RECEIVED REQUEST from upper server [" << clientIdentity << "], message [" << messageId << "] body " << messageBody.size()
295  << " bytes n=" << getStatRequests() << flush;
296 
298  zmsg msg(messageId.c_str());
300  msg.append(messageBody.c_str());
302  msg.append(messageRoute.c_str());
303 
304  if(getDataClientSockMode() == 0){
306  msg.send(*_dataClientSock);
307  unsigned int statResponses = getStatResponses();
308  setStatResponses(++statResponses);
309  setStatResponsesSize(getStatResponsesSize() + getStatRequestsSize());
310 
311  logger.log(NODE_LOG_MODE_TRACE) << " MSG TO dataclient" << endl << msg.dump() << flush;
312  logger.log(NODE_LOG_MODE_DEBUG) << " RESPONSE to dataclient socket from end-point node to upper server [" << clientIdentity << "] sent, n=" << getStatResponses() << flush;
313 
314  }else{
316  msg.send(*_inprocDataSock);
317  logger.log(NODE_LOG_MODE_DEBUG) << "REQUEST SENT to inproc_data socket" << flush;
318  }
319 
321  setHeartbeatedAt(s_clock());
322 
323  if (getHeartbeatMode() == HEARBEAT_MODE_ADAPTIVE){
324 
326  _shiftCounter.reset();
328  setHeartbeatAt(s_clock() + getHeartbeatDelay());
329  }
330  }
331  }
332 
333  void DataClientData::handleInternalMessage(void){
335  zmsg msg(*_inprocDataSock);
336 
337  logger.log(NODE_LOG_MODE_TRACE) << "MSG FROM inproc_data" << endl << msg.dump() << flush;
338 
340  msg.send(*_dataClientSock);
341 
342  unsigned int statResponses = getStatResponses();
343  setStatResponses(++statResponses);
344  setStatResponsesSize(getStatResponsesSize() + msg.size());
345 
346  logger.log(NODE_LOG_MODE_DEBUG) << "RESPONSE to dataclient socket from data node to upper server sent, n=" << getStatResponses() << flush;
347 
349  setHeartbeatAt(s_clock() + getHeartbeatDelay());
350  }
351 
352  void DataClientData::heartbit(void){
353  if (getHeartbeatMode() == HEARBEAT_MODE_FIXED_TIME ||
354  getHeartbeatMode() == HEARBEAT_MODE_MIXED){
355  if((s_clock() > getHeartbeatAt()) && _dataClientSockConnected){
356 
358  sendHeartbeat();
359 
360  logger.log(NODE_LOG_MODE_TRACE) << "HEARBEAT_MODE_FIXED_TIME || HEARBEAT_MODE_MIXED" << flush;
361  logger.log(NODE_LOG_MODE_DEBUG) << "SENT HEARTBIT from client [" << getClientIdentity() << "] to server" << flush;
362  }
363  }else if(getHeartbeatMode() == HEARBEAT_MODE_ADAPTIVE){
364  _shiftCounter.responceTimeStamp(getHeartbeatedAt());
365  _shiftCounter.recount(getHeartbeatTimeout());
366 
367  logger.log(NODE_LOG_MODE_TRACE) << "DataClientData::heartbit HEARBEAT_MODE_ADAPTIVE delta = " << (s_clock()- getHeartbeatAt()+getHeartbeatDelay())
368  << " _shiftCounter.getShift()= " << _shiftCounter.getShift() << flush;
369 
370  size_t delta = s_clock()-getHeartbeatAt()+getHeartbeatDelay();
371  if (delta > (getHeartbeatTimeout()+_shiftCounter.getShift()) && _dataClientSockConnected){
373  sendHeartbeat();
375  _shiftCounter.reset();
376 
377  logger.log(NODE_LOG_MODE_DEBUG) << "SENT HEARTBIT from client [" << getClientIdentity() << "] to server getHeartbeatAt() = '"
378  << Poco::DateTimeFormatter::format(Poco::LocalDateTime(Poco::Timestamp(Poco::Timestamp::fromEpochTime(getHeartbeatAt()/1000))), "%Y-%m-%d %H:%M:%S") << "'" << flush;
379  }
381  setHeartbeatedAt(s_clock());
382  }
383  }
384 
385  void DataClientData::sendHeartbeat(void){
387 // zmsg msg(NODE_MSG_HEARTBEAT.c_str());
388 // msg.append(getResourceUsageMessage().c_str());
389 // msg.send(*_dataClientSock);
390 
391  std::vector<ClientWorkerItem> queue;
392  HCE::handlers::HeartbeatMessage heartbeatMessage(queue, getResourceUsageMessage());
393  std::string message = heartbeatMessage.build(NODE_MSG_HEARTBEAT);
394  s_send(*_dataClientSock, message);
395 
396  logger.log(NODE_LOG_MODE_TRACE) << "SENT HEARTBIT MESSAGE: " << message << flush;
397 
399  setHeartbeatAt(s_clock() + getHeartbeatDelay());
400  }
401 
402  std::string DataClientData::getResourceUsageMessage(void) const{
404  HCE::drce::ResourceUsageExtractor::getInstance().update();
405  HCE::drce::ResourceUsage resourceUsage = HCE::drce::ResourceUsageExtractor::getInstance().extract();
406  HCE::drce::ResourceUsageConverter resourceUsageConverter(resourceUsage);
407  return resourceUsageConverter.toString();
408  }
409 
410  void DataClientData::logPeriodicStatistics(void){
411  if(s_clock() > getPstatsAt()){
412  setStatQPS((getStatRequests() - getStatRequestsProcessed()) / (float)((STAT_INTERVAL) / 1000.0));
413  setStatRPS(((getStatResponses() - getStatResponsesProcessed()) / (float)((STAT_INTERVAL) / 1000.0)));
414  setStatRequestsBandwidth((getStatRequestsSize() - getStatRequestsProcessedSize()) / (float)((STAT_INTERVAL) / 1000.0));
415  setStatResponsesBandwidth((getStatResponsesSize() - getStatResponsesProcessedSize()) / (float)((STAT_INTERVAL) / 1000.0));
416 
417  setStatRequestsProcessed(getStatRequests());
418  setStatResponsesProcessed(getStatResponses());
419  setStatRequestsProcessedSize(getStatRequestsSize());
420  setStatResponsesProcessedSize(getStatResponsesSize());
421 
422  logger.log(NODE_LOG_MODE_DEBUG) << "requests=" << getStatRequests() << ", responses=" << getStatResponses() << ", qps=" << getStatQPS() << ", rps=" << getStatRPS()
423  << ", requestsSize=" << getStatRequestsSize() << ", responsesSize=" << getStatResponsesSize() << ", requestsBandwidth="
424  << getStatRequestsBandwidth() << ", responsesBandwidth=" << getStatResponsesBandwidth() << flush;
425 
427  setPstatsAt(s_clock() + STAT_INTERVAL);
428  }
429  }
430 
431  std::string DataClientData::getPropertyInformation(void){
432  std::stringstream result;
433  result << NODE_ADMIN_ERROR_OK
434  << NODE_ADMIN_COMMAND_DELIMITER << getProperties() << "requests=" << getStatRequests()
435  << NODE_ADMIN_COMMAND_DELIMITER << "responses=" << getStatResponses()
436  << NODE_ADMIN_COMMAND_DELIMITER << "requestsSize=" << getStatRequestsSize()
437  << NODE_ADMIN_COMMAND_DELIMITER << "responsesSize=" << getStatResponsesSize()
438  << NODE_ADMIN_COMMAND_DELIMITER << "clientIdentity=" << getClientIdentity()
439  << NODE_ADMIN_COMMAND_DELIMITER << "dataClientMode=" << static_cast<int>(getDataClientSockMode())
440  << NODE_ADMIN_COMMAND_DELIMITER << "pstatsAt=" << static_cast<uint64_t>(getPstatsAt())
441  << NODE_ADMIN_COMMAND_DELIMITER << "heartbeatAt=" << static_cast<uint64_t>(getHeartbeatAt())
442  << NODE_ADMIN_COMMAND_DELIMITER << "heartbeatedAt=" << static_cast<uint64_t>(getHeartbeatedAt())
443  << NODE_ADMIN_COMMAND_DELIMITER << "statQPS=" << getStatQPS()
444  << NODE_ADMIN_COMMAND_DELIMITER << "statRPS=" << getStatRPS()
445  << NODE_ADMIN_COMMAND_DELIMITER << "statRequestsBandwidth=" << getStatRequestsBandwidth()
446  << NODE_ADMIN_COMMAND_DELIMITER << "statResponsesBandwidth=" << getStatResponsesBandwidth();
447  return result.str();
448  }
449 
450  void DataClientData::externalConnectionRefresh(void){
451 
452  if (getHeartbeatMode() != HEARBEAT_MODE_ADAPTIVE){
453 
454  if(((s_clock() - getHeartbeatedAt()) > getHeartbeatTimeout()) && getInProgress() && _dataClientSockConnected){
455  logger.log(NODE_LOG_MODE_INFORMATION) << "RECONNECT client [" << getClientIdentity() << "] because HEARTBIT not received during time of "
456  << (getHeartbeatTimeout()+_shiftCounter.getShift()) << " msec" << flush;
457 
459  rebuildDataClientSocket();
460 
462  setHeartbeatedAt(s_clock());
463 
465  setHeartbeatAt(s_clock() + getHeartbeatDelay());
466  }
467  }else{ // HEARBEAT_MODE_ADAPTIVE
468 
469  _shiftCounter.recount(getHeartbeatTimeout());
470 
471  logger.log(NODE_LOG_MODE_TRACE) << "DataClientData::externalConnectionRefresh (s_clock()-getHeartbeatedAt())= " << (s_clock()-getHeartbeatedAt())
472  << " getHeartbeatTimeout()= " << getHeartbeatTimeout() << " _shiftCounter.getShift()= " << _shiftCounter.getShift() << flush;
473 
474  if(((s_clock() - getHeartbeatedAt()) > (getHeartbeatTimeout()+_shiftCounter.getShift())) && getInProgress() && _dataClientSockConnected){
475  logger.log(NODE_LOG_MODE_INFORMATION) << "RECONNECT client [" << getClientIdentity() << "] because HEARTBIT not received during time of "
476  << (getHeartbeatTimeout()+_shiftCounter.getShift()) << " msec" << flush;
477 
479  rebuildDataClientSocket();
480 
482  setHeartbeatedAt(s_clock());
483 
484  _shiftCounter.reset();
485 
487  setHeartbeatAt(s_clock() + getHeartbeatDelay());
488  }
489  }
490  }
491 
492  std::string DataClientData::processAdminCommands(std::string& command, std::string& parameters){
493  std::string ret = NODE_ADMIN_ERROR_OK;
494 
495  logger.log(NODE_LOG_MODE_DEBUG) << "Admin command requested [" << command << "], parameters [" << parameters << "]" << flush;
496 
498  if(command == NODE_MSG_STAT){
499  std::stringstream result;
500  result << NODE_ADMIN_ERROR_OK << NODE_ADMIN_COMMAND_DELIMITER << "requests=" << getStatRequests() << ",responses=" << getStatResponses()
501  << ",requests_size=" << getStatRequestsSize() << ",responses_size=" << getStatResponsesSize();
502  ret = result.str();
503  }else if(command == NODE_MSG_REBUILD_CLIENT_CONNECTION){
505  rebuildDataClientSocket();
506  }else if(command == NODE_MSG_DISCONNECT_CLIENT_CONNECTION){
508  _dataClientSock->close();
509  _dataClientSockConnected = false;
510  }else if(command == NODE_MSG_UPDATE_SCHEMA){
511  setConnectionString(parameters);
512  }else if(command == NODE_MSG_PROPERTIES){
514  ret = getPropertyInformation();
515  }else if(command == NODE_MSG_HEARTBEAT_DELAY_GET){
516  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_DELAY + "=" + std::to_string(getHeartbeatDelay()/1000);
517  }else if(command == NODE_MSG_HEARTBEAT_DELAY_SET){
519  unsigned int heartbeatDelay = HEARTBEAT_DELAY;
521  std::istringstream(parameters) >> heartbeatDelay;
522  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_DELAY + "=" + std::to_string(getHeartbeatDelay()/1000);
524  setHeartbeatDelay(heartbeatDelay*1000);
525  }else if(command == NODE_MSG_HEARTBEAT_TIMEOUT_GET){
526  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_TIMEOUT + "=" + std::to_string(getHeartbeatTimeout()/1000);
527  }else if(command == NODE_MSG_HEARTBEAT_TIMEOUT_SET){
529  unsigned int heartbeatTimeout = HEARTBEAT_TIMEOUT;
531  std::istringstream(parameters) >> heartbeatTimeout;
532  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_TIMEOUT + "=" + std::to_string(getHeartbeatTimeout()/1000);
534  setHeartbeatTimeout(heartbeatTimeout*1000);
535  }else if(command == NODE_MSG_HEARTBEAT_MODE_GET){
536  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_MODE + "=" + std::to_string(getHeartbeatMode());
537  }else if(command == NODE_MSG_HEARTBEAT_MODE_SET){
539  int hbMode = HEARBEAT_MODE_DEFAULT;
541  std::istringstream(parameters) >> hbMode;
542  ret = NODE_ADMIN_ERROR_OK + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_PROPERTY_HEARTBEAT_MODE + "=" + std::to_string(getHeartbeatMode());
544  setHeartbeatMode(static_cast<unsigned int>(hbMode));
545  }else{
547  ret = NODE_ADMIN_ERROR_ERROR + NODE_ADMIN_COMMAND_DELIMITER + NODE_ADMIN_ERROR_UNSUPPORTED_COMMAND;
548  }
549 
550  return ret;
551  }
552 
553  void* DataClientData::process(void){
554  logger.log(NODE_LOG_MODE_INFORMATION) << NODE_ADMIN_ERROR_MAIN_MSG_LOOP_STARTED << flush;
555 
556  long delay = getPollTimeout();
557 
558  while(getInProgress()){
559  try{
561  std::vector<zmq::pollitem_t> items;
562  if(_dataClientSockConnected){
563  items ={{*_dataClientSock, 0, ZMQ_POLLIN, 0},{*_inprocDataSock, 0, ZMQ_POLLIN, 0},{*_inprocAdminSock, 0, ZMQ_POLLIN, 0}};
564  } else{
565  items ={{*_inprocDataSock, 0, ZMQ_POLLIN, 0},{*_inprocAdminSock, 0, ZMQ_POLLIN, 0}};
566  }
567 
569  if (delay){
570  delay = getPollTimeout();
571  }
572 
574  zmq::poll(&items[0], items.size(), delay);
575 
577  if(_dataClientSockConnected){
578  if(items[0].revents & ZMQ_POLLIN){
579  handleExternalMessage();
580  }
581  }
582 
584  if(items[items.size() - 2].revents & ZMQ_POLLIN){
585  handleInternalMessage();
586  }
587 
589  if(items[items.size() - 1].revents & ZMQ_POLLIN){
590  handleAdminMessage();
591  if(getInProgress() == false){
592  delay = 0;
593  setInProgress(true);
594  continue;
595  }
596  }
597 
598  if(getInProgress() == true && delay == 0){
599  setInProgress(false);
600  continue;
601  }
602 
604  heartbit();
605 
607  logPeriodicStatistics();
608 
610  sendPeriodicProperties();
611 
613  dumpRunTimeVariables();
614 
616  externalConnectionRefresh();
617 
618  }catch(zmq::error_t& error){
619  if (EINTR!=error.num()){
620  std::cout << "DataClientData E: " << error.what() << std::endl;
621  }
622  }catch(std::exception& e){
623  std::cout << "DataClientData E: " << e.what() << std::endl;
624  }catch(...){
625  std::cout << "DataClientData E: Unknown exception" << std::endl;
626  }
627  }
628 
629  logger.log(NODE_LOG_MODE_INFORMATION) << NODE_ADMIN_ERROR_MAIN_MSG_LOOP_STOPPED << flush;
630 
631  return NULL;
632  }
633  }
634 }