package nxm.sys.prim;

//import nxm.sys.net.DatagramData;                 // referenced directly to avoid excess deprecation warnings
//import nxm.sys.net.DatagramData.IceTimeCallBack; // referenced directly to avoid excess deprecation warnings
import java.util.Arrays;
import java.util.StringTokenizer;

import nxm.sys.inc.MessageHandler;
import nxm.sys.lib.Convert;
import nxm.sys.lib.Data;
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.StringUtil;
import nxm.sys.lib.Table;
import nxm.sys.lib.TextFile;
import nxm.sys.lib.Time;
import nxm.sys.lib.XmlFile;

/**
  sourcedg - uses DatagramData class to reads (Raw or RTP) from Multicast or unicast datagram stream
  @deprecated Use {@link nxm.net.prim.sourcedg} instead - since 3.7.2
  @author MSZ
*/
@Deprecated
public class sourcedg extends Primitive implements nxm.sys.net.DatagramData.IceTimeCallBack {
  //may be multicast or unicast and may or may not be rtp or sap - SB in only
  private TextFile logfile;
  private DataFile pout,convout;          //pout is normal output, convout is for audio streams - converted from ULAW to Lin16
  private Data outdata;                   //hold convout data
  private nxm.sys.net.DatagramData dgd = null;

  private boolean abort=false, dgdclosed=true, linear=true, linearin=true, init=false;

  private short[] outbuf;
  private short[] decodeTable;
  private Table t3kstruct = null;
  private byte[] buf;
  private byte[] fbuf = new byte[1024];

  private static final int MAXPKTBUF=32768;
  private static final String defaulthost="127.0.0.1",defaultxmlmsgname="XMLMSGS";
  private static final int defaultport=18800,defaulttimeout=50,defaultintl=1000,defaulttll=128;
  private static final double defaultxstart=0.0,defaultxdelta=1.0;

  private String host=defaulthost,midasdatatype="SB";
  private int port=defaultport,timeOut=defaulttimeout;
  private int intl=defaultintl,outtl=intl*2,midasfiletype=1000,tll=defaulttll;
  private int offsetdata=0;         //offset of where to start reading data in packets
  private String msgnameforxml=defaultxmlmsgname;  //the XML main name sent to processMessage
  private Object msgid=null;        //where to send messages, from /MSGID=
  private int rtpfiltertype=-1;     //filter on specific rtp type - used to pull a "channel" out of stream based on type
  private String msgidname=null;      //msgid's name

  private int curlen=0; //curlen is last >0 length
  private int pktfrcount=0; //bytes written to current frame so far
  private int fullwrites,remain;   //used to calculate fill bytes needed per packet
  /** basically for startup condition to find a header = Current HeaDeR, but not always updated */
  private int chdr=-1;
  private int len,offset,fillnum,pktsinframe,seq,pcount,psize,i,j;
  private byte fillValue = 0;       // value used to fill missing buffers
  private int framesz=0;
  private double ystartjtime=0;
  long numrecs;

  private boolean verbose=false;      //debugging messages
  private boolean packetDebug=false;  //Datagram data prints out detailed packet info
  private boolean rtp=true;           //rtp packets
  private boolean sap=false;          //sap type packets
  private boolean alladdr=false;      //set to a port and any address
  private boolean xml3000out=true;    //converts XML input into 3000 file output - tags must match 3000 file columns
  private boolean icehdrinpipe=false; //include icehdr in data?
  private boolean icehdrframe=false;  //set framesize from ice header data
  private boolean fill=false;         //fill requires constant packet sizes, doesn't work with variable data packets
  private boolean msgonly=false;      //used to output only Tables converted from XML to processMessage loop

  private boolean doTimeCode = false; // Indicates if time-code processing is being performed.
  private boolean doXSXD     = true;  // Used with doTimeCode to enable updates of XS and XD.

  /** The time code start time and delta */
  private double xStart=defaultxstart;
  private double xDelta=defaultxdelta;
  private double dgdxs=0.0, dgdxd=0.0, dgjtime=0.0;
  private int dgdsize=0, dgdbytes=0;
  private String dgdformat=null;

  /** For debugging, converts time of the datagram to readable form. */
  private Time dgTime = new Time();

///////////////////////////////////////////////////////////////////////////////////////////////////

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

  /** Get if verbose is enabled
   *  @return true if verbose is set */
  public boolean getVerbose () { return verbose; }
  /** Set verbose status
   *  @param ptemp true if verbose */
  public void setVerbose (boolean ptemp) { verbose=ptemp; }
  /** Set debug status
   *  @param ptemp true if debug */
  public void setDebug (boolean ptemp) { verbose=ptemp; }  //same as setVerbose

  /** Get if debugging DatagramData socket settings/major events
   *  @return is debugging DatagramData socket settings/major events
   *  @since NeXtMidas 2.7.4
   */
  public boolean getSocketDebug () {
    if (dgd==null) return false;
    try {
      return dgd.getVerbose();
    } catch (Exception e) {
      e.printStackTrace();
    }
    return false;
  }

  /** Set if debugging DatagramData socket settings/major events
   *  @param ptemp True if to debug socket, false if not
   */
  public void setSocketDebug (boolean ptemp) {
    if (dgd==null) return;
    try {
      dgd.setVerbose(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get if debugging DatagramData packets, e.g. to print out detailed packet info.
   *  @return if packet debug is set
   *  @since NeXtMidas 2.7.4
   */
  public boolean getPacketDebug() { return packetDebug; }

  /** Set if debugging DatagramData packets, e.g. to print out detailed packet info.
   *  @param ptemp True if to debug packets, false if not
   */
  public void setPacketDebug (boolean ptemp) {
    packetDebug=ptemp;
    if (dgd==null) return;
    try {
      dgd.setPacketDebug(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get is reading RTP type packets
      @return is reading RTP type packets
      @since NeXtMidas 2.7.4
   */
  public boolean getRTP() { return rtp; }

  /** Set if reading RTP packets
   *  @param ptemp true if reading RTP packets
   */
  public void setRTP (boolean ptemp) {
    rtp=ptemp;
    if (dgd==null) return;
    try {
      dgd.setRTP(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**  Get if reading SAP packets
   *   @return true if reading SAP packets
   *   @since NeXtMidas 2.7.4
   */
  public boolean getSAP() { return sap; }

  /** Set if reading SAP packets
   *  @param ptemp true if reading SAP packet
   */
  public void setSAP (boolean ptemp) {
    sap=ptemp;
    if (dgd==null) return;
    try {
      dgd.setSAP(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get filter on specific RTP type - used to pull a "channel" out of stream based on type.
      @return -1 if no RTP filter is set, otherwise RTP type that being filtered
      @since NeXtMidas 2.7.4
   */
  public int getRTPFilter () { return rtpfiltertype; }
  /** Set filter on specific RTP type - used to pull a "channel" out of stream based on type.
      @param ptemp the filter (-1 means no RTP filter is set, otherwise RTP type that being filtered)
   */
  public void setRTPFilter (int ptemp) {
    rtpfiltertype=ptemp;
    if (dgd==null || rtpfiltertype<0) return;
    try {
      dgd.setRTPFilter(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get host address
   *  @return the host address
   */
  public String getHost () { return host; }
  /** Set host address
   *  @param ptemp the host address
   */
  public void setHost (String ptemp) {
    host=ptemp;
    if (dgd==null) return;
    try {
      dgd.setHost(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  /** Get address port
   *  @return the address port
   */
  public int getPort () { return port; }
  /** Set address port
   *  @param ptemp address port
   */
  public void setPort (int ptemp) {
    port=ptemp;
    if (dgd==null) return;
    try {
      dgd.setPort(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get socket timeout
   *  @return socket timeout (ms) to use for Datagrams.
   *  @since NeXtMidas 2.7.4
   */
  public int getTimeOut() { return timeOut; }

  /** Set the timout to use for datagrams
   *  @param ptemp the timeout
   */
  public void setTimeOut (int ptemp) {
    timeOut=ptemp;
    if (dgd==null) return;
    try {
      dgd.setTimeOut(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }


  /** Are all hosts are being listened to on a port?
   *  @return is listening to a specified port and any address (host)
   *  @since NeXtMidas 2.7.4
   */
  public boolean getAllHosts () { return alladdr; }
  /** Set the state of whether or not all hosts are being listened to on a port
   *  @param ptemp the state of whether or not all hosts are being listened to on a port
   */
  public void setAllHosts (boolean ptemp) {
    alladdr=ptemp;
    if (dgd==null) return;
    try {
      dgd.setAllHosts(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get the data offset for each read
   *  @return the data offset for each read
   */
  public int getOffset () { return offsetdata; }
  /** Set the data offset for each read
   *  @param ptemp the data offset for each read
   */
  public void setOffset (int ptemp) {
    if (ptemp<0 || ptemp>MAXPKTBUF) offsetdata=0;
    else offsetdata=ptemp;
  }
  /** Is XML3000 file mode being used, or NORMAL3000 file mode?
   *  @return true if XML3000 file mode is being used
   */
  public boolean getXML3000 () { return xml3000out; }
  /** Set the state of whether or not XML3000 file mode is being used
   *  @param ptemp the state of whether or not XML3000 file mode is being used
   */
  public void setXML3000 (boolean ptemp) { xml3000out=ptemp; }
  /** Get TimeLine length
   *  @return the TimeLine length
   */
  public int getTimeLineLength () { return tll; }
  /** Set TimeLine length
   *  @param ptemp the TimeLine length
   */
  public void setTimeLineLength (int ptemp) {
    if (ptemp<0) tll=defaulttll;
    else tll=ptemp;
  }
  /** Is there an ICE header frame?
   *  @return whether or not there is an ICE header frame
   */
  public boolean getIceHdrFrame() { return icehdrframe; }
  /** Set the state of whether or not there is an ICE header frame
   *  @param ptemp the state of whether or not there is an ICE header frame
   */
  public void setIceHdrFrame (boolean ptemp) { icehdrframe=ptemp; }

  /** Is there an ICE header in the pipe?
   *  @return whether or not there is an ICE header in the pipe
   */
  public boolean getIceHdrInPipe () { return icehdrinpipe; }
  /** Set the state of whether or not there is an ICE header in the pipe?
   *  @param ptemp the state of whether or not there is an ICE header in the pipe
   */
  public void setIceHdrInPipe (boolean ptemp) {
    icehdrinpipe=ptemp;
    if (dgd==null) return;
    try {
      dgd.setIceHdrInPipe(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get the value in which every value in the fill buffer will be set to
   *  @return the value in which every value in the fill buffer will be set to
   */
  public byte getFillValue() { return fillValue; }
  /** Set the value in which every value in the fill buffer will be set to
   *  @param filler the value in which every value in the fill buffer will be set to
   */
  public void setFillValue(byte filler) {
    fillValue = filler;
  }

  /** Fill in missing packets?
   *  @return whether or not to fill in missing packets
   */
  public boolean getFill () { return fill; }
  /** Set the state of whether or not to fill in missing packets
   *  @param ptemp the state of whether or not to fill in missing packets
   */
  public void setFill (boolean ptemp) {
    fill=ptemp;
    if (fill) {
//    setIceHdrInPipe(false); //why am I setting this here? comment out in 3.6.0 per user request: TODO: remove in 3.7.1+ if no issues
      setRTP(true);
      dgd.setIceData(true);
      setIceHdrFrame(true); //need to fix this for Fill with no ICE Hdr in Frame
      if (verbose) M.info("Fill requires TC processing and NO XS-XD restarts");
    }
    if (dgd!=null) {
      try {
        dgd.setFill(ptemp);
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }
  /** Set the value for every element in the fill buffer
   *  @param val the value for every element in the fill buffer
   */
  public void setFillBufValue (byte val) { Arrays.fill(fbuf,val); }
  /** Set the fill buffer for data output
   *  @param pfbuf the fill buffer
   *  @return the length of the fill buffer array
   */
  public int setFillBuf (byte[] pfbuf) {
    try {
      if (pfbuf.length>0) {
        fbuf = new byte[pfbuf.length];
        System.arraycopy(pfbuf,0,fbuf,0,fbuf.length);
      }
    } catch (Exception e) {
      e.printStackTrace();
      M.error("sourcedg: setFillBuff failed:"+e);
    }
    if (fbuf==null) return 0;
    return fbuf.length;
  }

  /** Get message id name
   *  @return the message id name
   */
  public String getMSGID() { return msgidname; }
  /** Set message id name
   *  @param ptemp the message id name
   */
  public void setMSGID(String ptemp) {
    msgidname=ptemp;
    msgid=null;
    if (msgidname!=null && msgidname.length()>0) msgid = M.registry.getCommand(msgidname,5);
    if (msgonly && (msgid==null || !(msgid instanceof MessageHandler))) {
      M.warning("MSGONLY requires valid /MSGID - msgonly turned off");
      setMsgOnly(false);
    }
  }

  /** Indicates whether or not to send messages only and not pipes
   *  @return the state of whether or not to send just messages
   */
  public boolean getMsgOnly() { return msgonly; }
  /** Sets the state of whether or not to send messages only
   *  @param ptemp state of whether or not to send messages only
   */
  public void setMsgOnly(boolean ptemp) { msgonly=ptemp; }
  /** Indicates whether or not the output pipe will receive TimeCode data
   *  via a call-back from DatagramData
   *  @return the state of receiving TimeCode data
   */
  public boolean getTimeCode() { return doTimeCode; }
  /** Sets the state of whether or not the output pipe will receive TimeCode data
   *  via a call-back from DatagramData
   *  @param ptemp the state of receiving TimeCode data
   */
  public void setTimeCode(boolean ptemp) { doTimeCode=ptemp; }
  /** Set the state of restarting the pipe on XS,XD changes
   *  @return the state of restarting the pipe on XS,XD changes
   */
  public boolean getXSXD() { return doXSXD; }
  /** Restart pipe on XS,XD changes?
   *  @param ptemp new state for restarting pipe on XS,XD changes
   */
  public void setXSXD(boolean ptemp) { doXSXD=ptemp; }
  /** Get Midas output data type
   *  @return the Midas output data type
   */
  public String getMidaDataType() { return midasdatatype; }
  /** Set Midas output data type
   *  @param ptemp the Midas output data type
   */
  public void setMidasDataType(String ptemp) {
    if (ptemp==null || ptemp.length()<2) midasdatatype="SB";
    else midasdatatype=ptemp;
  }
  /** Get the X-axis time code start time
   *  @return the X-axis time code start time
   */
  public double getXStart() { return xStart; }
  /** Set the X-axis time code start time
   *  @param ptemp the X-axis time code start time
   */
  public void setXStart(double ptemp) { xStart=ptemp; }
  /** Get the X-axis time code delta
   *  @return the X=axis time code delta
   */
  public double getXDelta() { return xDelta; }
  /** Set the X-axis time code delta
   *  @param ptemp the X-axis time code delta
   */
  public void setXDelta(double ptemp) { xDelta=ptemp; }
  /** Get the Y-axis start time
   *  @return the Y-axis start time
   */
  public double getYStartTime() { return ystartjtime; }
  /** Set the Y-axis start time
   *  @param ptemp the Y-axis start time
   */
  public void setYStartTime(double ptemp) { ystartjtime=ptemp; }
  /** Get the datagram packet X-axis start time
   *  @return the datagram packet X-axis start time
   *  @since NeXtMidas 3.0.0
   */
  public double getPacketXStart() { return dgdxs; }
  /** Get Datagram packet X-axis delta
   *  @return the Datagram packet X-axis delta
   *  @since NeXtMidas 3.0.0
   */
  public double getPacketXDelta() { return dgdxd; }
  /** Get Datagram packet julian time
   *  @return the Datagrapm packet julian time
   *  @since NeXtMidas 3.0.0
   */
  public double getPacketJtime() { return dgjtime; }
  /** Get Datagram packet size
   *  @return the Datagram packet size
   *  @since NeXtMidas 3.0.0
   */
  public int getSize() { return dgdsize; }
  /** Get the number of bytes in the Datagram packet
   *  @return the number of bytes in the Datagram packet
   *  @since NeXtMidas 3.0.0
   */
  public int getBytes() { return dgdbytes; }
  /** Get Datagram packet format
   *  @return the Datagram packet format
   *  @since NeXtMidas 3.0.0
   */
  public String getFormat() { return dgdformat; }

  /** Get the XML main name sent to processMessage
   *  @return the XML main name sent to processMessage
   */
  public String getXMLMSGName() { return msgnameforxml; }
  /** Set the XML main name sent to processMessage
   *  @param ptemp the XML main name sent to processMessage
   */
  public void setXMLMSGName(String ptemp) {
    if (ptemp==null || msgnameforxml.length()<1) msgnameforxml=defaultxmlmsgname;
    else msgnameforxml=ptemp;
  }

  public String toString () { return "DG: Host="+host+" Port="+port+" RTP="+rtp; }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  public int open () {
    try {
      if (!init) { // skip the below initialization block on a RESTART
        init=true;
        // 2010-12-21 only init DatagramData once
        dgd = new nxm.sys.net.DatagramData();
        if (dgd==null) { M.error("sourcedg: Could not open socket"); return FINISH; }
        if (MA.find("/VERBOSE")) setVerbose(true);
        if (MA.find("/TIMECODE")) setTimeCode(true);
        if (MA.find("/PACKETS")) setPacketDebug(true);
        if (MA.find("/NOXSXD")) setXSXD(false);
        if (MA.find("/NORTP")) setRTP(false);
        if (MA.find("/ULAW")) { linear=false; linearin=false; }
        if (MA.find("/SAP")) setSAP(true);
        if (MA.find("/ALLHOSTS")) setAllHosts(true);
        if (MA.find("/MIDASSI")) setMidasDataType("SI");
        if (MA.find("/ICEHDRINPIPE")) setIceHdrInPipe(true);
        if (MA.find("/ICEHDRFRAME")) setIceHdrFrame(true);
        if (MA.find("/NORMAL3000")) setXML3000(false);
        if (MA.find("/MSGONLY")) setMsgOnly(true);  //wgd used XMLTBL but MSGONLY does same thing - XMLTBL XML Message always sent to MAIN
        setFrameSize(MA.getL("/FRAMESIZE",1024),false);
        if (MA.find("/FILL")) { setFill(true); setFillValue(MA.getB("/FILLVALUE", (byte)0)); }
        setAutoSeqResetMAX(MA.getL("/AUTOFILLRESET",0)); //to automatically reset fill stuff on certain counts
        setMSGID(MA.getU("/MSGID"));
        setTimeOut(MA.getL("/TIMEOUT",defaulttimeout));
        setRTPFilter(MA.getL("/RTPFILTER",-1));
        setOffset(MA.getL("/OFFSET",0));

        midasfiletype = MA.getL("/FTYPE",1000);
        intl = MA.getL("/INTL",defaultintl);
        if (intl<=1) intl=defaultintl;

        setXStart(MA.getD("/XSTART",defaultxstart));
        setXDelta(MA.getD("/XDELTA",defaultxdelta));
        setYStartTime(MA.getD("/INITIALTIME",-1.0));    //Initial ystart time (-1=Do not set, use qualifiers)
        setTimeLineLength(MA.getL("/TLL",defaulttll));  //TimeLine Length
        setXMLMSGName(MA.getS("/MSGNAME",defaultxmlmsgname));
        // check for log file
        if (MA.find("/LOG")) {
          logfile = MA.getTextFile("/LOG");
          if (logfile.open(TextFile.OUTPUT|TextFile.NOABORT)) {
            logfile.writeln("sourcedg log started at "+Time.tag());
          } else {
            M.warning("sourcedg: Unable to open log file "+logfile.getName());
            logfile = null;
          }
        }
        String hostport=MA.getS("HOSTPORT","");
        if (verbose) M.info("HOSTPORT="+hostport);
        i=hostport.indexOf(':');
        if (i<0) {
          if (sap) { host="224.2.127.254"; port=9875; }
          else { host=defaulthost; port=defaultport; }
        } else {
          host=hostport.substring(0,i).toLowerCase();
          port=Integer.parseInt(hostport.substring(i+1));
        }
        setHost(host);
        setPort(port);

        setFillBufValue(fillValue);

        outbuf = new short[MAXPKTBUF];
        buf = new byte[MAXPKTBUF];
        decodeTable = build_mu_decode_table();
      } //end initial startup

      chdr=-1; //reset current packet header
      if (!linear) outtl=intl*2; //2 is for linear out and byte in
      else outtl=intl;

      if (verbose) {
        M.info("sourcedg: TimeOut="+timeOut+". Verbose="+verbose+". PacketDebug="+packetDebug);
        M.info("sourcedg: RTP="+rtp+". XStart="+xStart+". XDelta="+xDelta);
        M.info("sourcedg: Out Type="+midasdatatype+" .SAP="+sap+" .ALL Hosts="+alladdr);
        M.info("sourcedg: Host="+host+". Port="+port);
      }

      if (msgonly) { //no pipes for messages so can stop here
        if (verbose) M.info("sourcedg: Messaging only");
        if (!getDgState()) {
          if (MA.find("/ON")) setDgState(MA.getState("/ON"));
        }
        return NORMAL;
      }

      pout = MA.getDataFile("OUT",""+midasfiletype,midasdatatype,DataFile.OUTPUT,0,null); //name,types,formats,0=size
      if (verbose) M.info("Pout="+pout);
      if (midasfiletype==2000) {
        pout.setType(midasfiletype);
        pout.setFrameSize(framesz);
        if (ystartjtime>=0) { pout.setYStart(ystartjtime); }
      } else if (midasfiletype==3000) {
        pout.setType(midasfiletype);
        try {
          t3kstruct = MA.getTable("/STRUCT");
          String tempstr="(";
          for (Table.Iterator ti=t3kstruct.iterator(); ti.getNext(); ) {
            tempstr=tempstr+ti.key+"|"+ti.value+",";
          }
          tempstr=tempstr.substring(0,tempstr.lastIndexOf(","));
          tempstr=tempstr+")";
          if (verbose) M.info("sourcedg: Setting 3000 file output");
          pout.setSubRecords(tempstr);
        } catch (Exception e) {
          throw new MidasException("sourcedg: Type 3000 file structure not defined. Use /STRUCT switch", e);
        }
      }
      pout.setTimeLineLength(tll);
      pout.setXStart(xStart);
      pout.setXDelta(xDelta);
      if (ystartjtime>=0) { pout.setTimeAt(ystartjtime); }
      if (doTimeCode){
        if (doXSXD) { pout.setXUnits(nxm.sys.inc.Units.FREQUENCY); }
        pout.setYUnits(nxm.sys.inc.Units.TIME);
        dgd.registerTimeCB(this);
      }
      pout.open();

      if (MA.find("CONVOUT")) {
        if (verbose) M.info("sourcedg: Convout created");
        convout = MA.getDataFile("CONVOUT","1000","SI",DataFile.OUTPUT,1,null);
        convout.setXStart(xStart);
        convout.setXDelta(xDelta);
        if (doTimeCode && doXSXD) convout.setXUnits(nxm.sys.inc.Units.TIME);
        convout.open();
        outdata=convout.getDataBuffer(outtl,INT);
      }

      if (getDgState()) setDgState("ON");
      else if (MA.find("/ON")) setDgState(MA.getState("/ON"));

    } catch (Exception e) {
      throw new MidasException("sourcedg: open failed", e);
    }

    return NORMAL;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  public int process () {
    if (dgdclosed) return NOOP;
    if (abort) return NORMAL;
    return (read())? NORMAL: NOOP;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  public int close () {
    try {
      if (verbose) System.out.println("Closing");
      if (dgd!=null) { dgd.close(); } //doesnt clear seq numbers
      FillPipeAtEnd();
      if (pout!=null) { pout.update(); pout.close(); }
      if (convout!=null) { convout.update(); convout.close(); }
      if (logfile!=null) {
        logfile.writeln("sourcedg log stopped at "+Time.tag());
        logfile.close();
      }
    } catch (Exception e) {
      throw new MidasException("sourcedg: close failed", e);
    }
    return NORMAL;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Fill in left over data at the end */
  public void FillPipeAtEnd() {
    try {
      if (fill && framesz>0) { //check for left over data on close and fill in left over
        if (verbose) M.info("sourcedg: FillPipeAtEnd and checking for leftover data.");
        if (curlen>0) {
          if (pktfrcount<framesz && pktfrcount>0) {
            fillbuf(fbuf,framesz-pktfrcount);
            if (verbose) M.info("sourcedg: FillPipeAtEnd and filling buffer with "+(framesz-pktfrcount));
            pktfrcount=framesz;
          }
        }
      }
    } catch (Exception e) {
      throw new MidasException("sourcedg: FillPipeAtEnd failed", e);
    }
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  public synchronized int processMessage (Message msg) {
    if (verbose) M.info("sourcedg: Processing a message: "+msg.name);
    if (msg.name.equals("OPENNET")) {
      setDgState("OFF");
      String hostport=(String)msg.data;
      i=hostport.indexOf(':');
      if (i<0) {
        host=defaulthost;
        port=defaultport;
      } else {
        host=hostport.substring(0,i).toLowerCase();
        port=Integer.parseInt(hostport.substring(i+1));
      }
      if (verbose) M.info("sourcedg: Host="+host+". Port="+port);
      setHost(host);
      setPort(port);
      setDgState("ON");
    } else if (msg.name.equals("CLOSENET")) {
      setDgState("OFF");
    } else if (msg.name.equals("EXITNET")) {
      setDgState("OFF");
      close();
      return FINISH;
    }
    return NORMAL;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
//  /** @deprecated Since NeXtMidas 2.5.0 - This illegally overrides {@link Command#setState(String)}.
//                  use {@link #setDgState(String)} in place of this.
//   */
//  public void setState(String pvalue) {
//    setDgState(pvalue);
//  }


  /**
   * Turns on/off sourcedg.
   * @param pvalue String value of true
   *  @since NeXtMidas 2.5.0
   */
  public void setDgState(String pvalue) {
    setDgState(StringUtil.isTrue(pvalue));
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Turns on/off sourcedg.
   *  @param on Datagram State ?true if done?
      @since NeXtMidas 2.5.0
   */
  public void setDgState(boolean on) {
    if (verbose) M.info("sourcedg: Set DgState:"+on);
    try {
      chdr=-1;
      if (!dgdclosed) {
        dgdclosed=true;
        if (dgd!=null) dgd.leave();
        FillPipeAtEnd();
      }
      if (on && dgd!=null) dgdclosed=!dgd.open();
    } catch (Exception e) {
       e.printStackTrace();
       M.error("sourcedg: setDgState failed:"+e);
    }
    if (verbose) M.info("sourcedg: Done:"+on+" and closed set="+dgdclosed);
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////

  /** Get if sourcedg is on/off
   *  @return true if sourcedg is on
   *  @since NeXtMidas 2.5.0
   */
  public boolean getDgState() { return !dgdclosed; }
  /** Set the linearin flag to true dynamically */
  public void setLinear() { linearin=true; }
  /** Set the linearin flag to false, since this is ULAW */
  public void setULAW() { linearin=false; }
  /** Set number of packet skips to auto reset the fill on; zero (0) is off
   *  @param value number of packet skips to auto reset the fill on; zero (0) is off
   */
  public void setAutoSeqResetMAX (int value) {
    if (dgd==null) return;
    try {
      dgd.setAutoSeqResetMAX(value);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get the frame size of the output data file
   *  @return the frame size of the output data file
   */
  public int getFrameSize () { return pout.getFrameSize(); }

  /** Set the frame size of the output data file
   *  @param value   The value to be set
   *  @param setpipe true if to set the value in the pipe
   */
  public void setFrameSize (int value,boolean setpipe) {
    framesz=value;
    if (setpipe && pout!=null) {
      pout.setFrameSize(value);
      pout.update();
    }
  }
  /** Calls {@link #setFrameSize(int, boolean)} to set the frame size of the output data file
   *  This version always sets the value in the pipe
   *  @param value The value to be set
   */
  public void setFrameSize (int value) { setFrameSize(value,true); if (fill) chdr = -1; }

  /** Get YDelta of output file
   * @return Y delta of the output data file
   */
  public double getYDelta () { return pout.getYDelta(); }

  /** Set the Y delta of the output data file. Also, updates the header.
   *  @param value The value of ydelta
   */
  public void setYDelta (double value) {
    pout.setYDelta(value);
    pout.update();
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Reads a packet This is called by {@link #process()}.
   *  @return true if a packet was read, false otherwise.
   */
  public boolean read () {
    try {
       if (dgd==null) return false;
       len=dgd.read(buf,offsetdata);
       if (len<=0) return true;
       curlen=len;
       offset=0;
       if (fill) {
         fillnum=dgd.getFillNum(); //fillnum just tells us the count of missing packets - no header calculations
         if (!icehdrframe && framesz<=0) { //framesz<=0 for 1000 file with fill on
           if (verbose) System.out.println("FILL:: 1000 file fill="+fillnum);
           if (fillnum>0) fillbuf(fbuf,len,fillnum);
           return true;
         }
         if (verbose) System.out.println("FILL:: Packet:"+dgd.getCurHdrSeq()+":"+dgd.getSeq()+":"+len+",fn="+fillnum);

         if (chdr<0) { //initial startup cases
           if (!dgd.isIceHdrPacket()) {
             if (verbose) System.out.println("FILL:: Skipping startup packets until header");
             return true; //startup and need to skip until packet with a header
           }
           chdr=dgd.getCurHdrSeq();
           if (verbose) System.out.println("FILL:: Startup packet:"+chdr);
           if (icehdrframe) {
             if (dgd.getBytes()>0) {
               //currently does not check all headers - only startup -should do all the time
               if (getIceHdrInPipe()) setFrameSize((dgd.getBytes()+ICEPacket.headerLength)/Data.getBPA(midasdatatype),true); //Since 3.5.4 Bug 2921
             else setFrameSize(dgd.getBytes(), true);
             }
             if (verbose) System.out.println("FILL:: Set Frame Size="+framesz);
           }
           pktfrcount=0; //gets incremented on write
           fillnum=0; //no fill needed on startup with hdr
           dgramTime(dgd.getICETime(),dgd.getDU2(),dgd.getDU1());
         }

         if (fillnum>0) { //know we skipped packets before this one when get here
           if (pktfrcount>=framesz) { //filled last packet so either skip or reset
             if (dgd.isIceHdrPacket()) { //current packet is header case
               pktfrcount=0; //gets incremented on write
               dgramTime(dgd.getICETime(),dgd.getDU2(),dgd.getDU1());
             } else { //ignore this packet
               if (verbose) System.out.println("FILL:: Skip packet, not header and previous frame filled fn="+fillnum+
                                               " pktfrcount="+pktfrcount+" framesz="+framesz+" len="+len);
               return true;
             }
           } else { //missing frame data
             if (dgd.isIceHdrPacket()) { //need to fill buffer as current is header frame
               fillbuf(fbuf,framesz-pktfrcount);
               if (verbose) System.out.println("FILL:: Header case framesz="+framesz+" pktfrcount="+pktfrcount+",fn="+fillnum);
               pktfrcount=0;
               dgramTime(dgd.getICETime(),dgd.getDU2(),dgd.getDU1());
             } else { //not header frame
               if (icehdrinpipe) pktsinframe=(framesz-ICEPacket.headerLength)/len;  // Since 3.5.4 Bug 2921
               else pktsinframe=framesz/len;
               chdr=dgd.getCurHdrSeq();
               seq=dgd.getSeq();
               pcount=seq-chdr;
               if (pcount<0) {
                 if (verbose) System.out.println("FILL:: rollover or prev case seq="+seq+",hdr="+chdr);
                 if (pcount<=-100) pcount=pcount+65536; //likely case rollover, else very strange case
                 else {
                   if (verbose) System.out.println("FILL:: Out of order case seq="+seq+",hdr="+chdr);
                   return true;
                 }
               }
               if (pcount>pktsinframe) {
                 fillbuf(fbuf,framesz-pktfrcount);
                 if (verbose) System.out.println("FILL:: Skip Cross frame case pktsinframe="+pktsinframe+" fn="+fillnum+
                                                 " pcount="+pcount+" seq="+seq+" chdr="+chdr+",fill="+(framesz-pktfrcount));
                 pktfrcount=framesz;
                 return true;
               } else {
                 pktfrcount+=fillbuf(fbuf,fillnum*len);
                 if (verbose) System.out.println("FILL:: Within frame case pktsinframe="+pktsinframe+" fn="+fillnum);
               }
             }
           }
         } else { //when fillnum<0 by more than some frames, restart
           if (dgd.isIceHdrPacket()) {
             if (verbose && pktfrcount<framesz && pktfrcount!=0) System.out.println("FILL:: ERROR - frame not filled in framesz="+framesz+","+pktfrcount);
             pktfrcount=0;   //reset on packetheader
             dgramTime(dgd.getICETime(),dgd.getDU2(),dgd.getDU1());
           } else if (pktfrcount>=framesz) {
             if (verbose) System.out.println("FILL:: Skip packet, looking for header and previous frame filled");
             return true;
           }
         }
       }

       if (len>0) {
         if (msgonly) { //no pipe case
           String tmpbuf = new String(buf,0,len);
           if (verbose) System.out.println("XML In="+tmpbuf);
           Table XMLTab = XmlFile.fastXmlToTable(tmpbuf, null, null, -1, true);
           sendMessage((MessageHandler)msgid,new Message(msgnameforxml,1,XMLTab));
         } else {
//         if (pout.avail()<len) {
//           M.info("sourcedg: Pipe full - dumping data. Len="+len+". Avail="+pout.avail());
//           double p = pout.avail();
//           len=(int)p;
//         }
           if (pout==null) {
             M.info("sourcedg: No output pipe.");
             return true;
           }
           if (pout.io==null) {
             M.error("sourcedg: Pipe open failed");
             return true;
           }
           if (midasfiletype==2000 || midasfiletype==1000) {
             if (verbose) System.out.println("  Writing len="+len);
             pout.write(buf,0,len);
             if (fill) pktfrcount+=len;
             if (pktfrcount>framesz) {
               if (verbose) System.out.println("ERROR IN CALCULATION="+framesz+">pkt="+pktfrcount);
               pktfrcount=framesz;
             }
             if (convout!=null && convout.io!=null && intl>0) {
               numrecs=(long)(len/intl);
               psize=len;
               if (numrecs<1) numrecs=1;
               if (len>intl) psize=intl;
               offset=0;

               if (convout.avail()<(len+len)) M.info("sourcedg: Convpipe full: len="+len+", avail="+convout.avail());
               else {
                  if (!linear && !linearin) {
                   for (j=0;j<numrecs;j++) {
                     outbuf=outdata.castI(false);
                     for (i=0;i<psize; i++) outbuf[i]=decodeTable[buf[offset+i]+128];
                     outdata.uncast(outbuf,true);
                     convout.write(outdata,psize);
                     offset+=psize;
                   }
                 } else convout.write(buf,0,len);
               }
             }
           } else if (midasfiletype==3000) {
           //Skip the first 8 bytes
             if (xml3000out) {
               String tmpbuf = new String(buf,8,buf.length-8);
               //Only handles XML in single string
               if (verbose) System.out.println("SDG: XML '"+tmpbuf+"'");
               Table XMLTab = XmlFile.fastXmlToTable(tmpbuf, null, null, -1, true);
               Table outtab = matchTableInfo(t3kstruct,XMLTab);
               if (verbose) {
                 System.out.println("SDG: XML as a Table "+XMLTab);
                 System.out.println("SDG: OUTPUT Table "+outtab);
               }
               //XML data must match type 3000 file structure
               pout.setData(1,outtab);
             } else pout.write(buf,0,len);
           }
         }
       }
       return true;
     } catch (Exception e) {//receive can throw IO, SocketTimeOut,PortUnreachable and IllegalBlockingMode Exceptions
        e.printStackTrace();
        M.warning("sourcedg: Read failed. "+e);
     }
     return false;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Calls {@link #fillbuf(byte[], int)} to write out data buffer to output DataFile
   *  @param fbuf    the buffer containing the data to write out
   *  @param len     the length of the data buffer information
   *  @param fillnum the count of missing packets ??
   *  @return the number of bytes written out
   */
  public int fillbuf(byte[] fbuf,int len,int fillnum) {
    if (fillnum<=0 || len<1) return 0;
    return fillbuf(fbuf,len*fillnum);
  }
  /** Write out data buffer to output DataFile
   *  @param fbuf     the buffer containing the data to write out
   *  @param totbytes the number of bytes to write out
   *  @return the number of bytes written out
   */
  public int fillbuf(byte[] fbuf,int totbytes) {
    if (verbose) System.out.println("  Filling totbytes="+totbytes);
    if (pout==null) return -1;
    if (totbytes<1) return 0;
    if (totbytes<=fbuf.length) {
      pout.write(fbuf,0,totbytes);
      return totbytes;
    }
    fullwrites=totbytes/fbuf.length;
    for (i=0;i<fullwrites;i++) {
      pout.write(fbuf,0,fbuf.length);
    }
    remain=totbytes-(fullwrites*fbuf.length);
    if (remain>0){
      pout.write(fbuf,0,remain);
    }
    return totbytes;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Sends a message using a Message Handler
   *  @param msgid Message Handler in which to send the message
   *  @param msg the message to be sent
   */
  public void sendMessage (MessageHandler msgid, Message msg) {
//    if (msgid instanceof MessageHandler) { // NTN: since 3.1.1: this check is NOT needed
      if (msgid == msg.from) return;    // don't send it back
      msg.to = msgid;
      msg.from = this;
      msgid.processMessage(msg);
//    }
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  private Table matchTableInfo (Table structure, Table inTable) {
    String keyvals = structure.getKeyList();
    StringTokenizer keyST = new StringTokenizer(keyvals,",",false);
    Table retTab = new Table();

    while (keyST.hasMoreTokens()) {
      String structKey = keyST.nextToken();
      if (structKey.length()>4) structKey = structKey.substring(0,4);
      String fmt = structure.getS(structKey);
      String tblVal = inTable.getS(structKey);
      if (tblVal != null) {
        if (fmt.equals("SD")) retTab.put(structKey,Convert.s2d(tblVal));
        else if (fmt.equals("SL")) retTab.put(structKey,Convert.s2l(tblVal));
        //else if (fmt.equals("SB"))
        else retTab.put(structKey,tblVal);
      }
    }
    return retTab;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  private short[] build_mu_decode_table () {
    short[] convbuf = new short[256];

    int i,j,x,S,L,V;
    for (i=0; i<256; i++) {
      j = i-128;
      x = j^127;
      S = 1-(2*((x&128)/128));
      L = (x&112)/16;
      V = x&15;
      convbuf[i] = (short) Math.round(((S*(Math.pow(2,L)*(2*V+33)-33)*32767.0)/8031.0));
    }
    return convbuf;
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  public void dgramTime (double jtime, double inxStart, double inxDelta){
    try {
//      if (packetDebug) {
      if (verbose) {
        dgTime.fromJ1950(jtime);
        System.out.println("sourcedg: dgramTime = "+dgTime+" XS="+inxStart+" XD="+inxDelta);
      }

      pout.setTimeAt(jtime);
      if (convout!=null) convout.setTimeAt(jtime);
      dgjtime=jtime;
      dgdxs=inxStart;
      dgdxd=inxDelta;
      dgdsize=dgd.getSize();
      dgdbytes=dgd.getBytes();
      dgdformat=dgd.getFormat();

      if ((icehdrframe || doXSXD) && (inxStart!=getXStart() || inxDelta!=getXDelta())) {
        if (verbose) M.info("Update message XS="+inxStart+"("+getXStart()+") XD="+inxDelta+"("+getXDelta()+")");
        setXStart(inxStart);
        setXDelta(inxDelta);
        if (icehdrframe && msgid instanceof MessageHandler) {
          Table XMLTab = new Table();
          XMLTab.put("TYPE","XSXDUPD");
          XMLTab.put("XD",inxDelta);
          XMLTab.put("XS",inxStart);
          XMLTab.put("SIZE",dgdsize);
          XMLTab.put("BYTES",dgdbytes);
          XMLTab.put("FORMAT",dgdformat);
          sendMessage((MessageHandler)msgid,new Message(msgnameforxml,1,XMLTab));
        }
        if (doXSXD) {
          if (verbose) M.info("sourcedg: Restarting with xStart="+inxStart+" xDelta="+inxDelta);
          restartSourcedg(jtime);
        }
      }
    } catch (Exception e) {
      e.printStackTrace();
      M.warning("sourcedg: checkIceParms failed. "+e);
    }
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Restart Sourcedg */
  public void restartSourcedg() { setState(RESTART); }
  private void restartSourcedg(double jtime) {
    setYStartTime(jtime);
    restartSourcedg();
  }
}
