package brn.swans.app.rtcp;

import java.util.List;
import java.util.Vector;

import jist.runtime.JistAPI;
import jist.runtime.Main;
import jist.runtime.JistAPI.Continuation;
import jist.swans.Constants;
import jist.swans.misc.Message;
import jist.swans.misc.MessageAnno;
import jist.swans.misc.Util;
import jist.swans.net.NetAddress;
import jist.swans.trans.TransInterface;
import jist.swans.trans.TransInterface.TransUdpInterface;

import org.apache.log4j.Logger;

import brn.swans.app.rtcp.msg.RtcpByeMessage;
import brn.swans.app.rtcp.msg.RtcpCompoundMessage;
import brn.swans.app.rtcp.msg.RtcpMessage;
import brn.swans.app.rtcp.msg.RtcpReceiverReportMessage;
import brn.swans.app.rtcp.msg.RtcpSDESMessage;
import brn.swans.app.rtcp.msg.RtcpSenderReportMessage;
import brn.swans.app.rtp.RtpEndsystemModule;
import brn.swans.app.rtp.msg.RtpMessage;

/**
 * The class RTCPEndsystemModule is responsible for creating, receiving and
 * processing of rtcp packets. It also keeps track of this and other rtp end
 * systems.
 */
public class RtcpEndsystemModule {

  /**
   * Logger.
   */
  public static final Logger log = Logger.getLogger(RtcpEndsystemModule.class.getName());

  protected static interface Scheduler extends JistAPI.Proxiable {
    /**
     * Schedule a RTCP packet.
     */
    void schedule();
  }

  protected class SchedulerImpl implements Scheduler {
    public void schedule(){
      schedulePacket();
    }
  }

  // ////////////////////////////////////////////////
  // constants
  //

  // ////////////////////////////////////////////////
  // locals
  //

  /**
   * The CNAME of this end system.
   */
   private String commonName;

   /**
   * The maximum size an RTCPCompundPacket can have.
   */
  protected int mtu;

  /**
   * The bandwidth for this rtp session.
   */
  protected int bandwidth;

  /**
   * The percentage of bandwidth for rtcp.
   */
  protected int rtcpPercentage;

  /**
   * True when this end system has chosen its ssrc identifier.
   */
  protected boolean ssrcChosen;

  /**
   * True when this end system is about to leave the session.
   */
  protected boolean leaveSession;

  /**
   * The RTPSenderInfo about this end system.
   */
  protected RtpSenderInfo senderInfo;

  /**
   * Information about all known rtp end system participating in this rtp
   * session.
   */
  protected List participantInfos;

  /**
   * The server socket for receiving rtcp packets.
   */
//  protected NetAddress socketAddrIn;
  protected int socketPortIn;

  /**
   * The client socket for sending rtcp packets.
   */
  protected NetAddress socketAddrOut;

  protected int socketPortOut;

  /**
   * The number of packets this rtcp module has calculated.
   */
  protected int packetsCalculated;

  /**
   * The average size of an RTCPCompoundPacket.
   */
  protected double averagePacketSize;

  /**
   * The output vector for statistical data about the behaviour of rtcp. Every
   * participant's rtcp module writes its calculated rtcp interval (without
   * variation by multiplicating a random number in 0.5 .. 1).
   */
  protected List rtcpIntervalOutVector;

  /**
   * The rtp module.
   */
  protected RtpEndsystemModule rtpModule;

  /**
   * Self-referencing reference.
   */
  protected Scheduler self;

  /**
   * Reference to the udp entity, as reported from rtp.
   */
  protected TransInterface.TransUdpInterface udpEntity;

  // ////////////////////////////////////////////////
  // initialization
  //

  public RtcpEndsystemModule() {

    // initialize variables
    ssrcChosen = false;
    leaveSession = false;

    packetsCalculated = 0;

    averagePacketSize = 0.0;

    participantInfos = new Vector();

    rtcpIntervalOutVector = new Vector();

    self = (Scheduler) JistAPI.proxy(new SchedulerImpl(), Scheduler.class);
  }

  // ////////////////////////////////////////////////
  // entity hookup
  //

  /**
   * Hook up with the rtp module.
   *
   * @param module
   *          rtp module
   */
  public void setRtpModuleEntity(RtpEndsystemModule module) {
    this.rtpModule = module;
  }

  /**
   * Hook up with the transport entity.
   *
   * @param udpEntity
   *          udp entity
   */
  public void setUdpEntity(TransUdpInterface udpEntity) {
    if (!JistAPI.isEntity(udpEntity))
      throw new IllegalArgumentException("expected entity");

    this.udpEntity = udpEntity;
  }

  // ////////////////////////////////////////////////
  // accessors
  //

  public int getSsrc() {
    return senderInfo.getSsrc();
  }

  // ////////////////////////////////////////////////
  // overwrites
  //

  public String toString() {
    return //"RtcpEndsystemModule:" +
      commonName + " " + Util.timeSeconds();
  }


  // ////////////////////////////////////////////////
  // Called by the rtp module
  //

  /**
   * Initializes the rtcp module when the session is started.
   */
  public void initializeRTCP(String commonName, int mtu, int bandwidth,
      int rtcpPercentage, NetAddress destAddr, int rtcpPort) {
    this.mtu = mtu;
    this.bandwidth = bandwidth;
    this.commonName = commonName;
    this.rtcpPercentage = rtcpPercentage;
    this.socketAddrOut = destAddr;
    this.socketPortOut = rtcpPort;
    this.socketPortIn  = rtcpPort;

    this.senderInfo = new RtpSenderInfo();

    SDESItem sdesItem = new SDESItem(SDESItem.SDES_CNAME, commonName);
    senderInfo.addSDESItem(sdesItem);

    if (log.isDebugEnabled())
      log.debug(this + " initializing rtcp");

    // create server socket for receiving rtcp packets
    createServerSocket();
    createClientSocket();
  }

  /**
   * Stores information about the new transmission.
   *
   * @param clockRate
   * @param timeStampBase
   * @param sequenceNumberBase
   */
  public void senderModuleInitialized(int clockRate, int timeStampBase,
      short sequenceNumberBase) {

    if (log.isDebugEnabled())
      log.debug(this + " sender initialized, clock=" + clockRate + ", timestamp=" +
          timeStampBase + ", seqNo=" + sequenceNumberBase);

    senderInfo.setStartTime(JistAPI.getTime());
    senderInfo.setClockRate(clockRate);
    senderInfo.setTimeStampBase(timeStampBase);
    senderInfo.setSequenceNumberBase(sequenceNumberBase);
  }

  /**
   * Stores information about an outgoing rtp data packet.
   */
  public void dataOut(Message msg, int srcPort) {
    RtpMessage rtpPacket = (RtpMessage) msg;
    processOutgoingRTPPacket(rtpPacket);
  }

  /**
   * Stores information about an outgoing rtp data packet.
   */
  public void dataIn(Message msg, NetAddress src, int srcPort) {

    if (log.isDebugEnabled())
      log.debug(this + " received rtcp packet " + msg);

    RtpMessage rtpPacket = (RtpMessage) msg;
    processIncomingRTPPacket(rtpPacket, src, srcPort);
  }

  /**
   * Makes the rtcp module send an RtcpByeMessage in the next RTCPCompoundPacket
   * to tell other participants in the rtp session that this end system leaves.
   */
  public void leaveSession() {

    if (log.isDebugEnabled())
      log.debug(this + " leaving session ssrc=" + getSsrc());

    leaveSession = true;
  }

  // ////////////////////////////////////////////////
  // Called by the udp socket
  //

  protected class SocketHandler implements TransInterface.SocketHandler {

    /**
     * Called when this rtcp module receives data from the socket layer.
     */
    public void receive(Message msg, MessageAnno anno, NetAddress src, int srcPort)
        throws Continuation {
      RtcpCompoundMessage packet = RtcpCompoundMessage.fromBytes(msg);

      packet.setArrivalTime(JistAPI.getTime());
      processIncomingRTCPPacket(packet, src, srcPort);
    }

  }

  /**
   * Request a server socket from the socket layer.
   */
  protected void createServerSocket() {
    if (log.isDebugEnabled())
      log.debug(this + " listening on port " +
          this.socketPortIn);

    // Create socket listener
    udpEntity.addSocketHandler(this.socketPortIn, new SocketHandler());
  }

  /**
   * Requests a client socket from the socket layer.
   */
  protected void createClientSocket() {
    // schedule first rtcp packet
    long intervalLength = Constants.MICRO_SECOND * 2500
        * (Constants.random.nextLong() % 500 + 1000);

    if (log.isDebugEnabled())
      log.debug(this + " scheduling first rtcp packet for " + intervalLength);

    JistAPI.sleepBlock(intervalLength);

    // it's time to create an rtcp packet
    if (!ssrcChosen) {
      chooseSSRC();

      if (log.isDebugEnabled())
        log.debug(this + " rtcp initialized" );
    }

    self.schedule();
  }

  // ////////////////////////////////////////////////
  // Self-calls
  //

  public void schedulePacket() {
    if (Main.ASSERT)
      Util.assertion(ssrcChosen);

    createPacket();

    if (!leaveSession) {
      scheduleInterval();
    }
  }

  // ////////////////////////////////////////////////
  // internals
  //

  /**
   * Chooses the ssrc identifier for this end system.
   */
  protected void chooseSSRC() {

    int ssrc = 0;
    boolean ssrcConflict = false;
    do {
      ssrc = Constants.random.nextInt();
      ssrcConflict = (null != findParticipantInfo(ssrc));
    } while (ssrcConflict);

    if (log.isDebugEnabled())
      log.debug(this + " choosen ssrc=" + ssrc);

    senderInfo.setSsrc(ssrc);
    participantInfos.add(senderInfo);
    ssrcChosen = true;
  }

  /**
   * Calculates the length of the next rtcp interval an issues a self message to
   * remind itself.
   */
  protected void scheduleInterval() {

    double senderFactor = (senderInfo.isSender() ? 1.0 : 0.75) / 100.0;
    long intervalLength = Constants.SECOND
        * (long) (averagePacketSize * (double) (participantInfos.size())
            / bandwidth * rtcpPercentage * senderFactor);

    // write the calculated interval into file
    rtcpIntervalOutVector.add(new Long(intervalLength));

    // interval length must be at least 5 seconds
    if (intervalLength < 5 * Constants.SECOND)
      intervalLength = 5 * Constants.SECOND;

    // to avoid rtcp packet bursts multiply calculated interval length
    // with a random number between 0.5 and 1.5
    intervalLength = (long) (intervalLength * (0.5 + Constants.random
        .nextDouble()));

    JistAPI.sleep(intervalLength);
    self.schedule();
  }

  /**
   * Creates and sends an RTCPCompoundPacket.
   */
  protected void createPacket() {
    if (log.isDebugEnabled())
      log.debug(this + " creating packet" );

    // first packet in an rtcp compound packet must
    // be a sender or receiver report
    RtcpReceiverReportMessage reportPacket;

    // if this rtcp end system is a sender (see SenderInformation::isSender()
    // for
    // details) insert a sender report
    if (senderInfo.isSender()) {
      RtcpSenderReportMessage senderReportPacket = new RtcpSenderReportMessage();
      senderReportPacket.setSenderReport(senderInfo.senderReport(JistAPI
          .getTime()));
      reportPacket = senderReportPacket;
    } else {
      reportPacket = new RtcpReceiverReportMessage();
    }
    reportPacket.setSsrc(senderInfo.getSsrc());

    // insert receiver reports for packets from other sources
    for (int i = 0; i < participantInfos.size(); i++)
    {
      RtpParticipantInfo participantInfo = (RtpParticipantInfo) participantInfos
          .get(i);
      if (participantInfo.getSsrc() != senderInfo.getSsrc()) {
        ReceptionReport report = ((RtpReceiverInfo) participantInfo)
            .receptionReport(JistAPI.getTime());
        if (null != report) {
          reportPacket.addReceptionReport(report);

        }
      }
      participantInfo.nextInterval(JistAPI.getTime());

      if (participantInfo.toBeDeleted(JistAPI.getTime())) {

        participantInfos.remove(participantInfo);
        participantInfo = null;
        // perhaps inform the profile
      }
    }

    // insert source description items (at least common name)
    RtcpSDESMessage sdesPacket = new RtcpSDESMessage();

    SDESChunk chunk = senderInfo.getSdesChunk();
    sdesPacket.addSdesChunk(chunk);

    RtcpCompoundMessage compoundPacket = new RtcpCompoundMessage();

    compoundPacket.addRTCPPacket(reportPacket);

    compoundPacket.addRTCPPacket(sdesPacket);

    // create rtcp app/bye packets if needed
    if (leaveSession) {
      RtcpByeMessage byePacket = new RtcpByeMessage();
      byePacket.setSsrc(senderInfo.getSsrc());
      compoundPacket.addRTCPPacket(byePacket);
    }

    calculateAveragePacketSize(compoundPacket.getLength());

    if (log.isDebugEnabled())
      log.debug(this + " sending packet " + compoundPacket );

    udpEntity.send(compoundPacket, null, socketAddrOut, socketPortOut,
        socketPortIn, Constants.NET_PRIORITY_D_VOICE);

    if (leaveSession) {
      rtpModule.sessionLeft();
    }

  }

  /**
   * Extracts information of a sent RTPPacket.
   */
  protected void processOutgoingRTPPacket(RtpMessage packet) {
    senderInfo.processRTPPacket(packet, JistAPI.getTime());
  }

  /**
   * Extracts information of a received RTPPacket.
   */
  protected void processIncomingRTPPacket(RtpMessage packet,
      NetAddress address, int port) {

    int ssrc = packet.getSsrc();
    RtpParticipantInfo participantInfo = findParticipantInfo(ssrc);
    if (null == participantInfo) {
      participantInfo = new RtpReceiverInfo(ssrc);
      participantInfo.setAddress(address);
      participantInfo.setRtpPort(port);
      participantInfos.add(participantInfo);
    } else {
      // check for ssrc conflict
      if (participantInfo.getAddress() != address) {
        // we have an address conflict
      }
      if (participantInfo.getRtpPort() == 0) {
        participantInfo.setRtpPort(port);
      } else if (participantInfo.getRtpPort() != port) {
        // we have an rtp port conflict
        throw new RuntimeException("rtp port conflict");
      }
    }
    participantInfo.processRTPPacket(packet, packet.getArrivalTime());
  }

  /**
   * Extracts information of a received RTCPCompoundPacket.
   */
  protected void processIncomingRTCPPacket(RtcpCompoundMessage packet,
      NetAddress address, int port) {
    if (Main.ASSERT)
      Util.assertion(null != address);

    calculateAveragePacketSize(packet.getLength());

    List rtcpPackets = packet.getRtcpPackets();

    long arrivalTime = packet.getArrivalTime();
    packet = null;

    for (int i = 0; i < rtcpPackets.size(); i++) {
      // remove the rtcp packet from the rtcp compound packet
      RtcpMessage rtcpPacket = (RtcpMessage) rtcpPackets.get(i);

      if (rtcpPacket.getPacketType() == RtcpMessage.RTCP_PT_SR) {

        if (log.isDebugEnabled())
          log.debug(this + " processing received sender report");

        RtcpSenderReportMessage rtcpSenderReportPacket = (RtcpSenderReportMessage) rtcpPacket;
        int ssrc = rtcpSenderReportPacket.getSsrc();
        RtpParticipantInfo participantInfo = findParticipantInfo(ssrc);

        if (null == participantInfo) {
          participantInfo = new RtpReceiverInfo(ssrc);
          participantInfo.setAddress(address);
          participantInfo.setRtcpPort(port);
          participantInfos.add(participantInfo);
        } else {
          if (participantInfo.getAddress() == address) {
            if (participantInfo.getRtcpPort() == 0) {
              participantInfo.setRtcpPort(port);
            } else {
              // TODO check for ssrc conflict
            }
          } else {
            // TODO check for ssrc conflict
          }
        }
        participantInfo.processSenderReport(rtcpSenderReportPacket
            .getSenderReport(), JistAPI.getTime());

        List receptionReports = rtcpSenderReportPacket.getReceptionReports();
        for (int j = 0; j < receptionReports.size(); j++) {
          ReceptionReport receptionReport = (ReceptionReport) receptionReports
              .get(i);
          if (null != senderInfo) {
            if (receptionReport.getSsrc() == senderInfo.getSsrc()) {
              senderInfo.processReceptionReport(receptionReport, JistAPI
                  .getTime());
            }
          }
          // else
          // delete receiverReport;
        }
        receptionReports = null;

      } else if (rtcpPacket.getPacketType() == RtcpMessage.RTCP_PT_RR) {

        if (log.isDebugEnabled())
          log.debug(this + " processing received receiver report");

        RtcpReceiverReportMessage rtcpReceiverReportPacket = (RtcpReceiverReportMessage) rtcpPacket;
        int ssrc = rtcpReceiverReportPacket.getSsrc();
        RtpParticipantInfo participantInfo = findParticipantInfo(ssrc);
        if (participantInfo == null) {
          participantInfo = new RtpReceiverInfo(ssrc);
          participantInfo.setAddress(address);
          participantInfo.setRtcpPort(port);
          participantInfos.add(participantInfo);
        } else {
          if (participantInfo.getAddress() == address) {
            if (participantInfo.getRtcpPort() == 0) {
              participantInfo.setRtcpPort(port);
            } else {
              // check for ssrc conflict
            }
          } else {
            // check for ssrc conflict
          }
        }

        List receptionReports = rtcpReceiverReportPacket.getReceptionReports();
        for (int j = 0; j < receptionReports.size(); j++) {
          ReceptionReport receptionReport = (ReceptionReport) receptionReports
              .get(j);
          if (null != senderInfo) {
            if (receptionReport.getSsrc() == senderInfo.getSsrc()) {
              senderInfo.processReceptionReport(receptionReport, JistAPI
                  .getTime());
            }
          }

          // else
          // delete receiverReport;
        }
        receptionReports = null;
      } else if (rtcpPacket.getPacketType() == RtcpMessage.RTCP_PT_SDES) {

        if (log.isDebugEnabled())
          log.debug(this + " processing received sdes");

        RtcpSDESMessage rtcpSDESPacket = (RtcpSDESMessage) rtcpPacket;
        List sdesChunks = rtcpSDESPacket.getSdesChunks();

        for (int j = 0; j < sdesChunks.size(); j++) {
          // remove the sdes chunk from the cArray of sdes chunks
          SDESChunk sdesChunk = (SDESChunk) sdesChunks.get(j);
          // this is needed to avoid seg faults
          // sdesChunk.setOwner(this);
          int ssrc = sdesChunk.getSsrc();
          RtpParticipantInfo participantInfo = findParticipantInfo(ssrc);
          if (participantInfo == null) {
            participantInfo = new RtpReceiverInfo(ssrc);
            participantInfo.setAddress(address);
            participantInfo.setRtcpPort(port);
            participantInfos.add(participantInfo);
          } else {
            // check for ssrc conflict
          }
          participantInfo.processSDESChunk(sdesChunk, arrivalTime);
        }
        sdesChunks = null;

      } else if (rtcpPacket.getPacketType() == RtcpMessage.RTCP_PT_BYE) {

        if (log.isDebugEnabled())
          log.debug(this + " processing received bye packet");

        RtcpByeMessage RtcpByeMessage = (RtcpByeMessage) rtcpPacket;
        int ssrc = RtcpByeMessage.getSsrc();
        RtpParticipantInfo participantInfo = findParticipantInfo(ssrc);

        if (participantInfo != null && participantInfo != senderInfo) {
          participantInfos.remove(participantInfo);

          participantInfo = null;
          // perhaps it would be useful to inform
          // the profile to remove the corresponding
          // receiver module
        }
      }
      else {
        // app rtcp packets
        log.warn(this + " unable to handle unknown packet " + rtcpPacket);
      }
      rtcpPacket = null;
    }
    rtcpPackets = null;
  }

  /**
   * Returns the RtpParticipantInfo object used for storing information about
   * the rtp end system with this ssrc identifier. Returns null if this end
   * system is unknown.
   */
  protected RtpParticipantInfo findParticipantInfo(int ssrc) {

    for (int i = 0; i < participantInfos.size(); i++) {
      RtpParticipantInfo info = (RtpParticipantInfo) participantInfos.get(i);
      if (ssrc == info.getSsrc())
        return info;
    }

    return null;
  }

  /**
   * Recalculates the average size of an RTCPCompoundPacket when one of this
   * size has been sent or received.
   */
  protected void calculateAveragePacketSize(int size) {
    // add size of ip and udp header to given size before calculating
    averagePacketSize = ((double) (packetsCalculated) * averagePacketSize + (double) (size + 20 + 8))
        / (double) (++packetsCalculated);
  }

}
