<commit_before>#include "n3rvservice.hpp"
#include <thread>
#include <regex>
#include <iostream>

namespace n3rv {


  service::service(                
                   std::string controller_host, 
                   int controller_port) {


        this->namespace_ = "";
        this->service_class = "";
        this->name = "";

        this->poll_timeout = 1000;

        this->ll = new logger(LOGLV_NORM);
        this->zctx = zmq::context_t(1);
        this->controller_host = controller_host;
        this->controller_port = controller_port;

        // Controller has 2 I/O channels: 1 for req/resp and one for multicast 
        // (the later mostly for directory updates). Therefore, each service must 
        // connect to the 2 of them.

        this->ctlr_ch1 = new qhandler();
        this->ctlr_ch2 = new qhandler();

        this->ctlr_ch1->cid = randstr(10);
        this->ctlr_ch2->cid = randstr(10);

        this->connections[this->ctlr_ch1->cid].socket = new zmq::socket_t(this->zctx,ZMQ_REQ);
        this->connections[this->ctlr_ch2->cid].socket = new zmq::socket_t(this->zctx,ZMQ_SUB);

        // Connects to controller on both channels.
        std::stringstream ss;
        ss << "tcp://" << controller_host << ":" << controller_port;

        this->ll->log(LOGLV_NORM, "Connecting to " + controller_host + " controller.." );
        this->connections[this->ctlr_ch1->cid].socket->connect(ss.str().c_str());

        ss.str(std::string());
        ss.clear();
        
        ss << "tcp://" << controller_host << ":" << (controller_port + 1);
        this->connections[this->ctlr_ch2->cid].socket->connect(ss.str().c_str());
        this->connections[this->ctlr_ch2->cid].socket->setsockopt(ZMQ_SUBSCRIBE,"",0);
        
        // Attaches controller's channel 2 to directory updater callback. 
        this->attach(this->ctlr_ch2,this->directory_update);

  }

  void service::set_uid(std::string namespace_, std::string service_class, std::string name) {

    this->namespace_ = namespace_;
    this->service_class = service_class;
    this->name = name;

  }

  void service::set_uid(std::string uid) {

    std::vector<std::string> uid_parts;
    std::regex dotsplit("\\.");
    std::sregex_token_iterator iter(uid.begin(),
    uid.end(),
    dotsplit,
    -1);
    std::sregex_token_iterator end;
    for ( ; iter != end; ++iter) uid_parts.emplace_back(*iter);

    if (uid_parts.size() != 3) {
      this->ll->log(LOGLV_CRIT, "cannot set UID: Invalid string");
      return;
    }

    this->namespace_ = uid_parts[0];
    this->service_class = uid_parts[1];
    this->name = uid_parts[2];

  }



  qhandler* service::connect(std::string name, int connection_type, qhandler* hdlref) {

    qhandler* hdl;
    
    if (hdlref == nullptr) {
      hdl = new qhandler();
      hdl->cid = randstr(9);
    }

    else {
      hdl = hdlref;
    }
    
    this->ll->log(LOGLV_NORM,"connecting to " + name);
    binding* b =  blookup(this->directory, name);

    if (b != nullptr) {
      
      qserv* s = (qserv*) b->parent;

      this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, connection_type );
      std::stringstream ep;
      ep << "tcp://" << s->ip << ":" << b->port;
      this->connections[hdl->cid].socket->connect(ep.str().c_str());

      //Adds sockopt if zmq socket type is ZMQ_SUB
      if (connection_type == ZMQ_SUB) {
        this->connections[hdl->cid].socket->setsockopt(ZMQ_SUBSCRIBE,"",0);
      }

      hdl->peer_uid = s->namespace_ + "." + 
                      s->service_class + "." + 
                      s->node_name + "." +
                      b->name;
                      
    }

    else {
      this->ll->log(LOGLV_WARN,"peer not found in directory, deferring connection..");
      n3rv::qdef cqd;
      cqd.name = name;
      cqd.socket_type = connection_type;
      cqd.hdl  = hdl;
      this->deferred.emplace_back(cqd);
    }

    return hdl;

  }

  qhandler* service::zbind(std::string bind_name, std::string endpoint, int bind_type ) { 
    
     qhandler* hdl = new qhandler();
     hdl->cid = randstr(8);

    this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, bind_type);
    this->connections[hdl->cid].socket->bind(endpoint.c_str());

    return hdl;

  }

  qhandler* service::bind(std::string bind_name, std::string ip , int bind_type, int port ) {

    qhandler* hdl = new qhandler();
    hdl->cid = randstr(8);

    this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, bind_type);

    //Port Autobinding (if 0) 
    if (port == 0) {

      srand(time(nullptr));
      port = rand() % 50000 + 10000;

      try {
        std::stringstream ep;
        ep << "tcp://" << ip << ":" << port;
        this->connections[hdl->cid].socket->bind(ep.str().c_str());
      }

      catch(const zmq::error_t& e) {
        return this->bind(bind_name, ip, bind_type, 0);
      }


    }

    else {
      std::stringstream ep;
      ep << "tcp://" << ip << ":" << port;
      this->connections[hdl->cid].socket->bind(ep.str().c_str());
    }
     
    this->subscribe(bind_name, port);
    return hdl;

  }


  service::~service() {

  }

  int service::initialize() {

  }

  void service::map_callbacks() {

  }


  zmq::pollitem_t* service::refresh_pollitems() {

    this->last_nconn = this->connections.size();
    this->last_connlist.clear();    

    zmq::pollitem_t* items  = (zmq::pollitem_t*) malloc(sizeof(zmq::pollitem_t) * this->last_nconn );

    int i = 0;
    for(std::map<std::string, n3rv::qconn>::iterator iter = this->connections.begin(); 
        iter != this->connections.end(); 
        ++iter) {

      if (i >= last_nconn - 1 ) break;

      std::string k = iter->first;
      zmq::socket_t* s = iter->second.socket;

      if (k == this->ctlr_ch1->cid) continue;

      this->last_connlist.emplace_back(k);
      
      //JEEEZZZZZ, that sucks !
      items[i].socket = static_cast<void*> (*s);
      items[i].fd = 0;
      items[i].events = ZMQ_POLLIN;
      items[i].revents = 0;
      i++;

    }

    return items;

  }


  int service::run() {

    zmq::message_t message;

    /** Main service loop, listens to open connections and forwards 
     * the data to the correct handler.
     */
    while(1) {

      zmq::pollitem_t* items = this->refresh_pollitems(); 
      try {
        zmq::poll (items,this->last_connlist.size(), this->poll_timeout);
      }
      catch(const zmq::error_t& e) {}

       for (int j=0;j < this->last_connlist.size(); j++) {
         
         if (items[j].revents & ZMQ_POLLIN) {

            this->connections[this->last_connlist[j]].socket->recv(&message);

            if ( this->chmap.find(this->last_connlist[j]) != this->chmap.end() ) {
              (*this->chmap[this->last_connlist[j]])(this, &message);
            }
         }
       }

       this->hkloop();      
    }

  }

  std::thread* service::run_async() {

    std::thread* t = new std::thread([this]() {  this->run(); });
    t->detach();
    return t;

  }

  int service::terminate() {

      for (auto n3sock: this->connections) {

        try {
          n3sock.second.socket->close();
        }
        catch (const zmq::error_t& e) {}        
      }
      //this->zctx.destroy();
  }


  void service::hkloop() {

  }

  int service::attach(qhandler* hdl, fctptr callback) {
    this->chmap[hdl->cid]  = callback;
  }

  int service::attach(qhandler* hdl, std::string callback_name) {
    this->chmap[hdl->cid] = this->cbmap[callback_name];
  }


  int service::fetch_topology() {

      n3rv::message m;  
      m.action = "topology";
      m.payload = "";
      this->send(this->ctlr_ch1,m,0);

      //fetches response
      zmq::message_t r1;
      this->connections[this->ctlr_ch1->cid].socket->recv(&r1);

      char * rawmsg = (char*) calloc(r1.size()+1, sizeof(char)   );
      memcpy(rawmsg, r1.data(), r1.size() );
      std::string topo_resp = rawmsg;
      std::cout << topo_resp << std::endl;
      
      if (topo_resp != "ERR: NO TOPOLOGY") {
        topology* t = topology::parse(topo_resp);
        this->load_topology(t);
      }
      return 0;
  }



  std::map<std::string, qhandler*> service::load_topology(std::string path) {
    topology* t = topology::load(path);
    this->load_topology(t);
  }

  std::map<std::string, qhandler*> service::load_topology(topology* t) {

    std::map<std::string, qhandler*> res;  
    std::map<std::string, int> zmq_sockmap;

    zmq_sockmap["ZMQ_PUB"] = ZMQ_PUB;
    zmq_sockmap["ZMQ_SUB"] = ZMQ_SUB;
    zmq_sockmap["ZMQ_REQ"] = ZMQ_REQ;
    zmq_sockmap["ZMQ_REP"] = ZMQ_REP;
    zmq_sockmap["ZMQ_XPUB"] = ZMQ_XPUB;
    zmq_sockmap["ZMQ_XSUB"] = ZMQ_XSUB;
    zmq_sockmap["ZMQ_XREQ"] = ZMQ_XREQ;
    zmq_sockmap["ZMQ_XREP"] = ZMQ_XREP;
    zmq_sockmap["ZMQ_PULL"] = ZMQ_PULL;
    zmq_sockmap["ZMQ_PUSH"] = ZMQ_PUSH;
    zmq_sockmap["ZMQ_DEALER"] = ZMQ_DEALER;
    zmq_sockmap["ZMQ_ROUTER"] = ZMQ_ROUTER;
    zmq_sockmap["ZMQ_PAIR"] = ZMQ_PAIR;

    for (auto& nmap : t->nodes) {

      auto& key = nmap.first;
      auto& n = nmap.second;

      if ( key == this->service_class  ) {

        for (auto& b: n.bindings) {
          this->bind(this->name + "." + b.name, "0.0.0.0", b.port, zmq_sockmap[b.type] );
        }

        for (auto c: n.connections) {
          qhandler* h = this->connect(c.name, zmq_sockmap[c.type]);
          res[c.name] = h;
        }

        for (auto& cb: n.callbacks) {
          this->attach(res[cb.connection_name], cb.callback_name);
        }
        break;
      }
    }

    return res;
  }



  int service::subscribe(std::string binding_name, int port) {

      n3rv::message m;
      
      m.sender = name;
      m.action = "subscribe";
      m.args.emplace_back(this->namespace_);
      m.args.emplace_back(this->service_class);
      m.args.emplace_back(this->name);
      
      m.args.emplace_back(binding_name);

      std::stringstream ss;
      ss << port;

      m.args.emplace_back(ss.str());
      m.payload = "";

      std::string to_send = serialize_msg(m);

      zmq::message_t req (to_send.size());
      memcpy (req.data(), to_send.data() , to_send.size());

      this->connections[this->ctlr_ch1->cid].socket->send(req);

      //Waits for response
      zmq::message_t r1;
      this->connections[this->ctlr_ch1->cid].socket->recv(&r1);
      return 0;

  }


  n3rv::qconn& service::get_connection(qhandler& hdl) {
     return this->connections[hdl.cid];
  }


  int service::send(qhandler* hdl, std::string& data, int flags = 0) {

    zmq::message_t msg(data.size());
    memcpy(msg.data(), data.data(), data.size());
    this->connections[hdl->cid].socket->send(msg);

    return 0;

  }

  int service::send(qhandler* hdl, void* data, size_t size, int flags = 0) {

    zmq::message_t msg(size);
    memcpy(msg.data(), data, size);
    this->connections[hdl->cid].socket->send(msg);

    return 0;

  }

  int service::send(qhandler* hdl, message& msg, int flags=0) {
    if (msg.sender == "") msg.sender = this->name;
    std::string msg_ser = serialize_msg(msg);
    return this->send(hdl,msg_ser, flags);
  }

  int service::send(qhandler* hdl, zmq::message_t* zmsg, int flags =0) {
    this->connections[hdl->cid].socket->send(*zmsg);
  }

  int service::check_deferred() {

    this->ll->log(n3rv::LOGLV_XDEBUG,"checking deferred list..");
    std::vector<n3rv::qdef> deferred_iter(this->deferred);

    int res = 0;
    for (auto def: deferred_iter) {

      binding* b = blookup(this->directory,def.name);
      if (b != nullptr) {
        
        this->ll->log(n3rv::LOGLV_NORM,"reconnecting to " + def.name);
        this->connect(def.name, def.socket_type,def.hdl);
        this->deferred.erase(this->deferred.begin() + res);

        res++;
      }

    }
    return res;
  }


  void* service::directory_update(void* objref, zmq::message_t* dirmsg) {

    service* self = (service*) objref;

    self->ll->log(LOGLV_DEBUG,"updating Directory..");
    std::string dirstring(static_cast<char*>(dirmsg->data()), dirmsg->size());
    self->directory = parse_directory(dirstring);   
    self->check_deferred();

  }

  void service::set_poll_timeout(int poll_timeout) {
    this->poll_timeout = poll_timeout;
  }

}<commit_msg>service: Additional security mechanisms for bind()<commit_after>#include "n3rvservice.hpp"
#include <thread>
#include <regex>
#include <iostream>

namespace n3rv {


  service::service(                
                   std::string controller_host, 
                   int controller_port) {


        this->namespace_ = "";
        this->service_class = "";
        this->name = "";

        this->poll_timeout = 1000;

        this->ll = new logger(LOGLV_NORM);
        this->zctx = zmq::context_t(1);
        this->controller_host = controller_host;
        this->controller_port = controller_port;

        // Controller has 2 I/O channels: 1 for req/resp and one for multicast 
        // (the later mostly for directory updates). Therefore, each service must 
        // connect to the 2 of them.

        this->ctlr_ch1 = new qhandler();
        this->ctlr_ch2 = new qhandler();

        this->ctlr_ch1->cid = randstr(10);
        this->ctlr_ch2->cid = randstr(10);

        this->connections[this->ctlr_ch1->cid].socket = new zmq::socket_t(this->zctx,ZMQ_REQ);
        this->connections[this->ctlr_ch2->cid].socket = new zmq::socket_t(this->zctx,ZMQ_SUB);

        // Connects to controller on both channels.
        std::stringstream ss;
        ss << "tcp://" << controller_host << ":" << controller_port;

        this->ll->log(LOGLV_NORM, "Connecting to " + controller_host + " controller.." );
        this->connections[this->ctlr_ch1->cid].socket->connect(ss.str().c_str());

        ss.str(std::string());
        ss.clear();
        
        ss << "tcp://" << controller_host << ":" << (controller_port + 1);
        this->connections[this->ctlr_ch2->cid].socket->connect(ss.str().c_str());
        this->connections[this->ctlr_ch2->cid].socket->setsockopt(ZMQ_SUBSCRIBE,"",0);
        
        // Attaches controller's channel 2 to directory updater callback. 
        this->attach(this->ctlr_ch2,this->directory_update);

  }

  void service::set_uid(std::string namespace_, std::string service_class, std::string name) {

    this->namespace_ = namespace_;
    this->service_class = service_class;
    this->name = name;

  }

  void service::set_uid(std::string uid) {

    std::vector<std::string> uid_parts;
    std::regex dotsplit("\\.");
    std::sregex_token_iterator iter(uid.begin(),
    uid.end(),
    dotsplit,
    -1);
    std::sregex_token_iterator end;
    for ( ; iter != end; ++iter) uid_parts.emplace_back(*iter);

    if (uid_parts.size() != 3) {
      this->ll->log(LOGLV_CRIT, "cannot set UID: Invalid string");
      return;
    }

    this->namespace_ = uid_parts[0];
    this->service_class = uid_parts[1];
    this->name = uid_parts[2];

  }



  qhandler* service::connect(std::string name, int connection_type, qhandler* hdlref) {

    qhandler* hdl;
    
    if (hdlref == nullptr) {
      hdl = new qhandler();
      hdl->cid = randstr(9);
    }

    else {
      hdl = hdlref;
    }
    
    this->ll->log(LOGLV_NORM,"connecting to " + name);
    binding* b =  blookup(this->directory, name);

    if (b != nullptr) {
      
      qserv* s = (qserv*) b->parent;

      this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, connection_type );
      std::stringstream ep;
      ep << "tcp://" << s->ip << ":" << b->port;
      this->connections[hdl->cid].socket->connect(ep.str().c_str());

      //Adds sockopt if zmq socket type is ZMQ_SUB
      if (connection_type == ZMQ_SUB) {
        this->connections[hdl->cid].socket->setsockopt(ZMQ_SUBSCRIBE,"",0);
      }

      hdl->peer_uid = s->namespace_ + "." + 
                      s->service_class + "." + 
                      s->node_name + "." +
                      b->name;
                      
    }

    else {
      this->ll->log(LOGLV_WARN,"peer not found in directory, deferring connection..");
      n3rv::qdef cqd;
      cqd.name = name;
      cqd.socket_type = connection_type;
      cqd.hdl  = hdl;
      this->deferred.emplace_back(cqd);
    }

    return hdl;

  }

  qhandler* service::zbind(std::string bind_name, std::string endpoint, int bind_type ) { 
    
     qhandler* hdl = new qhandler();
     hdl->cid = randstr(8);

    this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, bind_type);
    this->connections[hdl->cid].socket->bind(endpoint.c_str());

    return hdl;

  }

  qhandler* service::bind(std::string bind_name, std::string ip , int bind_type, int port ) {

    if (this->namespace_ == "" || 
        this->service_class == "" || 
        this->name == "") {
      this->ll->log(LOGLV_CRIT, "Cannot create binding: Please ensure to call set_uid() beforehand.");
      return nullptr;
    }

    if (bind_name == "") {
       this->ll->log(LOGLV_CRIT, "Cannot bind with an empty name");
      return nullptr;
    }
    
    qhandler* hdl = new qhandler();
    hdl->cid = randstr(8);

    this->connections[hdl->cid].socket = new zmq::socket_t(this->zctx, bind_type);

    //Port Autobinding (if 0) 
    if (port == 0) {

      srand(time(nullptr));
      port = rand() % 50000 + 10000;

      try {
        std::stringstream ep;
        ep << "tcp://" << ip << ":" << port;
        this->connections[hdl->cid].socket->bind(ep.str().c_str());
      }

      catch(const zmq::error_t& e) {
        return this->bind(bind_name, ip, bind_type, 0);
      }


    }

    else {
      std::stringstream ep;
      ep << "tcp://" << ip << ":" << port;
      this->connections[hdl->cid].socket->bind(ep.str().c_str());
    }
     
    this->subscribe(bind_name, port);
    return hdl;

  }


  service::~service() {

  }

  int service::initialize() {

  }

  void service::map_callbacks() {

  }


  zmq::pollitem_t* service::refresh_pollitems() {

    this->last_nconn = this->connections.size();
    this->last_connlist.clear();    

    zmq::pollitem_t* items  = (zmq::pollitem_t*) malloc(sizeof(zmq::pollitem_t) * this->last_nconn );

    int i = 0;
    for(std::map<std::string, n3rv::qconn>::iterator iter = this->connections.begin(); 
        iter != this->connections.end(); 
        ++iter) {

      if (i >= last_nconn - 1 ) break;

      std::string k = iter->first;
      zmq::socket_t* s = iter->second.socket;

      if (k == this->ctlr_ch1->cid) continue;

      this->last_connlist.emplace_back(k);
      
      //JEEEZZZZZ, that sucks !
      items[i].socket = static_cast<void*> (*s);
      items[i].fd = 0;
      items[i].events = ZMQ_POLLIN;
      items[i].revents = 0;
      i++;

    }

    return items;

  }


  int service::run() {

    zmq::message_t message;

    /** Main service loop, listens to open connections and forwards 
     * the data to the correct handler.
     */
    while(1) {

      zmq::pollitem_t* items = this->refresh_pollitems(); 
      try {
        zmq::poll (items,this->last_connlist.size(), this->poll_timeout);
      }
      catch(const zmq::error_t& e) {}

       for (int j=0;j < this->last_connlist.size(); j++) {
         
         if (items[j].revents & ZMQ_POLLIN) {

            this->connections[this->last_connlist[j]].socket->recv(&message);

            if ( this->chmap.find(this->last_connlist[j]) != this->chmap.end() ) {
              (*this->chmap[this->last_connlist[j]])(this, &message);
            }
         }
       }

       this->hkloop();      
    }

  }

  std::thread* service::run_async() {

    std::thread* t = new std::thread([this]() {  this->run(); });
    t->detach();
    return t;

  }

  int service::terminate() {

      for (auto n3sock: this->connections) {

        try {
          n3sock.second.socket->close();
        }
        catch (const zmq::error_t& e) {}        
      }
      //this->zctx.destroy();
  }


  void service::hkloop() {

  }

  int service::attach(qhandler* hdl, fctptr callback) {
    this->chmap[hdl->cid]  = callback;
  }

  int service::attach(qhandler* hdl, std::string callback_name) {
    this->chmap[hdl->cid] = this->cbmap[callback_name];
  }


  int service::fetch_topology() {

      n3rv::message m;  
      m.action = "topology";
      m.payload = "";
      this->send(this->ctlr_ch1,m,0);

      //fetches response
      zmq::message_t r1;
      this->connections[this->ctlr_ch1->cid].socket->recv(&r1);

      char * rawmsg = (char*) calloc(r1.size()+1, sizeof(char)   );
      memcpy(rawmsg, r1.data(), r1.size() );
      std::string topo_resp = rawmsg;
      std::cout << topo_resp << std::endl;
      
      if (topo_resp != "ERR: NO TOPOLOGY") {
        topology* t = topology::parse(topo_resp);
        this->load_topology(t);
      }
      return 0;
  }



  std::map<std::string, qhandler*> service::load_topology(std::string path) {
    topology* t = topology::load(path);
    this->load_topology(t);
  }

  std::map<std::string, qhandler*> service::load_topology(topology* t) {

    std::map<std::string, qhandler*> res;  
    std::map<std::string, int> zmq_sockmap;

    zmq_sockmap["ZMQ_PUB"] = ZMQ_PUB;
    zmq_sockmap["ZMQ_SUB"] = ZMQ_SUB;
    zmq_sockmap["ZMQ_REQ"] = ZMQ_REQ;
    zmq_sockmap["ZMQ_REP"] = ZMQ_REP;
    zmq_sockmap["ZMQ_XPUB"] = ZMQ_XPUB;
    zmq_sockmap["ZMQ_XSUB"] = ZMQ_XSUB;
    zmq_sockmap["ZMQ_XREQ"] = ZMQ_XREQ;
    zmq_sockmap["ZMQ_XREP"] = ZMQ_XREP;
    zmq_sockmap["ZMQ_PULL"] = ZMQ_PULL;
    zmq_sockmap["ZMQ_PUSH"] = ZMQ_PUSH;
    zmq_sockmap["ZMQ_DEALER"] = ZMQ_DEALER;
    zmq_sockmap["ZMQ_ROUTER"] = ZMQ_ROUTER;
    zmq_sockmap["ZMQ_PAIR"] = ZMQ_PAIR;

    for (auto& nmap : t->nodes) {

      auto& key = nmap.first;
      auto& n = nmap.second;

      if ( key == this->service_class  ) {

        for (auto& b: n.bindings) {
          this->bind(this->name + "." + b.name, "0.0.0.0", b.port, zmq_sockmap[b.type] );
        }

        for (auto c: n.connections) {
          qhandler* h = this->connect(c.name, zmq_sockmap[c.type]);
          res[c.name] = h;
        }

        for (auto& cb: n.callbacks) {
          this->attach(res[cb.connection_name], cb.callback_name);
        }
        break;
      }
    }

    return res;
  }



  int service::subscribe(std::string binding_name, int port) {

      n3rv::message m;
      
      m.sender = name;
      m.action = "subscribe";
      m.args.emplace_back(this->namespace_);
      m.args.emplace_back(this->service_class);
      m.args.emplace_back(this->name);
      
      m.args.emplace_back(binding_name);

      std::stringstream ss;
      ss << port;

      m.args.emplace_back(ss.str());
      m.payload = "";

      std::string to_send = serialize_msg(m);

      zmq::message_t req (to_send.size());
      memcpy (req.data(), to_send.data() , to_send.size());

      this->connections[this->ctlr_ch1->cid].socket->send(req);

      //Waits for response
      zmq::message_t r1;
      this->connections[this->ctlr_ch1->cid].socket->recv(&r1);
      return 0;

  }


  n3rv::qconn& service::get_connection(qhandler& hdl) {
     return this->connections[hdl.cid];
  }


  int service::send(qhandler* hdl, std::string& data, int flags = 0) {

    zmq::message_t msg(data.size());
    memcpy(msg.data(), data.data(), data.size());
    this->connections[hdl->cid].socket->send(msg);

    return 0;

  }

  int service::send(qhandler* hdl, void* data, size_t size, int flags = 0) {

    zmq::message_t msg(size);
    memcpy(msg.data(), data, size);
    this->connections[hdl->cid].socket->send(msg);

    return 0;

  }

  int service::send(qhandler* hdl, message& msg, int flags=0) {
    if (msg.sender == "") msg.sender = this->name;
    std::string msg_ser = serialize_msg(msg);
    return this->send(hdl,msg_ser, flags);
  }

  int service::send(qhandler* hdl, zmq::message_t* zmsg, int flags =0) {
    this->connections[hdl->cid].socket->send(*zmsg);
  }

  int service::check_deferred() {

    this->ll->log(n3rv::LOGLV_XDEBUG,"checking deferred list..");
    std::vector<n3rv::qdef> deferred_iter(this->deferred);

    int res = 0;
    for (auto def: deferred_iter) {

      binding* b = blookup(this->directory,def.name);
      if (b != nullptr) {
        
        this->ll->log(n3rv::LOGLV_NORM,"reconnecting to " + def.name);
        this->connect(def.name, def.socket_type,def.hdl);
        this->deferred.erase(this->deferred.begin() + res);

        res++;
      }

    }
    return res;
  }


  void* service::directory_update(void* objref, zmq::message_t* dirmsg) {

    service* self = (service*) objref;

    self->ll->log(LOGLV_DEBUG,"updating Directory..");
    std::string dirstring(static_cast<char*>(dirmsg->data()), dirmsg->size());
    self->directory = parse_directory(dirstring);   
    self->check_deferred();

  }

  void service::set_poll_timeout(int poll_timeout) {
    this->poll_timeout = poll_timeout;
  }

}
{'fraction_non_alphanumeric': 0.16784594355551982, 'fraction_numerical': 0.007397282302806143, 'mean_word_length': 2.5515419760137066, 'pattern_counts': {'":': 10, '<': 121, '<?xml version=': 0, '>': 390, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7515790', 'n_tokens_mistral': 9196, 'n_tokens_neox': 8688, 'n_words': 1510}