statepublisher_zmq.cpp

Go to the documentation of this file.

#include "statepublisher_zmq.h"
#include <QTextStream>
#include <QFile>
#include <QMessageBox>
#include <zmq.hpp>
#include <thread>

StatePublisherZMQ::StatePublisherZMQ()
  : _prev_dataplot(0), _thread(std::bind(&StatePublisherZMQ::run_thread, this)), _prev_time(0)
{
}

StatePublisherZMQ::~StatePublisherZMQ()
{
}

void StatePublisherZMQ::run_thread()
{
  zmq::context_t context(1);
  zmq::socket_t socket(context, ZMQ_REP);
  socket.bind("tcp://*:6665");

  while (true)
  {
    zmq::message_t request;

    //  Wait for next request from client
    socket.recv(&request);
    const char* request_data = (const char*)request.data();

    if (strncmp(request_data, "[get_data_names]", 16) == 0)
    {
      QString string_reply;

      _mutex.lock();

      std::map<QString, double>::iterator it;
      for (it = _current_data.begin(); it != _current_data.end(); it++)
      {
        string_reply.append(it->first + QString(" "));
      }
      _mutex.unlock();

      zmq::message_t reply(string_reply.size());
      socket.send(reply);
    }
    else if (strncmp(request_data, "[get_data]", 10) == 0)
    {
      bool abort = false;
      QString string_request = QString::fromUtf8(&request_data[10], request.size() - 10);
      QStringList names = string_request.split(';');

      _mutex.lock();
      QString string_reply;

      for (int i = 0; i < names.count(); i++)
      {
        std::map<QString, double>::iterator it = _current_data.find(names.at(i));
        if (it == _current_data.end())
        {
          abort = true;
          break;
        }
        else
        {
          double value = it->second;
          string_reply.append(QString::number(value) + QString(" "));
        }
      }
      _mutex.unlock();

      if (abort)
      {
        zmq::message_t reply(5);
        memcpy(reply.data(), "Error", 5);
        socket.send(reply);
      }
      else
      {
        zmq::message_t reply(string_reply.size());
        socket.send(reply);
      }
    }
    else
    {
      zmq::message_t reply(5);
      memcpy(reply.data(), "Error", 5);
      socket.send(reply);
    }
  }
}

void StatePublisherZMQ::updateState(PlotDataMap* datamap, double current_time)
{
  if (datamap == 0)
  {
    _prev_dataplot = datamap;
    _prev_time = current_time;
    _mutex.lock();
    _current_data.clear();
    _mutex.unlock();
    return;
  }

  PlotDataMap::iterator it;

  _mutex.lock();
  if (datamap != _prev_dataplot || current_time != _prev_time)
  {
    for (it = datamap->begin(); it != datamap->end(); it++)
    {
      const QString& name = it->first;
      PlotDataPtr plotdata = it->second;
      _current_data[name] = plotdata->getY(current_time);
    }
  }
  _mutex.unlock();

  _prev_dataplot = datamap;
  _prev_time = current_time;
}