EndpointDiscovery.cpp

Go to the documentation of this file.

#include "EndpointDiscovery.h"
#include "madara/utility/Utility.h"
#include "madara/logger/GlobalLogger.h"

namespace utility = madara::utility;
typedef madara::knowledge::KnowledgeRecord::Integer   Integer;

madara::filters::EndpointDiscovery::EndpointDiscovery (
  const std::string & prefix,
  knowledge::KnowledgeRecord::Integer heart_beat)
  : initialized_ (false), prefix_ (prefix),
    heart_beat_ (heart_beat * 1000000000), last_clear_ (0)
{
}

madara::filters::EndpointDiscovery::~EndpointDiscovery ()
{
}


void
madara::filters::EndpointDiscovery::filter (
  knowledge::KnowledgeMap & records,
  const transport::TransportContext & transport_context,
  knowledge::Variables & vars)
{
  if (!initialized_)
  {
    endpoints_.set_name (prefix_, vars);
    initialized_ = true;
  }

  madara_logger_log (vars.get_context ()->get_logger (),
    logger::LOG_MAJOR,
    "EndpointDiscovery::filter:" \
    " Processing a new update with %d records\n",
    (int)records.size ());

  std::string endpoint (transport_context.get_endpoint ());
  Integer cur_time = (Integer)transport_context.get_current_time ();

  endpoints_.set (endpoint, cur_time);

  if (heart_beat_ > 0 && last_clear_ != cur_time)
  {
    std::vector <std::string> keys;
    endpoints_.sync_keys ();

    endpoints_.keys (keys);

    for (size_t i = 0; i < keys.size (); ++i)
    {
      // if the current endpoint is old, erase it
      if (cur_time - endpoints_[keys[i]].to_integer () > heart_beat_)
      {
        madara_logger_log (vars.get_context ()->get_logger (),
          logger::LOG_MINOR,
          "EndpointDiscovery::filter:" \
          " Erasing endpoint %s\n",
          keys[i].c_str ());

        endpoints_.erase (keys[i]);
      }
    }

    last_clear_ = cur_time;
  }
}