SpliceDataReaderListener.cpp

Go to the documentation of this file.

#include "madara/transport/splice/SpliceDataReaderListener.h"
#include "madara/utility/LogMacros.h"
#include "madara/knowledge/UpdateTypes.h"
#include "madara/utility/Utility.h"

#include <iostream>
#include <sstream>

madara::transport::SpliceDataReaderListener::SpliceDataReaderListener(
  const std::string & id,
  madara::knowledge::ThreadSafeContext & context)
: id_ (id), context_ (context)
{
}

madara::transport::SpliceDataReaderListener::SpliceDataReaderListener(
  const SpliceDataReaderListener &ref)
: id_ (ref.id_), context_ (ref.context_)
{
}

madara::transport::SpliceDataReaderListener::~SpliceDataReaderListener()
{}

void madara::transport::SpliceDataReaderListener::handle_assignment (
  Knowledge::Update & data)
{
  if (data.key.val ())
  {
    // if we aren't evaluating a message from ourselves, process it
    std::string key = data.key.val ();
    long long value = data.value;
    int result = 0;

    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "SpliceDataReaderListener::handle_assignment:" \
      " waiting to process assignment\n");
      
    context_.lock ();
    unsigned long long cur_clock = context_.get_clock (key);
    unsigned long cur_quality = context_.get_quality (key);

    // if the data we are updating had a lower clock value or less quality
    // then that means this update is the latest value. Among
    // other things, this means our solution will work even
    // without FIFO channel transports

    // if the data we are updating had a lower clock value
    // then that means this update is the latest value. Among
    // other things, this means our solution will work even
    // without FIFO channel transports
    result = context_.set_if_unequal (key, value, 
                                      data.quality, data.clock, false);

    context_.unlock ();
    
    // if we actually updated the value
    if (result == 1)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "SpliceDataReaderListener::handle_assignment:" \
        " received data[%s]=%q from %s\n",
        key.c_str (), value, data.originator.val ());
    }
    // if the data was already current
    else if (result == 0)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "SpliceDataReaderListener::handle_assignment:" \
        " discarded data[%s]=%q from %s as the value was already set\n",
        key.c_str (), value, data.originator.val ());
    }
    else if (result == -1)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "SpliceDataReaderListener::handle_assignment:" \
        " discarded data due to null key\n",
        key.c_str (), value, data.originator.val ());
    }
    else if (result == -2)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "SpliceDataReaderListener::handle_assignment:" \
        " discarded data[%s]=%q due to lower quality (%u vs %u)\n",
        key.c_str (), value, cur_quality, data.quality);
    }
    else if (result == -3)
    {
      madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
        "SpliceDataReaderListener::handle_assignment:" \
        " discarded data[%s]=%q due to older timestamp (%Q vs %Q)\n",
        key.c_str (), value, cur_clock, data.clock);
    }
  }
}

void madara::transport::SpliceDataReaderListener::handle_multiassignment (
  Knowledge::Update & data)
{
  if (data.key.val ())
  {
    std::string key;
    char symbol;
    long long value;
    std::stringstream stream (data.key.val ());

    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "SpliceDataReaderListener::handle_multiassignment:" \
      " waiting to process multiassignment\n");

    context_.lock ();
    
    madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
      "SpliceDataReaderListener::handle_multiassignment:" \
      " processing multiassignment (%s)\n",
      data.key.val ());

    while (!stream.eof ())
    {
      stream >> key >> symbol >> value >> symbol;

      int result = 0;
      unsigned long long cur_clock = context_.get_clock (key);
      unsigned long cur_quality = context_.get_quality (key);

      // if the data we are updating had a lower clock value
      // then that means this update is the latest value. Among
      // other things, this means our solution will work even
      // without FIFO channel transports
      result = context_.set_if_unequal (key, value, 
                                        data.quality, data.clock, false);

      // if we actually updated the value
      if (result == 1)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "SpliceDataReaderListener::handle_multiassignment:" \
          " received data[%s]=%q from %s\n",
          key.c_str (), value, data.originator.val ());
      }
      // if the data was already current
      else if (result == 0)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "SpliceDataReaderListener::handle_multiassignment:" \
          " discarded data[%s]=%q from %s as the value was already set\n",
          key.c_str (), value, data.originator.val ());
      }
      else if (result == -1)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "SpliceDataReaderListener::handle_multiassignment:" \
          " discarded data due to null key\n",
          key.c_str (), value, data.originator.val ());
      }
      else if (result == -2)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "SpliceDataReaderListener::handle_multiassignment:" \
          " discarded data[%s]=%q due to lower quality (%u vs %u)\n",
          key.c_str (), value, cur_quality, data.quality);
      }
      else if (result == -3)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MAJOR,
          "SpliceDataReaderListener::handle_multiassignment:" \
          " discarded data[%s]=%q due to older timestamp (%Q vs %Q)\n",
          key.c_str (), value, cur_clock, data.clock);
      }
    }
    
    context_.unlock ();
  }
}

void
madara::transport::SpliceDataReaderListener::on_sample_lost (
  DDS::DataReader_ptr, const DDS::SampleLostStatus &status)
{
}

void
madara::transport::SpliceDataReaderListener::on_sample_rejected (
  DDS::DataReader_ptr, const DDS::SampleRejectedStatus &status)
{
}

void
madara::transport::SpliceDataReaderListener::on_requested_incompatible_qos (
  DDS::DataReader_ptr, const DDS::RequestedIncompatibleQosStatus &status)
{
}

void
madara::transport::SpliceDataReaderListener::on_requested_deadline_missed (
  DDS::DataReader_ptr reader, const DDS::RequestedDeadlineMissedStatus & status)
{
}

void
madara::transport::SpliceDataReaderListener::on_liveliness_changed (
  DDS::DataReader_ptr reader, const DDS::LivelinessChangedStatus & status)
{
}

void
madara::transport::SpliceDataReaderListener::on_subscription_matched (
  DDS::DataReader_ptr reader, const DDS::SubscriptionMatchedStatus & status)
{
  context_.set_changed ();
}

void 
madara::transport::SpliceDataReaderListener::on_data_available(
  DDS::DataReader_ptr reader)
{
  DDS::SampleInfoSeq_var infoList = new DDS::SampleInfoSeq;
  DDS::ReturnCode_t      dds_result;
  int                    amount;
  DDS::Boolean           result = false;
  Knowledge::UpdateSeq_var update_data_list_ = new Knowledge::UpdateSeq;

  Knowledge::UpdateDataReader_ptr update_reader = 
    dynamic_cast<Knowledge::UpdateDataReader_ptr> (reader);

  if (update_reader == 0)
  {
    madara_logger_log (context_->get_logger (), logger::LOG_ERROR,
      "SpliceDataReaderListener::on_data_available:" \
      " Unable to create specialized reader. Leaving callback...\n");

    return;
  }

  dds_result = update_reader->take (update_data_list_, infoList, 1, 
    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 () || 
        id_ == update_data_list_[i].originator.val ())
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
          "SpliceDataReaderListener::on_data_available:" \
          " discarding null originator event\n");

        continue;
      }

      if (madara::knowledge::ASSIGNMENT == update_data_list_[i].type)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
          "SpliceDataReaderListener::on_data_available:" \
          " processing %s=%q from %s with time %Q and quality %u\n", 
          update_data_list_[i].key.val (), update_data_list_[i].value, 
          update_data_list_[i].originator.val (),
          update_data_list_[i].clock, update_data_list_[i].quality);

        handle_assignment (update_data_list_[i]);
      }
      else if (madara::knowledge::MULTIPLE_ASSIGNMENT == update_data_list_[i].type)
      {
        madara_logger_log (context_->get_logger (), logger::LOG_MINOR,
          "SpliceDataReaderListener::on_data_available:" \
          " processing multassignment from %s with time %Q and quality %u\n",
          update_data_list_[i].originator.val (),
          update_data_list_[i].clock, update_data_list_[i].quality);

        handle_multiassignment (update_data_list_[i]);
      }

      // otherwise the key was null, which is unusable
    }
  }
  dds_result = update_reader->return_loan (update_data_list_, infoList);

}