JLigierMirror.cc

Go to the documentation of this file.

#include <chrono>
#include <iostream>
#include <set>
#include <string>

#include "JNet/JControlHost.hh"
#include "JIO/JByteArrayIO.hh"
#include "JDAQ/JDAQTags.hh"
#include "JDAQ/JDAQPreambleIO.hh"
#include "JDAQ/JDAQEventHeaderIO.hh"
#include "JSystem/JDateAndTime.hh"

#include "Jeep/JParser.hh"
#include "Jeep/JMessage.hh"

/**
 * \file
 *
 * A tool to forward messages with the given tags from one ControlHost server (e.g. JLigier) to another.
 *
 * The options <tt>-H <source>[:port]</tt> and <tt>-X <target>[:port]</tt>
 * correspond to the host name and the port of the source and target server, respectively.\n
 * The options <tt>-t</tt> and <tt>-T</tt> correspond to the ControlHost tag(s)
 * with the mode subscription "any" and subscription "all", respectively.
 * \author tgal, mdejong
 */
int main(int argc, const char *argv[])
{
  using namespace std;
  using namespace JPP;
  using namespace KM3NETDAQ;

  string          source;
  string          target;
  int             report_interval;  // in seconds
  set<JNET::JTag> tagList;
  set<JNET::JTag> TagList;
  JDAQTriggerMask trigger_mask;
  int             debug;

  try {

    JParser<> zap("Program to forward messages from one ControlHost server to another.");

    zap['H'] = make_field(source)          = "localhost";
    zap['X'] = make_field(target)          = "localhost";
    zap['t'] = make_field(tagList)         = JPARSER::initialised();
    zap['T'] = make_field(TagList)         = JPARSER::initialised();
    zap['@'] = make_field(trigger_mask)    = TRIGGER_MASK_ON;
    zap['i'] = make_field(report_interval) = 30;
    zap['d'] = make_field(debug)           = 2;

    zap(argc, argv);
  }
  catch(const exception &error) {
    FATAL(error.what() << endl);
  }


  if (tagList.empty() && TagList.empty()) {
    FATAL("No tags specified.");
  }

  NOTICE("Forwarding messages from " << endl << "    " << source << " -> " << target << endl);


  JControlHost::Throw(true);

  try {

    JControlHost in (source);
    JControlHost out(target);

    {
      NOTICE("with the following tags: ");

      JSubscriptionList buffer;

      for (set<JTag>::const_iterator i = tagList.begin(); i != tagList.end(); ++i) {
        buffer.add(JSubscriptionAny(*i));
        NOTICE(*i << "(any) ");
      }

      for (set<JTag>::const_iterator i = TagList.begin(); i != TagList.end(); ++i) {
        buffer.add(JSubscriptionAll(*i));
        NOTICE(*i << "(all) ");
      }

      NOTICE(endl);

      in.Subscribe(buffer);
      in.SendMeAlways();
    }


    JPrefix      prefix;
    vector<char> buffer;
    unsigned int message_count[] = { 0, 0 };
    float        milliseconds_passed;
    std::chrono::steady_clock::time_point start_time = std::chrono::steady_clock::now();

    for (const string stop("stop"); buffer.size() != stop.size() || string(buffer.data(), stop.size()) != stop; ) {

      in.WaitHead(prefix);

      buffer.resize(prefix.getSize());

      in.GetFullData(buffer.data(), buffer.size());

      message_count[0] += 1;

      DEBUG(getDateAndTime() << ' ' << left << setw(8) << prefix.getTag() << ' ' << right << setw(8) << prefix.getSize() << endl);

      bool dos = false;

      if (prefix.getTag() == IO_EVENT) {

        JDAQPreamble     preamble;
        Version_t        version;
        JDAQEventHeader  header;

        JByteArrayReader reader(buffer.data(), buffer.size());

        reader >> preamble;
        reader >> version;
        reader >> header;

        dos = header.hasTriggerMask(trigger_mask);

      } else {

        dos = true;
      }

      if (dos) {

        message_count[1] += 1;

        out.PutFullData(prefix.getTag(), buffer.data(), buffer.size());
      }

      milliseconds_passed = chrono::duration_cast<chrono::milliseconds>(std::chrono::steady_clock::now() - start_time).count();

      if (milliseconds_passed > report_interval * 1e3) {

        STATUS(getDateAndTime() << " : " << "Message rate: " 
               << message_count[0] / milliseconds_passed * 1e3 << " "
               << message_count[1] / milliseconds_passed * 1e3 << " Hz" << endl);

        start_time = std::chrono::steady_clock::now();

        message_count[0] = 0;
        message_count[1] = 0;
      }
    }
  }
  catch(const JControlHostException& error) {
    ERROR(error << endl);
  }
}