source: trunk/ETSI-Testsuites/ETSI_auto_IOT/adapter/src/LowerTestAdapter/LowerTestAdapter.cpp @ 53

#include "LowerTestAdapter.h"

#include <iostream>
#include <sys/time.h> 

#include "Helper/Singleton.h"
#include "Logger/Logger.h"
#include "Messages/CommonTrafficCaptureMessages.h"
#include "Messages/TrafficCaptureMessageFactory.h"
#include "Messages/CapturedData.h"
#include "Dispatcher/EthernetDispatcher.h"
#include "Ports/MessagesIdentifiers.h"

LowerTestAdapter::LowerTestAdapter() 
  :m_dataPort(0),
   m_adapterConfigPort(0) {
  
  Logger::Instance().LogInfo("LowerTester creation");
}

LowerTestAdapter::~LowerTestAdapter() {
}

void LowerTestAdapter::RegisterDataPort(DataPort *pPort) {
  m_dataPort = pPort;
}

void LowerTestAdapter::RegisterAdapterConfigPort(AdapterConfigPort *pPort) {
  m_adapterConfigPort = pPort;
}

void LowerTestAdapter::RegisterFilter(const ComponentId *pComponent, const FilterSet & filter) {

    m_filter.RegisterComponent(pComponent, filter);
}

void LowerTestAdapter::InitializeCapture(boost::shared_ptr<GeneralConfigurationParams> configParams)  throw (LowerTestAdapterException) {

  clog << ">>> LowerTestAdapter::InitializeCapture" << endl;
  
  // Process OpenDevice message
  OpenDeviceRequest rq;
  // Set capture mode.
  ECaptureMode ecMode;
  clog << "LowerTestAdapter::InitializeCapture: Prepare mode:" << static_cast<int>(configParams.get()->Mode()) << endl;
  switch (configParams.get()->Mode()) {
    case 0x00:
      ecMode = ECaptureMode_Live;
      break;
    case 0x01:
      ecMode = ECaptureMode_Offline;
      break;
    default:
      clog << "LowerTestAdapter::InitializeCapture: set live" << endl;
      ecMode = ECaptureMode_Live;
  }
  clog << "LowerTestAdapter::InitializeCapture: Set mode:" << ecMode << endl;
  rq.SetCaptureMode(ecMode);
  // Set capture type.
  rq.SetCaptureType(ECaptureType_PCAP);
  
  // Process merge operation first.
  if ((ecMode == ECaptureMode_Offline) && (configParams.get()->MergerFileList().size() != 0))
  {
    clog << "LowerTestAdapter::InitializeCapture: process merge operation first" << endl;
    MergePcapFilesRequest mergeReq;
    // Populate Merge message.
    mergeReq.SetMergeSource(configParams.get()->MergerFilePath(), configParams.get()->MergerFileList());
    // Populate Merge tool path.
    mergeReq.SetMergecapDirectory(configParams.get()->PcapMergeTool());
    // Send parameters to the TrafficCapture process.
    std::clog << "LowerTestAdapter::InitializeCapture: Merge tool path" << std::endl;
    m_captureSocket->Send(mergeReq.GetEncodedMessage(), mergeReq.GetEncodedDataLength());
    // TODO: Process the response !!!
  }
   // Continue with OpenDevice reauest processing
  if (ecMode == ECaptureMode_Live)
  {
    std::vector<std::string> ifaces = configParams.get()->Ifaces();
    // Sanity check.
    if (ifaces.size() == 0)
    {
      std::cerr << "LowerTestAdapter::InitializeCapture: Ifaces are not well configured" << std::endl;
      throw LowerTestAdapterException("LowerTestAdapter::InitializeCapture: Ifaces are not well configured");
    }
    // Set interfaces.
    for(vector<string>::iterator it = ifaces.begin() ; it != ifaces.end(); ++it)
    {
      std::clog << "LowerTestAdapter::InitializeCapture: set iface: " << *it << std::endl;
      rq.SetParameters(*it);
    }
  }
  else // Off line mode
  {
    // Check file to play
    if ((configParams.get()->MergerFileList().size() == 0) && (configParams.get()->FileToPlay().length() != 0))
    {
      rq.SetParameters(configParams.get()->FileToPlay());
    }
    else
    {
      // TODO: To be continued.
      std::cerr << "LowerTestAdapter::InitializeCapture: Offline mode is not properly configure" << std::endl;
    }
    // Set timestamp offset if not 0
    if ((configParams.get()->PcapTimeStampSeconds() != 0) || (configParams.get()->PcapTimeStampMicroseconds() != 0))
    {
      struct timeval timestamp = { 
        (time_t)(configParams.get()->PcapTimeStampSeconds()), 
        configParams.get()->PcapTimeStampMicroseconds() };
      std::clog << "LowerTestAdapter::InitializeCapture: timestamp={" << timestamp.tv_sec << ", " << timestamp.tv_usec << "}" << std::endl;
      rq.SetTimestamp(timestamp);
    }
  }
  // Send parameters to the TrafficCapture process.
  std::clog << "LowerTestAdapter::InitializeCapture: Open device" << std::endl;
  SendMessage(&rq);
}

void LowerTestAdapter::SendMessage(TrafficCaptureMessage * pMsg)
{
  m_captureSocket->Send(pMsg->GetEncodedMessage(), pMsg->GetEncodedDataLength());
}

void LowerTestAdapter::StartCapture() {
  std::stringstream ss;

  // Set capture filter
  SetFilterRequest filterReq;
  filterReq.SetFilter(m_filter.GetPcapFilter());
  ss << "Capture Filter: " << m_filter.GetPcapFilter();
  Logger::Instance().LogDebug(ss.str());
  SendMessage(&filterReq);

  // Start capture
  StartCaptureRequest rq;
  std::clog << "LowerTestAdapter::StartCapture" << std::endl;
  SendMessage(&rq);
}

void LowerTestAdapter::StopCapture() {

  StopCaptureRequest rq;
  std::clog << "LowerTestAdapter::StopCapture" << std::endl;
  SendMessage(&rq);
  Logger::Instance().LogDebug("Filter clean-up");
  m_filter.Clear();
}

void LowerTestAdapter::ConnectToCaptureServer(boost::shared_ptr<GeneralConfigurationParams> configParams)  throw (LowerTestAdapterException) {

  clog << ">>> LowerTestAdapter::ConnectToCaptureServer: " << configParams.get()->CaptureModuleAddress() << " - " << configParams.get()->CaptureModulePort() << endl;
  
  // Sanity checks.
  if (configParams.get()->CaptureModuleAddress().length() == 0)
  {
    cerr << ">>> LowerTestAdapter::ConnectToCaptureServer: " << configParams.get()->CaptureModuleAddress() << " - " << configParams.get()->CaptureModulePort() << endl;
    throw LowerTestAdapterException("LowerTestAdapter::ConnectToCaptureServer: information is missing");
  }
  
  m_captureSocket = new SocketClient(configParams.get()->CaptureModuleAddress(), configParams.get()->CaptureModulePort());
  m_captureThread = new boost::thread(&Run);

  InitializeCapture(configParams);
}

void LowerTestAdapter::EnqueueMessage(const ComponentId * pComp, const unsigned char * pData, int nDataLen) {
  t3devlib::Bitstring bsCapturedBits;
  bsCapturedBits.Append(pData, nDataLen * 8);
  // Enqueue captured packet to monitor ports
  m_dataPort->EnqueueMsg(m_dataPort->GetConnectedPort(pComp), bsCapturedBits);
}

void LowerTestAdapter::ProcessCapturedData(const unsigned char * pCapturedData, unsigned int nDataLen) {
  DispatchInfo msgInfo;
  msgInfo.SetData(pCapturedData, nDataLen);
  ComponentFilter filterCopy (m_filter);
  bool bEnqueued = EthernetDispatcher::Instance().Dispatch(msgInfo, filterCopy);

  if(!bEnqueued) {
    std::stringstream ss;
    ss << "Captured packet not selected by any Monitor port";
    Logger::Instance().LogWarning(ss.str());
  }
}

void LowerTestAdapter::ProcessResponse(const std::string& message, bool result) {
  Logger::Instance().LogInfo("LowerTestAdapter::ProcessResponse");
  
  m_adapterConfigPort->ProcessResponse (message, result);
}

void LowerTestAdapter::Run() {
  Logger::Instance().LogInfo("LowerTester dispatch thread created");

  //Main loop
  while(true) {

    // Receive message header
    TMessageHeader msgHeader;
    int nb = Singleton<LowerTestAdapter>::Instance()
      .m_captureSocket->Receive(sizeof(TMessageHeader), 
        reinterpret_cast<char*>(&msgHeader));

    // Receive payload
    char *psPayload = new char[msgHeader.nMsgLen];
    nb = LowerTestAdapter::Instance()
      .m_captureSocket->Receive(msgHeader.nMsgLen, psPayload);

    // Analyze message
    TrafficCaptureMessage *pMsg 
      = TrafficCaptureMessageFactory::Instance().CreateMessage(msgHeader.nMsgType, psPayload, msgHeader.nMsgLen);
    if (pMsg == NULL)
    {
      Logger::Instance().LogError("LowerTestAdapter::Run: Create TrafficCaptureMessage failed");
      continue;
    }

    clog << "LowerTestAdapter::Run: receive response=" << pMsg->GetId() << endl;
    string message("");
    switch (pMsg->GetId()) {
    case CAPTURED_PACKET_IND:
      {
        CapturedData * pCapturedData = dynamic_cast<CapturedData*>(pMsg);
        const unsigned char * pData = reinterpret_cast<const unsigned char *>(pCapturedData->GetData());    
        LowerTestAdapter::Instance().ProcessCapturedData(pData, pCapturedData->GetDataLength());
      }
      break;
    case START_CAPTURE_REP:
      message.assign(MessagesIdentifiers::StartTrafficCaptureRsp);
      break;
    case STOP_CAPTURE_REP:
      message.assign(MessagesIdentifiers::StopTrafficCaptureRsp);
      break;
    case OPEN_DEVICE_REP:
      message.assign(MessagesIdentifiers::GeneralConfigurationRsp);
      break;
    case SET_FILTER_REP:
      //message.assign(MessagesIdentifiers::SetFilterRsp);
      break;
    case MERGE_FILES_REP:
      break;
    default:
      break;
    }
    
    if (message.length() != 0)
    {
      clog << "LowerTestAdapter::Run: Call LowerTestAdapter::Instance().ProcessResponse: " << message << " - " << ((CommonReplyMessage *)pMsg)->GetResult() << endl;
      LowerTestAdapter::Instance().ProcessResponse(message, ((CommonReplyMessage *)pMsg)->GetResult());
    }
    TrafficCaptureMessageFactory::Instance().DisposeMessage(pMsg);
  }
}