package nxm.net.prim;

import nxm.net.lib.dgnetsend;
import nxm.sys.inc.InternalUseOnly;
import nxm.sys.lib.DataFile;
import nxm.sys.lib.ICEPacket;
import nxm.sys.lib.Message;
import nxm.sys.lib.MidasException;
import nxm.sys.lib.Primitive;
import nxm.sys.lib.TextFile;
import nxm.sys.lib.Time;

/**
  sinkdg - uses dgnetsend class to write (Raw data) to Multicast or unicast datagrams

  @author MSZ / Neon Ngo
  @since 3.7.2 moved to NET from SYS for NXM-3048
*/
public class sinkdg extends Primitive {
  private TextFile logfile;
  private dgnetsend dgsend = null;

  private static final int defaultport=18800,defaulttimeout=5000,defaultttl=1,defaultrtp=0;
  private static final int PLEN=2048;
  private byte[] inbuf=null;
  private DataFile hin=null;
  private short ptype=97;
  private int ssrc=1,blen,csrc0=0;
  private double freq=0.0,bw=0.0;

  private String host=null,hinformat=null;
  private int tl,port=defaultport,timeOut=defaulttimeout,ttl=defaultttl,rtp=defaultrtp;
  private boolean verbose=false,packetDebug=false,keepopen=true,opened=false,popened=false,rtptime=false,icein=false;
  /** output network packets from file/pipe data read? */
  private boolean outputPackets=true;

  /** default constructor */
  public sinkdg () {}

  public void setVerbose(boolean ptemp) { verbose=ptemp; }
  /** Set RTP packet type
   *  @param rtptype the RTP packet type
   */
  public void setRTPType(int rtptype) { ptype=(short)rtptype; }
  /** Set synchronization source "ID" for RTP output
   *  @param pssrc the synchronization source "ID" for RTP output
   */
  public void setRTPSSRC(int pssrc) { ssrc=pssrc; }
  /** Set contributing source "ID" for 0th index for RTP output
   *  @param pcsrc the contributing source "ID" for 0th index for RTP output
   */
  public void setRTPCSRC(int pcsrc) { csrc0=pcsrc; }
  /** Set output verbosity for debugging
   *  @param ptemp the state for whether or not to show vervose output
   */
  public void setSocketDebug(boolean ptemp) { if (dgsend!=null) try { dgsend.setVerbose(ptemp); } catch (Exception e) { e.printStackTrace(); } }
  /** Set state for whether or not to show all packets for debugging
   *  @param ptemp the state for whether or not to show all packets for debugging
   */
  public void setPacketDebug(boolean ptemp) { packetDebug=ptemp; if (dgsend!=null) try { dgsend.setPacketDebug(ptemp); } catch (Exception e) { e.printStackTrace(); } }
  /** Set RTP packet type and synchronization source "ID"
   *  @param pptype the RTP packet type
   *  @param pssrc  the synchronization source "ID"
   */
  public void setPTypeSSRC(int pptype,int pssrc) { ptype=(short)pptype; ssrc=pssrc; }
  /** Set RTP packet type, synchronization source "ID", and contributing source "ID"
   *  @param pptype the RTP packet type
   *  @param pssrc  the synchronization source "ID"
   *  @param pcsrc  the contributing source "ID"
   */
  public void setPTypeSRCs(int pptype,int pssrc,int pcsrc) { ptype=(short)pptype; ssrc=pssrc; csrc0=pcsrc; }
  /** Sets socket timeout value in milliseconds
   *  TODO Not yet implemented ??
   *  @param ptimeout the socket timeout value in milliseconds
   */
  public void setTimeOut(int ptimeout) { }
  /** Set multicast time-to-live
   *  @param pttl multicast time-to-live
   */
  public void setTTL(int pttl) { ttl=pttl; if (dgsend!=null) try { dgsend.setTTL(ttl); } catch (Exception e) { e.printStackTrace(); } }
  public String toString() { return "DG: Host="+host+" Port="+port; }
  /** Set whether or not to keep the socket open if switch is present
   *  @param pkeep whether or not to keep the socket open if switch is present
   */
  public void setKeepOpen(boolean pkeep) { keepopen=pkeep; if (dgsend!=null) try { dgsend.setKeepOpen(pkeep); } catch (Exception e) { e.printStackTrace(); } }
  /** Set host
   *  @param phost the host
   */
  public void setHost(String phost) { host=phost; if (dgsend!=null) try { dgsend.setHost(phost); } catch (Exception e) { e.printStackTrace(); } }
  /** Get host
   *  @return the host
   *  @since NeXtMidas 2.9.2
   */
  public String getHost() { return host; }
  /** Set port
   *  @param pport the port
   */
  public void setPort(int pport) { port=pport; if (dgsend!=null) try { dgsend.setPort(pport); } catch (Exception e) { e.printStackTrace(); } }
  /** Get port
   *  @return the host port
   *  @since NeXtMidas 2.9.2
   */
  public int getPort() { return port; }
  /** Set host and port
   *  @param phost the host
   *  @param pport the port
   */
  public void setHostPort(String phost,int pport) {
    host=phost;
    port=pport;
    if (dgsend!=null) {
      try {
         dgsend.setHost(host);
         dgsend.setPort(port);
         if (!opened) open(); //does a close and then a reopen to create a new socket
      } catch (Exception e) { M.printStackTrace(e); }
    } else M.info("sinkdg: class not created ");
  }

  /** Get Dgnetsend socket for outputting Multicast messages
   *  @return the Dgnetsend socket for outputting Multicast messages
   *  @since NeXtMidas 3.0.0
   */
  public dgnetsend getDgnetsend() { return dgsend; }

  /** Get Dgnetsend socket for outputting Multicast messages
   *  @return the Dgnetsend socket for outputting Multicast messages
   */
  @InternalUseOnly("since initial version in NeXtMidas 3.5.4 for debugging purposes only")
  public dgnetsend getDgsend() { return dgsend; }

  /** Turn on/off output of network packets for data read from file/pipe.
      @param val set to false to pause sending network packets, true to on/resume
      @since NeXtMidas 2.9.0
   */
  public void setOutputPackets (boolean val) {
    outputPackets = val;
  }
  /** Get current output network packets state.
      @return true for on/send, false for off/paused.
      @since NeXtMidas 2.9.0
   */
  public boolean getOutputPackets () { return outputPackets; }

  public int open () {
    try {
      if (dgsend==null) {
        dgsend = new dgnetsend();
        if (dgsend==null) { M.error("sinkdg: Could not open socket"); return FINISH; }
        if (MA.find("/VERBOSE")) verbose=true;
        if (MA.find("/PACKETS")) packetDebug=true;
        timeOut = MA.getL("/TIMEOUT",defaulttimeout);
        if (MA.find("/KEEPOPEN")) keepopen=true;
        ttl = MA.getL("/TTL",defaultttl);
        rtp = MA.getL("/RTP",defaultrtp);
        rtptime = MA.getState("/RTPTIME",rtptime);
        icein = MA.getState("/ICEIN",icein);
        ptype = MA.getI("/PTYPE",ptype);
        ssrc = MA.getL("/SSRC",ssrc);
        csrc0 = MA.getL("/CSRC",csrc0);
        tl = MA.getL("/TL",PLEN);
        String hostport=MA.getS("/HOSTPORT","");

        if (MA.find("/LOG")) {
          logfile = MA.getTextFile("/LOG");
          if (!logfile.open(TextFile.OUTPUT|TextFile.NOABORT)) {
            logfile.writeln("sinkdg log started at "+Time.tag());
          } else {
            M.warning("Could not find URL="+logfile.getURL());
            logfile = null;
          }
        }
        setVerbose(verbose);
        setPacketDebug(packetDebug);
        setTTL(ttl);
        if (verbose) {
          M.info("sinkdg: Verbose="+verbose+". PacketDebug="+packetDebug+". Keepopen="+keepopen);
          M.info("sinkdg: TTL="+ttl);
        }
        if (keepopen && host==null) {
          int i=hostport.indexOf(':');
          if (i>0) {
            host=hostport.substring(0,i).toLowerCase();
            port=Integer.parseInt(hostport.substring(i+1));
            M.info("sinkdg: HOSTPORT="+hostport);
          }
        }
      } // end if (dgsend==null)

      // 2010-12-07 NTN: moved the below 4 line outside the above if (dgsend==null) block
      //                 so that SINKDG can support pipe restarts - BUG 2300
      hin = MA.getDataFile ("IN","1000,2000","",0);
      popened=false;
      if (hin==null) M.info("Invalid Input File - 1000,2000 only");
      hinformat=hin.getFormat();

      if (verbose && host!=null) M.info("sinkdg: HOST,PORT="+host+","+port);
      if (host!=null && hin!=null && !popened) {
        String fname = MA.getS("IN");
        if (fname.length()>0) {
          M.info("Opening hin "+fname+","+hin.isOpen()+" for "+host);
          hin.open();
          hinformat=hin.getFormat();
          popened=true;
          inbuf = new byte[tl];
          keepopen=true;
        } else { M.info("sinkdg Invalid Input File"); hin.close(); hin=null; }
      }

      if (keepopen) {
        if (host!=null) { //this hasn't been set yet
          dgsend.open(host,port);
          opened=true;
          if (verbose) M.info("sinkdg: Host="+host+". Port="+port);
        } else M.info("sinkdg: host is null");
      } else {
        if (host!=null) { //this hasn't been set yet
          dgsend.open();
          opened=true;
          if (verbose) M.info("sinkdg: Host="+host+". Port="+port);
        }
      }
    } catch (Exception e) {
      throw new MidasException("sinkdg: open failed", e);
    }
    return NORMAL;
  }

  @Override
  public int close () {
    try {
      if (logfile!=null) {
        logfile.writeln("sinkdg log stopped at "+Time.tag());
        logfile.close();
      }
      if (hin!=null) hin.close();
      if (dgsend!=null) dgsend.close();
    } catch (Exception e) {
      throw new MidasException("sinkdg: close failed", e);
    }
    return NORMAL;
  }

  /** Send packet
   *  @param buf buffer containing data to send
   *  @return 1 if sent successfully, 0 if unable to send, -1 for invalid input or error state */
  public int send (byte[] buf) {
    try {
      if (verbose) M.info("senddg: send(byte[] buf) Packet data="+new String(buf));
      if (buf==null || dgsend==null) return -1;
      boolean sent = dgsend.write(buf);
      if (!sent) return 0; // sent failed (since 3.3.0)
    } catch (Exception e) {
      throw new MidasException("sinkdg: send failed", e);
    }
    return 1;
  }

  /** Send packet
   *  @param data data to send
   *  @return 1  if sent successfully, 0 if unable to send, -1 for invalid input or error state */
  public int send (String data) {
    try {
      if (verbose) M.info("senddg: send(String data) Packet data="+data);
      if (data==null || dgsend==null) return -1;
      byte[] buf = data.getBytes();
      boolean sent = dgsend.write(buf);
      if (!sent) return 0; // sent failed (since 3.3.0)
    } catch (Exception e) {
      throw new MidasException("sinkdg: send failed", e);
    }
    return 1;
  }

  /** Send the data to the specified destination
   *  @param host destination address
   *  @param port destination port
   *  @param data data to send
   *  @return 1 if sent successfully, 0 if unable to send, -1 for invalid input or error state */
  public int send (String host, String port, String data) {
    try {
      if (verbose) M.info("senddg: send(host,port,data)->Packet data="+data);
      if (host==null || data==null || dgsend==null) return -1;
      int iPort = Integer.parseInt(port);
      if (iPort<0 || iPort>65535) return -1;
      boolean sent = dgsend.write(host,iPort,data);
      if (!sent) return 0; // sent failed (since 3.3.0)
    } catch (Exception e) {
      throw new MidasException("sinkdg: send failed", e);
    }
    return 1;
  }

  /** Send RTP packet
   *  @param buf   buffer containing data to send
   *  @param bsize buffer size
   *  @return 1 if sent successfully, 0 if unable to send, -1 for invalid input or error state */
  public int sendRTP (byte[] buf, int bsize) {
    try {
      if (verbose) M.info("senddg: send(byte[] buf) Packet data="+new String(buf));
      if (buf==null || dgsend==null) return -1; // sent failed (invalid data or null dgnetsend library)
      boolean sent = dgsend.writeRTP(buf,bsize,ptype,ssrc,0,csrc0);
      if (!sent) return 0; // sent failed (since 3.3.0)
    } catch (Exception e) {
      throw new MidasException("sinkdg: sendRTP failed", e);
    }
    return 1; // sent successful
  }

  /** Send RTP Packet
   *  @param data data to send
   *  @return 1 if sent successfully, 0 if unable to send, -1 for invalid input or error state */
  public int sendRTP (String data) {
    try {
      if (verbose) M.info("senddg: send(String data) Packet data="+data);
      if (data==null || dgsend==null) return -1; // sent failed (invalid data or null dgnetsend library)
      byte[] buf = data.getBytes();
      boolean sent = dgsend.writeRTP(buf,buf.length,ptype,ssrc,0,csrc0);
      if (!sent) return 0; // sent failed (since 3.3.0)
    } catch (Exception e) {
      throw new MidasException("sinkdg: sendRTP failed", e);
    }
    return 1; // sent successful
  }

  public int process () {
    if (inbuf==null) return NOOP;

    blen=hin.read(inbuf,0,tl);
    if (blen==0) return NOOP;
    if (blen<0) return FINISH;
    if (outputPackets == false) return NORMAL;

    if (rtp>0) {
      if (rtptime) {
        if (icein) {
          dgsend.writeRTPICE(inbuf,blen,ptype,ssrc,csrc0); //takes the ice header from start of pipe
        } else {
          dgsend.writeRTPICE(inbuf,blen,hinformat,Time.current(),freq,bw,ptype,ssrc,0,csrc0);
        }
      } else {
        if (icein) dgsend.writeRTP(inbuf,blen-ICEPacket.headerLength,ptype,ssrc,ICEPacket.headerLength,csrc0); //skip iceheader
        else       dgsend.writeRTP(inbuf,blen,ptype,ssrc,0,csrc0);
      }
    } else dgsend.write(inbuf);
    return NORMAL;
  }

  public synchronized int processMessage (Message msg) {
    try {
      if (verbose) M.info("sinkdg: Processing a message: "+msg.name);
      if (msg.name.equals("OPENNET")) {
        if (keepopen) {
          String hostport=(String)msg.data;
          int i=hostport.indexOf(':');
          if (i>0) {
            host=hostport.substring(0,i).toLowerCase();
            port=Integer.parseInt(hostport.substring(i+1));
            dgsend.open(host,port);
            if (verbose) M.info("sinkdg: Host="+host+". Port="+port);
          } else {
            M.error("sinkdg: open failed: host:port wrong"+host+":"+port+". hostport="+hostport);
          }
        } else dgsend.open();
      } else if (msg.name.equals("SENDNET")) {
//      if (keepopen)  { dgsend.write(msg.data); }
//      else {
//        String hostport=msg.info;
//        int i=hostport.indexOf(':');
//        if (i>0) {
//          host=hostport.substring(0,i).toLowerCase();
//          port=Integer.parseInt(hostport.substring(i+1));
//          if (verbose) M.info("sinkdg: Host="+host+". Port="+port);
//          dgsend.write(host,port,msg.data);
//        }
//      }
      } else if (msg.name.equals("CLOSENET")) {
        dgsend.close();
      } else if (msg.name.equals("EXITNET")) {
        dgsend.close();
        return FINISH;
      }
    } catch (Exception e) {
      throw new MidasException("sinkdg: processMessage failed", e);
    }
    return NORMAL;
  }

  /** Get synchronization source "ID" for RTP output
   *  @return the synchronization source "ID" for RTP output
   */
  public int getSSRC() { return ssrc; }

}
