package nxm.net.prim;

import java.util.Arrays;
import java.util.StringTokenizer;
import java.util.logging.Level;

import nxm.net.lib.DatagramData;
import nxm.net.lib.DatagramData.IceTimeCallBack;
import nxm.net.lib.RTPICEPacketsCache;
import nxm.sys.inc.InternalUseOnly;
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.LogUtilities;
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.TimeLine;
import nxm.sys.lib.XmlFile;
import nxm.sys.libm.Tolerance;

/**
  sourcedg - uses DatagramData class to reads (Raw or RTP) from Multicast or unicast datagram stream

  Handling of out-of-order packets was added in NeXtMidas 3.8.1
  <br>
  <b>Note: Handling of out-of-order is not going to work without valid ICE Headers</b>

  @author MSZ
  @since 3.7.2 moved to NET from SYS for NXM-3048
*/
public class sourcedg extends Primitive implements IceTimeCallBack {
  //may be multicast or unicast and may or may not be rtp or sap - SB in only
  private TextFile logfile;
  /** pout is normal output */
  private DataFile pout;
  /** convout is for audio streams - converted from ULAW to Lin16 */
  private DataFile convout;
  /** holds convout data */
  private Data outdata;
  /** DatagramData instance */
  private DatagramData dgd = null;
  private static final Class<?> THIS_CLASS = sourcedg.class;
  private boolean abort=false, dgdclosed=true, linear=true, linearin=true, init=false;

  private short[] outbuf;
  private short[] decodeTable;
  private Table t3kstruct = null;
  /** buffer that is populated from DatagramData */
  private byte[] buf;
  /** fill buffer */
  private byte[] fbuf = new byte[1024];

  private static final int MAXPKTBUF=32768;
  /** default host */
  private static final String defaulthost="127.0.0.1";
  /** default message name */
  private static final String 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;
  private String midasdatatype="SB";
  private int port=defaultport,timeOut=defaulttimeout;
  /** input transfer length */
  private int intl=defaultintl;
  /** output transfer length */
  private int outtl=intl*2;
  /** Midas file type */
  private int midasfiletype=1000;
  /** timeline length */
  private int tll=defaulttll;
  /** offset of where to start reading data in packets (within the data area?) */
  private int offsetdata=0;
  /** the XML main name sent to processMessage */
  private String msgnameforxml=defaultxmlmsgname;
  /** where to send messages, from /MSGID= */
  private Object msgid=null;
  /** filter on specific rtp type - used to pull a "channel" out of stream based on type */
  private int rtpfiltertype=-1;
  /** msgid's name */
  private String msgidname=null;

  /** curlen is last >0 length */
  private int curlen=0;
  /** bytes written to current frame so far */
  private int pktfrcount=0;
  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;
  /** length of data in current packet in bytes ?? */
  private int len;
  private int offset;
  /** number of packets worth of data to fill (because they are missing) */
  private int fillnum;
  /** number of packets needed to get full frame worth of data */
  private int pktsinframe;
  /** sequence number from Octets 2 & 2 or RTP packet header */
  private int seq;
  /** packets in this frame - based on current sequence number and sequence number of last header */
  private int pcount;
  private int psize,i,j;
  /** value used to fill missing buffers */
  private byte fillValue = 0;
  /** frame size */
  private int framesz=0;
  private double ystartjtime=0;
  long numrecs;

  /** debugging messages */
  private boolean verbose=false;
  /** Datagram data prints out detailed packet info */
  private boolean packetDebug=false;
  /** RTPICEPacketsCache data prints out detailed packet/cache info */
  private boolean reorderingDebug=false;
  /** rtp packets */
  private boolean rtp=true;
  /** sap type packets */
  private boolean sap=false;
  /** set to a port and any address */
  private boolean alladdr=false;
  /** converts XML input into 3000 file output - tags must match 3000 file columns */
  private boolean xml3000out=true;
  /** include icehdr in data written out */
  private boolean icehdrinpipe=false;
  /** set framesize from ice header data */
  private boolean icehdrframe=false;
  /** fill requires constant packet sizes, doesn't work with variable data packets */
  private boolean fill=false;
  /** used to output only Tables converted from XML to processMessage loop */
  private boolean msgonly=false;
  /** Indicates if time-code processing is being performed */
  private boolean doTimeCode = false;
  /** Used with doTimeCode to enable updates of XS and XD */
  private boolean doXSXD     = true;

  /** 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;
  /** DatagramData size of packet data in elements */
  private int dgdsize=0;
  /** DatagramData size of packet data in bytes */
  private int dgdbytes=0;
  /** DatagramData current Midas digraph format, byte mode,byte type */
  private String dgdformat=null;
  private int logFactor = -1;
  /** For debugging, converts time of the datagram to readable form. */
  private Time dgTime = new Time();

  /////////////////////////////////////////////////////////////////////////////////////////////////////
  // The following were added in NeXtMidas 3.8.1 for handling out of order packets
  /** Handling out-of-order RTP Packets (that have an ICE Packet Header for each frame of data) */
  private boolean handlingOutOfOrder = false;
  /** Cache that handles reordering of packets (that havee an ICE Header for each frame) */
  private RTPICEPacketsCache iceCache = null;
  // --- initialization and inittialization status fields ----
  /** True if we have done first ice header, now we need second to calculate Y-Delta */
  private boolean haveFirstIceHeader = false;
  /** True if have both the frame size and yDelta information for header */
  private boolean haveFrameSzAndYDeltaForCache = false;
  /** YDelta calculations require 2 ICE Packets - this holds the first */
  private ICEPacket initialICEPacketForYDelta = null;
  //--- ICEPacketCache operation specific fields ----
  /** Determine and use the data per packet within a frame pattern */
  private boolean usePacketSizePattern = false;
  /** The max amount of missing frames before dropping (i.e. filling) */
  private int maxFramesMissing;
  //--- restart related fields ----
  /** Tolerance on XStart changes before a restart is needed */
  private double xStartToleranceRestart;
  /** Tolerance on XDelta changes before a restart is needed */
  private double xDeltaToleranceRestart;
  /** Tolerance on YDelta changes before a restart is needed */
  private double yDeltaToleranceRestart;
  /** local yStart value used during restarts */
  private Time   yStart = null;
  /** local yDelta value used during restarts */
  private double yDelta = 0;

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

  /** Logging level defaulted to Level.FINEST */
  private static Level logLevel= Level.FINEST;
  private static Level logLevelSysout = Level.INFO;

///////////////////////////////////////////////////////////////////////////////
  /** 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();
    }
  }

  /** Set reordering related debug
   *  @param debug true if doing reordering debug
   */
  @InternalUseOnly
  public void setReorderingDebug (boolean debug) {
    this.reorderingDebug = debug;
  }

  /** Get if reading RTP 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 -1 if 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
   *  @return host
   */
  public String getHost () { return host; }

  /** Set host
   *  @param ptemp host
   */
  public void setHost (String ptemp) {
    host=ptemp;
    if (dgd==null) return;
    try {
      dgd.setHost(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get port
   *  @return port
   */
  public int getPort () { return port; }

  /** Set port
   *  @param ptemp port number
   */
  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();
    }
  }


  /** Get if is listening to a specified port and any address (host)
   *  @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 offset of data in buffer
   *  @return offset of data in buffer
   */
  public int getOffset () { return offsetdata; }

  /** Set offset of data in buffer
   *  @param ptemp offset of data in buffer
   */
  public void setOffset (int ptemp) {
    if (ptemp<0 || ptemp>MAXPKTBUF) offsetdata=0;
    else offsetdata=ptemp;
  }

  /** Get is converting XML input into 3000 file output - tags must match 3000 file columns
   *  @return is converting XML input into 3000 file output
   */
  public boolean getXML3000 () { return xml3000out; }

  /** Set is converting XML input into 3000 file output - tags must match 3000 file columns
   *  @param ptemp true if converting XML input into 3000 file output
   */
  public void setXML3000 (boolean ptemp) { xml3000out=ptemp; }

  /** Get TimeLine length
   *  @return timeline length (-1 if grow as needed)
   */
  public int getTimeLineLength () { return tll; }

  /** Set TimeLine length
   *  @param ptemp timeline length (-1 if grow as needed)
   */
  public void setTimeLineLength (int ptemp) {
    if (ptemp<0) tll=defaulttll;
    else tll=ptemp;
  }

  /** Get if setting framesize from ice header data
   *  Note: This value is not used with /OUTOFORDER, as it  always sets
   *  framesize from ice header data
   *  @return true if if setting framesize from ice header data
   */
  public boolean getIceHdrFrame() { return icehdrframe; }

  /** Set if setting framesize from ice header data
   *  Note: This value is not used with /OUTOFORDER, as it  always sets
   *  framesize from ice header data
   *  @param ptemp true if setting framesize from ice header data
   */
  public void setIceHdrFrame (boolean ptemp) { icehdrframe=ptemp; }

  /** Get if including ICE Header in pipe
   *  @return true if including ICE Header in pipe
   */
  public boolean getIceHdrInPipe () { return icehdrinpipe; }

  /** Set if including ICE Header in pipe
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER
   *  @param ptemp true if including ICE Header in pipe
   */
  public void setIceHdrInPipe (boolean ptemp) {
    icehdrinpipe=ptemp;
    if (dgd==null) return;
    try {
      dgd.setIceHdrInPipe(ptemp);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /** Get value to use when filling in missing data
   *  @return value to use when filling in missing data
   */
  public byte getFillValue() { return fillValue; }

  /** Set value to use when filling in missing data
   *  @param filler value to use when filling in missing data
   */
  public void setFillValue(byte filler) {
    fillValue = filler;
  }

  /** Get if we are filling in missing data and can assume that all packets
   *  have an equal amount of data
   *  @return if we are filling in missing data and can assume that all packets
   *          have an equal amount of data
   */
  public boolean getFill () { return fill; }

  /** Set if we are filling in missing data and can assume that all packets
   *  have an equal amount of data
   *  @param ptemp true if we are filling in missing data and can assume that all
   *               packets have an equal amount of data
   */
  public void setFill (boolean ptemp) {
    fill=ptemp;
    if (fill) {
      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();
      }
    }
  }

  /** Fill fill buffer with value provided
   *  @param val value to use in fill buffer
   */
  public void setFillBufValue (byte val) { Arrays.fill(fbuf,val); }

  /** Set fill buffer to buffer provided
   *  @param pfbuf buffer to use as fill buffer
   *  @return length fill buffer
   */
  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 message id name
   */
  public String getMSGID() { return msgidname; }

  /** Set message id name
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER
   *  @param ptemp 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);
    }
  }

  /** Get if outputting only Tables converted from XML
   *  @return if outputting only Tables converted from XML
   */
  public boolean getMsgOnly() { return msgonly; }

  /** Set if outputting only Tables converted from XML
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER
   *  @param ptemp true if outputting only Tables converted from XML
   */
  public void setMsgOnly(boolean ptemp) { msgonly=ptemp; }

  /** Get if time-code processing is being performed
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER, as that
   *  has separate Timecode processing
   *  @return true if time-code processing is being performed
   */
  public boolean getTimeCode() { return doTimeCode; }

  /** Set if time-code processing is being performed
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER, as that
   *  has separate Timecode processing
   *  @param ptemp true if time-code processing is being performed
   */
  public void setTimeCode(boolean ptemp) { doTimeCode=ptemp; }

  /** Get if updating XStart XDelta (along with Timecode)
   *  @return true if updating XStart XDelta
   */
  public boolean getXSXD() { return doXSXD; }

  /** Set if updating XStart XDelta (along with Timecode)
   *  Note: This SHOULD NOT be set if handling /OUTOFORDER, as that
   *  has separate XStart XDelta handling
   *  @param ptemp true if updating XStart XDelta
   */
  public void setXSXD(boolean ptemp) { doXSXD=ptemp; } //Restart pipe on XS,XD changes?

  /** 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 start time
   *  @return the X-axis time code start time
   */
  public double getXStart() { return xStart; }
  /** Set the X-axis start time
   *  @param ptemp the X-axis time code start time
   */
  public void setXStart(double ptemp) { xStart=ptemp; }
  /** Get the X-axis delta
   *  @return the X=axis delta
   */
  public double getXDelta() { return xDelta; }
  /** Set the X-axis delta
   *  @param ptemp the X-axis 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 as a double
   */
  public void setYStartTime(double ptemp) { ystartjtime=ptemp; }
  /** Set the Y-axis start time
   *  @param time the Y-axis start time
   */
  public void setYStart(Time time) { yStart=time; }
  /** 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; }  // since 3.0.0
  /** Get Datagram packet X-axis delta
   *  @return the Datagram packet X-axis delta
   *  @since NeXtMidas 3.0.0
   */
  public double getPacketXDelta() { return dgdxd; }  // since 3.0.0
  /** Get Datagram packet julian time
   *  @return the Datagrapm packet julian time
   *  @since NeXtMidas 3.0.0
   */
  public double getPacketJtime() { return dgjtime; } // since 3.0.0

  /** 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 DatagramData();
        if (dgd==null) { M.error("sourcedg: Could not open socket"); return FINISH; }

        // Logging and debug
        if (MA.find("/VERBOSE")) setVerbose(true);
        if (MA.find("/PACKETS")) setPacketDebug(true);

        if (MA.find("/ALLHOSTS")) setAllHosts(true);
        intl = MA.getL("/INTL",defaultintl);
        if (intl<=1) intl=defaultintl;
        setOffset(MA.getL("/OFFSET",0));
        setTimeLineLength(MA.getL("/TLL",defaulttll));  //TimeLine Length
        setTimeOut(MA.getL("/TIMEOUT",defaulttimeout));

        midasfiletype = MA.getL("/FTYPE",1000);

        handlingOutOfOrder = MA.find("/OUTOFORDER");
        if (handlingOutOfOrder && (midasfiletype != 2000)) {
          M.warning("sourcedg switch /OUTOFORDER is currently only compatible with File Type 2000 - /FTYPE="+midasfiletype);
        }

        if (!handlingOutOfOrder) {
          // Switches not used or handled separately for handling out-of-order code
          setAutoSeqResetMAX(MA.getL("/AUTOFILLRESET",0)); //to automatically reset fill stuff on certain counts
          if (MA.find("/FILL")) { setFill(true); setFillValue(MA.getB("/FILLVALUE", (byte)0));}
          setFrameSize(MA.getL("/FRAMESIZE",1024),false);
          if (MA.find("/ICEHDRFRAME")) setIceHdrFrame(true);
          if (MA.find("/ICEHDRINPIPE")) setIceHdrInPipe(true);
          setYStartTime(MA.getD("/INITIALTIME",-1.0));    //Initial ystart time (-1=Do not set, use qualifiers)
          if (MA.find("/MIDASSI")) setMidasDataType("SI");
          setMSGID(MA.getU("/MSGID"));
          setXMLMSGName(MA.getS("/MSGNAME",defaultxmlmsgname));
          if (MA.find("/MSGONLY")) setMsgOnly(true);  //wgd used XMLTBL but MSGONLY does same thing - XMLTBL XML Message always sent to MAIN
          if (MA.find("/NORMAL3000")) setXML3000(false);
          if (MA.find("/NORTP")) setRTP(false);
          if (MA.find("/NOXSXD")) setXSXD(false);
          setRTPFilter(MA.getL("/RTPFILTER",-1));
          if (MA.find("/SAP")) setSAP(true);
          if (MA.find("/TIMECODE")) setTimeCode(true);
          if (MA.find("/ULAW")) { linear=false; linearin=false; }
          setXDelta(MA.getD("/XDELTA",defaultxdelta));
          setXStart(MA.getD("/XSTART",defaultxstart));
        }
        else {
          // Reordering debug
          if (MA.find("/REORDERDEBUG")) setReorderingDebug(true);
          // Switches and values specific to handling out-of-order code (Since 3.8.1 for NXM-3223)
          dgd.setIceData(true); // since 3.8.1 need to cache the ICE data when handling out-of-order packets
          setFillValue(MA.getB("/FILLVALUE", RTPICEPacketsCache.FRAME_FILL_DEFAULT));
          setMaxFramesMissing(MA.getL("/MAXMISSINGFRAMES", RTPICEPacketsCache.DEFAULT_MAX_MISSING_FRAMES));
          setUsePacketSizePattern(MA.getState("/USEPACKETSIZEPATTERN", true) || MA.find("/FILL"));
          if (MA.find("/XSTART_RESTART_TOLERANCE")) setXStartToleranceRestart(MA.getD("/XSTART_RESTART_TOLERANCE", Tolerance.getTolerance()));
          if (MA.find("/XDELTA_RESTART_TOLERANCE")) setXDeltaToleranceRestart(MA.getD("/XDELTA_RESTART_TOLERANCE", Tolerance.getTolerance()));
          if (MA.find("/YDELTA_RESTART_TOLERANCE")) setYDeltaToleranceRestart(MA.getD("/YDELTA_RESTART_TOLERANCE", Tolerance.getTolerance()));
        }
        setLogFactor(MA.getI("/LOGFACTOR", (short)50));
        // 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 (state==RESTART && handlingOutOfOrder && haveFirstIceHeader) {
        setPoutHeaderData(iceCache.getIcePacketRestart());
      }
      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 (yStart != null) { // set high precision if available 3.8.1
        pout.setTimeAt(yStart);
        pout.setTime(yStart);

      }
      else if (ystartjtime>=0) {
        pout.setTimeAt(ystartjtime);
      }
      if (yDelta != 0) {  // set yDelta from restart information
        pout.setYDelta(yDelta);
      }
      if (doTimeCode || handlingOutOfOrder){ // 3.8.1 assuming handlingOutOfOrder means type 2000 for now
        if (doXSXD || handlingOutOfOrder) { pout.setXUnits(nxm.sys.inc.Units.FREQUENCY); }
        pout.setYUnits(nxm.sys.inc.Units.TIME);
        if (!handlingOutOfOrder) {
          dgd.registerTimeCB(this);
        }
        else if (doTimeCode && handlingOutOfOrder ){
          LogUtilities.log(THIS_CLASS, Level.INFO,"Currently not supporting call-back with out-of-order");
        }
      }
      if (!handlingOutOfOrder || haveFrameSzAndYDeltaForCache) 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 {
      LogUtilities.log(THIS_CLASS, logLevel, "Closing");
      if (dgd!=null) { dgd.close(); } //doesn't clear seq numbers
      FillPipeAtEnd();
      if (pout!=null && pout.isOpen) { pout.update(); pout.close(); } // added check to see if it is open because you can't update and close something that is not open
      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();
    }
  }

  /** Set Frame Size
   *  @param value frame size
   */
  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) {
    yDelta = value;  // local yDelta used during restarts
    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); // len number of bytes of actual data OR that plus the ICE Header length (icehdrinpipe=true)
      if (len <= 0) return true;
      curlen = len;
      offset = 0;
      if (handlingOutOfOrder) {
        // Do the handling of out of order data processing
        handleOutOfOrder();
        // Do not do anymore processing
        return true;
      }
      else if (fill) {
        boolean notDoingAnyMoreProcessing = handleFill ();
        if (notDoingAnyMoreProcessing) return true;
      }
      if (len>0) {
        if (msgonly) { //no pipe case
          String tmpbuf = new String(buf,0,len);
          LogUtilities.log(THIS_CLASS, logLevel, "XML In={0}", tmpbuf);
          Table XMLTab = XmlFile.fastXmlToTable(tmpbuf, null, null, -1, true);
          sendMessage((MessageHandler)msgid,new Message(msgnameforxml,1,XMLTab));
        } else {
          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) {
            LogUtilities.log(THIS_CLASS, logLevel,"  Writing len={0}",len);
            pout.write(buf,0,len);
            if (fill) pktfrcount+=len;
            if (pktfrcount>framesz) {
              LogUtilities.log(THIS_CLASS, logLevel,"ERROR IN CALCULATION={0}>pkt={1}",framesz, 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)  LogUtilities.log(THIS_CLASS, logLevelSysout,"SDG: XML '{0}'",tmpbuf);
              Table XMLTab = XmlFile.fastXmlToTable(tmpbuf, null, null, -1, true);
              Table outtab = matchTableInfo(t3kstruct,XMLTab);
              if (verbose) {
                LogUtilities.log(THIS_CLASS, logLevel,"SDG: XML as a Table {0}",XMLTab);
                LogUtilities.log(THIS_CLASS, logLevel,"SDG: OUTPUT Table {0}",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;
  }

  /** Process data including the handling of out-of-order packets
   *  @since NeXtMidas 3.8.1 for NXM-3223
   */
  private void handleOutOfOrder () {
    try {
      if (!haveFirstIceHeader ||
          (!haveFrameSzAndYDeltaForCache && (dgd.isValidIceHdrPacket() &&
          !RTPICEPacketsCache.isValidICEPacketPairForYDelta(dgd.getICE(), initialICEPacketForYDelta)))) {
        // either we do not have an ice header packet yet this stream or we are just starting up and
        // we received another ice header with either the same count as the first or one that is
        // not a nearby frame (restart and immediately restart again scenario)
        if (haveFirstIceHeader && (initialICEPacketForYDelta.getUnsignedCount() != dgd.getICE().getUnsignedCount())) {
          // We are here because the ICE Packet we received is not valid for use with the previous ICE header
          // to calculate a yDelta.  If this is anything other than the ICE Counts being the same (likely a stream
          // change immediately followed by a stream change), then is could be a more serious condition
          // with invalid ICE Headers
          M.warning("Initial ICE Packet Pair can not be used for yDelta calculation, restarting with latest ICEPacket\n"+
                    "Current ICE Packet: "+dgd.getICE()+"\nPrevious ICE Packet: "+initialICEPacketForYDelta);
        }
        if (dgd.isValidIceHdrPacket()) { // This is the first ice header, lets set up the header info.
          haveFirstIceHeader = true;
          iceCache = iceCache == null ? new RTPICEPacketsCache(dgd, dgd.getICE().getBytes(), pout) : iceCache;
          initializeRTPICEPacketsCache();
        }
        // create basic iceCache if it hasn't been created yet.
        iceCache = iceCache == null ? new RTPICEPacketsCache(dgd, pout) : iceCache;
        iceCache.setMissingFillValue(fillValue);
        iceCache.addPacketToCacheAndDropOld(buf, len);
        if (reorderingDebug) RTPICEPacketsCache.setDebugLevel(Level.INFO);
        // Do not do anything else, return here
        return;
      }
      else if (!haveFrameSzAndYDeltaForCache) {
        // this is the only time we get a different pout due to restart
        if (pout != iceCache.getPipeOut()) {
          iceCache.setPipeOut(pout);
        }
        if (dgd.isValidIceHdrPacket()) {
          // we have our second ICE Header!! We can get yDelta!
          yDelta = RTPICEPacketsCache.calcYDelta(dgd.getICE(), initialICEPacketForYDelta, -1);
          pout.setYDelta(yDelta);
          pout.open(); // Now we are ready to open
          iceCache.setInitialYDelta(yDelta); // set the ydelta up in ICEPacektsCache
          // Setup is done, ready for normal operation on all packets
          haveFrameSzAndYDeltaForCache = true;
        }
        else {
          // we are still waiting for our second ICE Header, so just add this packet to the cache
          iceCache.addPacketToCacheAndDropOld(buf, len);
          // Do not do anything else, return here
          return;
        }
      }
      // add packet and try to write frame if it is ready, if it returns true, we need to do a restart.
      if (iceCache.addPacketAndOutputFullFrames(buf, len)) {
        // if it is a hard restart, we are blowing up iceCache and restarting from scratch with the current packet (it is
        // an ICEHeader)
        // in cases where it does not need to blow up iceCache, it will skip this part
        if (iceCache.isHardRestart()) {
          // Create a new RTPICEPacketsCache as it is a complete restart
          iceCache = new RTPICEPacketsCache(dgd, dgd.getICE().getBytes(), pout);
  	// Call RTPICEPacketsCache setup
          initializeRTPICEPacketsCache();
          iceCache.setMissingFillValue(fillValue);
          iceCache.setIcePacketRestart(dgd.getICE());
          if (reorderingDebug) RTPICEPacketsCache.setDebugLevel(Level.INFO);
          iceCache.addPacketToCacheAndDropOld(buf, len);
        }
        restartSourcedg(iceCache.getIcePacketRestart());
        haveFrameSzAndYDeltaForCache = false;
      }
    }
    catch (Exception e) {
      String extraInfo = "sourcedg /OUTOFORDER code has run into an invalid condition (probably due to bad ICE Header data)\n" +
          "Sequence number that was being processed: "+dgd.getSeq()+"\n";
      if (dgd.getICE() != null) {
        extraInfo += "Sequence number of last ICE Header: "+ dgd.getLastValidIceHeader()+
            "\n"+dgd.getICE();
      }
      M.warnStackTrace(extraInfo, e);
      // try to recover by starting from scratch
      haveFirstIceHeader = false;
      haveFrameSzAndYDeltaForCache = false;
      restartSourcedg();
    }
  }

  /**
   * Initialize basic values for RTPICEPacketsCache
   */
  private void initializeRTPICEPacketsCache () {
    iceCache.setUsePacketSizePattern(usePacketSizePattern);
    if (xStartToleranceRestart > 0) iceCache.setStartToleranceRestart(xStartToleranceRestart);
    if (xDeltaToleranceRestart > 0) iceCache.setXDeltaToleranceRestart(xDeltaToleranceRestart);
    if (yDeltaToleranceRestart > 0) iceCache.setYDeltaToleranceRestart(yDeltaToleranceRestart);
    if (logFactor >= 0) iceCache.setLogFactor(logFactor);
    setHeaderInfo(dgd.getICE());
    iceCache.setMaxMissingFrames(maxFramesMissing);
    initialICEPacketForYDelta = dgd.getICE(); // save this ICE to use to calculate yDelta

  }
  /** Handle /FILL case (when not handling /OUTOFORDER
   *  This code was broken out from the read method in 3.8.1
   *  @return true if not doing any more processing
   */
  private boolean handleFill () {

    // fill = true means we are requiring constant packet sizes
    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
      LogUtilities.log(THIS_CLASS, logLevel,"sourcereorderdg.read FILL:: 1000 file fill={0}", fillnum);
      if (fillnum>0) fillbuf(fbuf,len,fillnum);
      return true;
    }
    LogUtilities.log(THIS_CLASS, logLevel, "sourcereorderdg.read FILL:: Packet:{0}:{1}:{2},fn={3}",
        dgd.getCurHdrSeq(), dgd.getSeq(), len, fillnum );

    if (chdr<0) { //initial startup cases
      if (!dgd.isIceHdrPacket()) {
        if (verbose) LogUtilities.log(THIS_CLASS, logLevelSysout,"FILL:: Skipping startup packets until header");
        return true; //startup and need to skip until packet with a header
      }
      chdr=dgd.getCurHdrSeq();
      LogUtilities.log(THIS_CLASS, logLevel,"sourcereorderdg.read FILL:: Startup packet:{0}", 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);
        }
        LogUtilities.log(THIS_CLASS, logLevel,"sourcereorderdg.read FILL:: Set Frame Size={0}",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
          LogUtilities.log(THIS_CLASS, logLevel,
              "sourcereorderdg.read FILL:: Skip packet, not header and previous frame filled fn={0} pktfrcount={1} framesz={2} len={3}",
              fillnum, pktfrcount, framesz, len);
          return true;
        }
      } else { //missing frame data
        if (dgd.isIceHdrPacket()) { //need to fill buffer as current is header frame
          fillbuf(fbuf,framesz-pktfrcount);
          LogUtilities.log(THIS_CLASS, logLevel,
              "sourcereorderdg.read FILL:: Header case framesz={0} pktfrcount={1},fn={2}",
              framesz, pktfrcount, 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) {
            LogUtilities.log(THIS_CLASS, logLevel, "sourcereorderdg.read FILL:: rollover or prev case seq={0},hdr={1}",
                seq, chdr );
            if (pcount<=-100) pcount=pcount+65536; //likely case rollover, else very strange case
            else {
              LogUtilities.log(THIS_CLASS, logLevel, "sourcereorderdg.read FILL:: Out of order case seq={0},hdr={1}",
                  seq, chdr);
              return true;
            }
          }
          if (pcount>pktsinframe) {
            fillbuf(fbuf,framesz-pktfrcount);
            LogUtilities.log(THIS_CLASS, logLevel,
                "sourcereorderdg.read FILL:: Skip Cross frame case pktsinframe={0} fn={1} pcount={2} seq={3} chdr={4},fill={5}",
                fillnum, pcount, seq, chdr, (framesz - pktfrcount));
            pktfrcount=framesz;
            return true;
          } else {
            pktfrcount+=fillbuf(fbuf,fillnum*len);
            LogUtilities.log(THIS_CLASS, logLevel, "sourcereorderdg.read FILL:: Within frame case pktsinframe={0} fn={1}",
                pktsinframe, fillnum);
          }
        }
      }
    } else { //when fillnum<0 by more than some frames, restart
      if (dgd.isIceHdrPacket()) {
        if (pktfrcount < framesz && pktfrcount != 0)
          LogUtilities.log(THIS_CLASS, logLevel, "sourcereorderdg.read FILL:: ERROR - frame not filled in framesz={0},{1}",
              framesz, pktfrcount);
        pktfrcount=0;   //reset on packetheader
        dgramTime(dgd.getICETime(),dgd.getDU2(),dgd.getDU1());
      } else if (pktfrcount>=framesz) {
        if (verbose) LogUtilities.log(THIS_CLASS, logLevel,"FILL:: Skip packet, looking for header and previous frame filled");
        return true;
      }
    }
    return false;
  }

//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Write out fill buffer
   * @param fbuf    fill buffer
   * @param len     length of data in current packet in bytes ??
   * @param fillnum number of packets worth of data to fill (because they are missing)
   * @return total bytes
   */
  public int fillbuf(byte[] fbuf,int len,int fillnum) {
    if (fillnum<=0 || len<1) return 0;
    return fillbuf(fbuf,len*fillnum);
  }

  /** Write out fill buffer
   * @param fbuf     fill buffe
   * @param totbytes total bytes to write
   * @return total bytes
   */
  public int fillbuf(byte[] fbuf,int totbytes) {
    if (verbose) LogUtilities.log(THIS_CLASS, logLevelSysout,"  Filling totbytes={0}", 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;
  }

//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Send Message
   * @param msgid message id
   * @param msg   message to send
   */
  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);
        LogUtilities.log(THIS_CLASS, logLevelSysout, "sourcedg: dgramTime = {0} XS={1} XD={2}",
            new Object[] { dgTime, inxStart, inxDelta });
      }

      pout.setTimeAt(new Time(jtime), TimeLine.NotifyInvalidEntry.WARNING);
      if (convout!=null) convout.setTimeAt(new Time(jtime), TimeLine.NotifyInvalidEntry.WARNING);
      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);
    }
  }
//////////////////////////////////////////////////////////////////////////////////////////////////////////
  /** Set sourcedg state to RESTART */
  public void restartSourcedg() {
    setState(RESTART);
  }
  /** Set sourcedg state to RESTART with given start time
   *  @param restart start time as double */
  private void restartSourcedg(double jtime) {
    setYStartTime(jtime);
    restartSourcedg();
  }


  /**
   * This will kick off a restart of sourcedg.
   * @param iceP The ice header to be used for the restart
   */
  private void restartSourcedg(ICEPacket iceP) {
    pout.erase(true);
    setFrameSize(iceP.getBytes()/iceP.getBPA(), false); // this one setFrameSize is fine since it only sets locally
    setXStart(iceP.getUserDataD(ICEPacket.XSTART_POSITION_IN_USER_DATA));
    setXDelta(iceP.getUserDataD(ICEPacket.XDELTA_POSITION_IN_USER_DATA));
    setYStart(iceP.getTC());
    haveFrameSzAndYDeltaForCache = false;
    restartSourcedg();
  }

  /** Set pout header based on ICE Packet Header
   *  @param iceP ICE Packet Header to use for pout header
   */
  private void setPoutHeaderData (ICEPacket iceP) {
    pout.setFrameSize(iceP.getSize()); // requires resetting rest of Y fields including units
    pout.setXStart(xStart);
    pout.setXDelta(xDelta);
    pout.setYStart(yStart.getSec());
    pout.setYUnits(nxm.sys.inc.Units.TIME);
    pout.setTime(iceP.getTC());
    pout.setHeadRep(iceP.getRep());
    pout.setFormat(iceP.getFormat());
  }

  /** Set sourcedg copy of header information from ICE Packet
   *  @param iceP ICE Packet Header to use for setting sourcedg header info
   *  @since 3.8.1 for NXM-3223
   */
  private void setHeaderInfo(ICEPacket iceP) {
    setFrameSize(iceP.getSize(), false); // this one setFrameSize is fine since it only sets locally
    setXStart(iceP.getUserDataD(ICEPacket.XSTART_POSITION_IN_USER_DATA));
    setXDelta(iceP.getUserDataD(ICEPacket.XDELTA_POSITION_IN_USER_DATA));
    setYStart(iceP.getTC());
    setMidasDataType(iceP.getFormat());
    setPoutHeaderData(iceP);
  }

  /** Set if code that handles out-of-order packets is to be used
   * @param handlingOutOfOrder true if sourcedg should use  the handling out-of-order code
   * @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setHandlingOutOfOrder (boolean handlingOutOfOrder) {
    this.handlingOutOfOrder = handlingOutOfOrder;
  }

  /** Set if the handling out-of-order code should discover the pattern for
   *  the amount of data in each packet within a frame and use this information
   *  to fill in missing data
   *  @param usePacketSizePattern true if packet size pattern is to be used
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setUsePacketSizePattern (boolean usePacketSizePattern) {
    this.usePacketSizePattern = usePacketSizePattern;
  }

  /** Set the max amount of missing frames before dropping (i.e. filling)
   *  some of the missing data and writing out old frames
   *  @param maxFramesMissing max amount of missing frames before assume data
   *                          is missing
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setMaxFramesMissing (int maxFramesMissing) {
    this.maxFramesMissing = maxFramesMissing;
  }

  /** Get restart tolerance on XStart
   *  @return the restart tolerance on XStart
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public double getxStartToleranceRestart () {
    return xStartToleranceRestart;
  }

  /** Set tolerance on XStart changes before triggering a restart
   *  Used in conjunction with RTPICEPacketsCache and /OUTOFORDER
   *  @param xStartToleranceRestart the restart tolerance on XStart
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setXStartToleranceRestart (double xStartToleranceRestart) {
    this.xStartToleranceRestart = xStartToleranceRestart;
  }

  /** Get restart tolerance on XDelta
   *  @return the restart tolerance on XDelta
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public double getXDeltaToleranceRestart () {
    return xDeltaToleranceRestart;
  }

  /** Set tolerance on XDelta changes before triggering a restart
   *  Used in conjunction with RTPICEPacketsCache and /OUTOFORDER
   *  @param xDeltaToleranceRestart the restart tolerance on XDelta
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setXDeltaToleranceRestart (double xDeltaToleranceRestart) {
    this.xDeltaToleranceRestart = xDeltaToleranceRestart;
  }

  /** Set tolerance on YDelta changes before triggering a restart
   *  Used in conjunction with RTPICEPacketsCache and /OUTOFORDER
   *  @param yDeltaToleranceRestart the restart tolerance on YDelta
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setYDeltaToleranceRestart (double yDeltaToleranceRestart) {
    this.yDeltaToleranceRestart = yDeltaToleranceRestart;
  }

  /** Set how often to log dropped packets - modulo value
   *  Used in conjunction with RTPICEPacketsCache and /OUTOFORDER
   *  @param logFactor how often to log dropped packets - modulo value
   *  @since 3.8.1 for NXM-3223
   */
  @InternalUseOnly
  public void setLogFactor (int logFactor) {
    this.logFactor = logFactor;
  }
}
