SpliceTransportReadThread.cpp

Go to the documentation of this file.

#include "madara/transport/splice/SpliceTransportReadThread.h"
#include "madara/knowledge/UpdateTypes.h"
#include "madara/utility/Utility.h"
#include "madara/logger/GlobalLogger.h"
 
namespace logger = madara::logger;
namespace utility = madara::utility;
 
#include <iostream>
#include <sstream>
 
madara::transport::SpliceReadThread::SpliceReadThread(const std::string& id,
    const TransportSettings& settings,
    madara::knowledge::ThreadSafeContext& context,
    Knowledge::UpdateDataReader_ptr& update_reader,
    Knowledge::UpdateDataWriter_ptr& update_writer,
    BandwidthMonitor& send_monitor, BandwidthMonitor& receive_monitor,
    PacketScheduler& packet_scheduler)
  : id_(id),
    settings_(settings),
    context_(&context),
    update_reader_(update_reader),
    update_writer_(update_writer),
    send_monitor_(send_monitor),
    receive_monitor_(receive_monitor),
    packet_scheduler_(packet_scheduler)
{
}
 
void madara::transport::SpliceReadThread::init(
    knowledge::KnowledgeBase& knowledge)
{
  context_ = &(knowledge.get_context());
 
  // setup the receive buffer
  if (settings_.queue_length > 0)
    buffer_ = new char[settings_.queue_length];
 
  if (context_)
  {
    // check for an on_data_received ruleset
    if (settings_.on_data_received_logic.length() != 0)
    {
#ifndef _MADARA_NO_KARL_
      madara_logger_log(context_->get_logger(), logger::LOG_MAJOR,
          "UdpTransportReadThread::init:"
          " setting rules to %s\n",
          settings_.on_data_received_logic.c_str());
 
      madara::expression::Interpreter interpreter;
      on_data_received_ = context_->compile(settings_.on_data_received_logic);
#endif  // _MADARA_NO_KARL_
    }
    else
    {
      madara_logger_log(context_->get_logger(), logger::LOG_MINOR,
          "UdpTransportReadThread::init:"
          " no permanent rules were set\n");
    }
  }
}
 
void madara::transport::SpliceReadThread::cleanup(void) {}
 
void madara::transport::SpliceReadThread::rebroadcast(const char* print_prefix,
    MessageHeader* header, const knowledge::KnowledgeMap& records)
{
  int64_t buffer_remaining = (int64_t)settings_.queue_length;
  char* buffer = buffer_.get_ptr();
  unsigned long result = prep_rebroadcast(*context_, buffer, buffer_remaining,
      settings_, print_prefix, header, records, packet_scheduler_);
 
  if (result > 0)
  {
    ssize_t bytes_sent(result + sizeof(Knowledge::Update));
    DDS::ReturnCode_t dds_result;
    DDS::InstanceHandle_t handle;
 
    // get the maximum quality from the updates
    uint32_t quality = knowledge::max_quality(records);
 
    unsigned long long cur_clock = context_->get_clock();
 
    Knowledge::Update data;
 
    data.buffer =
        Knowledge::seq_oct(result, result, (unsigned char*)buffer_.get_ptr());
    data.clock = cur_clock;
    data.quality = quality;
    data.updates = DDS::ULong(records.size());
    data.originator = DDS::string_dup(id_.c_str());
    data.type = madara::transport::MULTIASSIGN;
    data.ttl = settings_.get_rebroadcast_ttl();
    data.timestamp = utility::get_time();
    data.madara_id = DDS::string_dup(MADARA_IDENTIFIER);
 
    handle = update_writer_->register_instance(data);
    dds_result = update_writer_->write(data, handle);
 
    madara_logger_log(context_->get_logger(), logger::LOG_MAJOR,
        "%s:"
        " Sent packet of size %d\n",
        print_prefix, bytes_sent);
 
    send_monitor_.add((uint32_t)bytes_sent);
 
    madara_logger_log(context_->get_logger(), logger::LOG_MINOR,
        "%s:"
        " Send bandwidth = %d B/s\n",
        print_prefix, send_monitor_.get_bytes_per_second());
  }
}
 
void madara::transport::SpliceReadThread::run(void)
{
  DDS::SampleInfoSeq_var infoList = new DDS::SampleInfoSeq;
  DDS::ReturnCode_t dds_result;
  int amount;
  DDS::Boolean result = false;
  const char* print_prefix = "SpliceReadThread::svc";
  Knowledge::UpdateSeq_var update_data_list_ = new Knowledge::UpdateSeq;
 
  DDS::WaitSet waitset_;
  DDS::StatusCondition_ptr condition_;
  // Add update datareader statuscondition to waitset
  condition_ = update_reader_->get_statuscondition();
  condition_->set_enabled_statuses(DDS::DATA_AVAILABLE_STATUS);
  waitset_.attach_condition(condition_);
 
  ::DDS::Duration_t wait_time;
  wait_time.sec = 3;
  wait_time.nanosec = 0;
 
  // if we don't check originator for null, we get phantom sends
  // when the program exits.
  madara_logger_log(context_->get_logger(), logger::LOG_MAJOR,
      "%s: entering processing loop.\n", print_prefix);
 
  // by using conditionals, we can wait for a message for a specific time limit
  // the conditionList would tell us which particular conditions were met, but
  // since we've only set up the wait
  DDS::ConditionSeq_var conditionList = new DDS::ConditionSeq();
  result = waitset_.wait(conditionList.inout(), wait_time);
 
  madara_logger_log(context_->get_logger(), logger::LOG_DETAILED,
      "%s: entering a take on the DDS reader.\n", print_prefix);
 
  dds_result = update_reader_->take(update_data_list_, infoList, 20,
      DDS::ANY_SAMPLE_STATE, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE);
 
  amount = update_data_list_->length();
 
  if (amount != 0)
  {
    for (int i = 0; i < amount; ++i)
    {
      // if we are evaluating a message from ourselves, just continue
      // to the next one. It's also possible to receive null originators
      // from what I can only guess is the ospl daemon messing up
      if (!update_data_list_[i].originator.val())
      {
        // if we don't check originator for null, we get phantom sends
        // when the program exits.
        madara_logger_log(context_->get_logger(), logger::LOG_DETAILED,
            "%s: discarding null originator event.\n", print_prefix);
 
        continue;
      }
 
      if (update_data_list_[i].type != madara::transport::MULTIASSIGN)
      {
        madara_logger_log(context_->get_logger(), logger::LOG_DETAILED,
            "%s: discarding non-assignment event.\n", print_prefix);
        // we do not allow any other type than multiassign
 
        continue;
      }
 
      knowledge::KnowledgeMap rebroadcast_records;
      MessageHeader* header = 0;
 
      process_received_update((char*)update_data_list_[i].buffer.get_buffer(),
          update_data_list_[i].buffer.length(), id_, *context_, settings_,
          send_monitor_, receive_monitor_, rebroadcast_records,
          on_data_received_, print_prefix, "", header);
 
      if (header)
      {
        if (header->ttl > 0 && rebroadcast_records.size() > 0 &&
            settings_.get_participant_ttl() > 0)
        {
          --header->ttl;
          header->ttl = std::min(settings_.get_participant_ttl(), header->ttl);
 
          rebroadcast(print_prefix, header, rebroadcast_records);
        }
 
        // delete header
        delete header;
      }
    }
  }
 
  dds_result = update_reader_->return_loan(update_data_list_, infoList);
}