code: /*
   * (c) 2005-2017  Copyright, Real-Time Innovations, Inc. All rights reserved.
   * Subject to Eclipse Public License v1.0; see LICENSE.md for details.
   */
  
  package com.rti.perftest.harness;
  
  import java.util.ArrayList;
  import com.rti.perftest.IMessagingCB;
  import com.rti.perftest.IMessagingReader;
  import com.rti.perftest.IMessagingWriter;
  import com.rti.perftest.TestMessage;
  
  //===========================================================================
  
  /**
   * Listener for the Subscriber side
   *
   * Keeps stats on data received per second.
   * Returns a ping for latency packets
   */
  /*package*/ final class ThroughputListener implements IMessagingCB {
      // -----------------------------------------------------------------------
      // Public Fields
      // -----------------------------------------------------------------------
  
      public long packetsReceived = 0;
      public long bytesReceived = 0;
      public long missingPackets = 0;
      public boolean end_test = false;
      public int  lastDataLength = -1;
  
      // store info for the last data set
      public int  intervalDataLength = -1;
      public long intervalPacketsReceived = 0;
      public long intervalBytesReceived = 0;
      public long intervalMissingPackets = 0;
      public long intervalTime = 0;
      public double missingPacketsPercent = 0.0;
      public CpuMonitor CpuMonitor = new CpuMonitor();
      public boolean change_size = false;
  
      // -----------------------------------------------------------------------
      // Private Fields
      // -----------------------------------------------------------------------
  
      private IMessagingWriter _writer;
      private IMessagingReader _reader = null;
      private long _last_seq_num[];
      private long _beginTime;
      private int _numPublishers;
      private ArrayList<Integer> _finished_publishers;
      private boolean _useCft = false;
  
  
  
      // -----------------------------------------------------------------------
      // Public Methods
      // -----------------------------------------------------------------------
  
      // --- Constructors: -----------------------------------------------------
  
      public ThroughputListener(IMessagingWriter writer,
              boolean UseCft,
              int numPublishers) {
          _writer = writer;
          _last_seq_num = new long[numPublishers];
          _numPublishers = numPublishers;
          _finished_publishers = new ArrayList<Integer>();
          _useCft = UseCft;
      }
  
  
      public ThroughputListener(IMessagingWriter writer,
                                IMessagingReader reader,
                                boolean UseCft,
                                int numPublishers) {
          _writer = writer;
          _reader = reader;
          _last_seq_num = new long[numPublishers];
          _numPublishers = numPublishers;
          _finished_publishers = new ArrayList<Integer>();
          _useCft = UseCft;
      }
  
  
      // -----------------------------------------------------------------------
  
      /**
       * Used for receiving data using a thread instead of callback.
       */
      public void ReadThread() {
          TestMessage message = null;
          while (!end_test) {
              // Receive message should block until a message is received
              message = _reader.receiveMessage();
              if (message != null) {
                  processMessage(message);
              }
          }
      }
  
      // --- From IMessagingCB: ------------------------------------------------
  
      public void processMessage(TestMessage message) {
  
          if (message.entity_id >= _numPublishers ||
              message.entity_id < 0) {
              System.out.println("ProcessMessage: message content no valid. message.entity_id out of bounds");
              return;
          }
          // Check for test initialization messages
          if (message.size == PerfTest.INITIALIZE_SIZE) {
              _writer.send(message, false); // Don't check return; exiting anyway
              _writer.flush();
              return;
          }
          if (message.size == PerfTest.FINISHED_SIZE) {
  
              /*
               * PERFTEST-97
               * We check the entity_id of the publisher to see if it has already
               * send a FINISHED_SIZE message. If he has we ignore any new one.
               * Else, we add it to a vector. Once that vector contains all the
               * ids of the publishers the subscriber is suppose to know, that
               * means that all the publishers have finished sending data samples,
               * so it is time to finish the subscriber.
               */
              if (_finished_publishers.contains(message.entity_id)) {
                  return;
              }
  
              if (end_test) {
                  return;
              }
              _finished_publishers.add(message.entity_id);
              if (_finished_publishers.size() >= _numPublishers) {
                  print_summary_throughput(message, true);
                  end_test = true;
              }
              return;
          }
  
          // Send back a packet if this is a ping
          if ((message.latency_ping == PerfTest.subID)  ||
                  (_useCft && message.latency_ping != -1)) {
              boolean sent = _writer.send(message, false);
              _writer.flush();
              if (!sent) {
                  System.out.println("*** send() failure");
                  return;
              }
          }
  
          // reset internals
          if (message.size == PerfTest.LENGTH_CHANGED_SIZE) {
              print_summary_throughput(message);
              change_size = true;
              return;
          }
  
          // case where not running a scan
          if (message.size != lastDataLength) {
              packetsReceived = 0;
              bytesReceived = 0;
              missingPackets = 0;
  
              for (int i=0; i<_numPublishers; i++) {
                  _last_seq_num[i] = 0;
              }
  
              _beginTime = PerfTest.getTimeUsec();
              PerfTest._printer.set_data_length(message.size +
                      PerfTest.OVERHEAD_BYTES);
              PerfTest._printer.print_throughput_header();
          }
  
          lastDataLength = message.size;
          ++packetsReceived;
          bytesReceived += (message.size + PerfTest.OVERHEAD_BYTES);
          if (!_useCft) {
              // detect missing packets
              if (_last_seq_num[message.entity_id] == 0) {
                  _last_seq_num[message.entity_id] = message.seq_num;
              } else {
                  if (message.seq_num != ++_last_seq_num[message.entity_id]) {
                      // only track if skipped, might have restarted pub
                      if (message.seq_num > _last_seq_num[message.entity_id]) {
                          missingPackets +=
                                  message.seq_num - _last_seq_num[message.entity_id];
                      }
                      _last_seq_num[message.entity_id] = message.seq_num;
                  }
              }
          }
      }
  
      public void print_summary_throughput(TestMessage message) {
          print_summary_throughput(message, false);
      }
  
      public void print_summary_throughput(TestMessage message, boolean endTest) {
  
          // store the info for this interval
          long now = PerfTest.getTimeUsec();
  
          if (intervalDataLength != lastDataLength) {
  
              if (!_useCft) {
                  // detect missing packets
                  if (message.seq_num != _last_seq_num[message.entity_id]) {
                      // only track if skipped, might have restarted pub
                      if (message.seq_num > _last_seq_num[message.entity_id])
                      {
                          missingPackets +=
                              message.seq_num - _last_seq_num[message.entity_id];
                      }
                  }
              }
  
              intervalTime = now - _beginTime;
              intervalPacketsReceived = packetsReceived;
              intervalBytesReceived = bytesReceived;
              intervalMissingPackets = missingPackets;
              intervalDataLength = lastDataLength;
              missingPacketsPercent = 0.0;
  
              // Calculations of missing package percent
              if (intervalPacketsReceived + intervalMissingPackets != 0) {
                  missingPacketsPercent = (intervalMissingPackets * 100.0)
                          / (float) (intervalPacketsReceived
                          + intervalMissingPackets);
              }
  
              double outputCpu = 0;
              if (PerfTest._showCpu) {
                  outputCpu = CpuMonitor.get_cpu_average();
              }
              PerfTest._printer.print_throughput_summary(
                      intervalDataLength + PerfTest.OVERHEAD_BYTES,
                      intervalPacketsReceived,
                      intervalTime,
                      intervalBytesReceived,
                      intervalMissingPackets,
                      missingPacketsPercent,
                      outputCpu);
  
  
              System.out.flush();
          } else if (endTest) {
              System.out.printf(
                      "\nNo samples have been received by the Subscriber side,\n"
                      + "however 1 or more Publishers sent the finalization message.\n\n"
                      + "There are several reasons why this could happen:\n"
                      + "- If you are using large data, make sure to correctly adjust your\n"
                      + "  sendQueue, reliability protocol and flowController.\n"
                      + "- Make sure your -executionTime or -numIter in the Publisher side\n"
                      + "  are big enough.\n"
                      + "- Try sending at a slower rate -pubRate in the Publisher side.\n\n");
          }
  
          packetsReceived = 0;
          bytesReceived = 0;
          missingPackets = 0;
          // length changed only used in scan mode in which case
          // there is only 1 publisher with ID 0
          _last_seq_num[0] = 0;
          _beginTime = now;
      }
  
  }
  
  // ===========================================================================
  
  