#include <boost/thread.hpp>
#include <boost/bind.hpp>

#include "FairMQLogger.h"
#include "FairMQMerger.h"

FairMQMerger::FairMQMerger()
{
}

FairMQMerger::~FairMQMerger()
{
}

void FairMQMerger::Run()
{
  FairMQLogger::GetInstance()->Log(FairMQLogger::INFO, ">>>>>>> Run <<<<<<<");

  boost::thread rateLogger(boost::bind(&FairMQDevice::LogSocketRates, this));

  struct nn_pollfd pfd [fNumInputs];

  for ( int i = 0; i < fNumInputs; i++ ) {
    pfd[i].fd = fPayloadInputs->at(i)->GetSocket();
    pfd[i].events = NN_POLLIN;
  }

  bool received = false;

  while ( fState == RUNNING ) {
    FairMQMessage msg;

    nn_poll(pfd, fNumInputs, 100);

    for ( int i = 0; i < fNumInputs; i++ ) {
      if (pfd[i].revents & NN_POLLIN) {
        received = fPayloadInputs->at(i)->Receive(&msg);
      }
      if (received) {
        fPayloadOutputs->at(0)->Send(&msg);
        received = false;
      }
    }
  }

  rateLogger.interrupt();
  rateLogger.join();
}