package nxm.sys.lib;

import nxm.sys.inc.Chainable;
import nxm.sys.inc.InternalUseOnly;
import nxm.sys.lib.CoreIO.IOOptions;

/** A class to define a Midas Pipe. <br>
    <br>
    Terms:
    <pre>
      Inlet    - The writer to the pipe (e.g. the {@link DataFile} WAVEFORM
                 uses for output).
      Outlet   - One of the readers from the pipe (e.g. the {@link DataFile}
                 PLOT uses for input).

      dfx[0]   - The current input for the writer. (special)
      dfc[0]   - The current input for the writer. This is what is returned by
                 {@link #getDataFile()} or getDataFile(0).

      dfx[n]   - The current output for reader number n.
      dfc[n]   - The current input  for reader number n. Although normally
                 Input directly maps to Output, during a pipe restart the
                 dfc[n] may point to the previous incarnation of the Inlet
                 until such a time that the reader has read all of the "old"
                 data from the pipe. This is what is returned by getDataFile(n).
    </pre>
    During a restart, this class persists, the inlet does not transition to 0
    when the writer closes. When the writer opens again, the new DataFile class
    is chained to the previous so readers can follow the cascaded pipes.<br>
    <br>
    During a restart, this class persists, the outlets chain to their previous
    incarnations to support look-back of time values, etc. When a {@link TimeLine}
    is used this also has implications for the restart. Although most "normal"
    primitives (e.g. NOOP, FFT) create a new {@link DataFile} instance for their
    input pipe when their <code>open()</code> is called in a restart situation, PLOT
    and RMIF are special in that they reuse their current {@link DataFile} instance
    (with the {@link DataFile#ALWAYSREOPEN} flag set) and simply trigger a manual
    reopen of the file. As a result, for PLOT and RMIF the "previous incarnations"
    are actually copies of the instance used.
    <pre>{@code
      TIME               NOOP/FFT                             PLOT/RMIF
       |               (as readers)                          (as readers)
       |          (standard/normal scenario)           (non-standard scenario)
       V       +------------+  +------------+       +------------+  +------------+
          dfx->| DataFile 0 |->| TimeLine 0 |  dfx->| DataFile 0 |->| TimeLine 0 |
               +------------+  +------------+       +------------+  +------------+

       ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ PIPE RESTART #1 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

               +------------+  +------------+       +------------+  +------------+
          dfx->| DataFile 1 |->| TimeLine 1 |  dfx->| DataFile 0 |->| TimeLine 0 |
               |            |  |            |       | (Original) |  |            |
               +------------+  +------------+       +------------+  +------------+
                 ^    |previous                       ^    |previous     ^
             next|    v                           next|    v             |
               +------------+  +------------+       +------------+       |
               | DataFile 0 |->| TimeLine 0 |       | DataFile 0 |-------+
               |            |  |            |       |  (Copy 1)  |
               +------------+  +------------+       +------------+

       ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ PIPE RESTART #2 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

               +------------+  +------------+       +------------+  +------------+
          dfx->| DataFile 2 |->| TimeLine 2 |  dfx->| DataFile 0 |->| TimeLine 0 |
               |            |  |            |       | (Original) |  |            |
               +------------+  +------------+       +------------+  +------------+
                 ^    |previous                       ^    |previous     ^  ^
             next|    v                           next|    v             |  |
               +------------+  +------------+       +------------+       |  |
               | DataFile 1 |->| TimeLine 1 |       | DataFile 0 |-------+  |
               |            |  |            |       |  (Copy 2)  |          |
               +------------+  +------------+       +------------+          |
                 ^    |previous                        ^   |previous        |
             next|    v                            next|   v                |
               +------------+  +------------+       +------------+          |
               | DataFile 0 |->| TimeLine 0 |       | DataFile 0 |----------+
               |            |  |            |       |  (Copy 1)  |
               +------------+  +------------+       +------------+
    }</pre>
    NOTE1: Since NeXtMidas 3.7.0, the previous DataFile links, are now truncated after maxHistory
           which defaults to DEFAULT_MAX_HISTORY (10) to prevent holding onto the
           resource/memory indefinitely during pipe restarts until the pipe is closed.<br>
    NOTE2: Prior to NeXtMidas 3.7.0, in the non-standard scenario (special case with
           how PLOT and RMIF handles Pipe restarts), the old DataFile Copies's next always
           pointed to the original DataFile (e.g. DataFile 0 (Copy 1)'s next points to (Original).
           This was corrected in NeXtMidas 3.7.0 to be consistent with the normal scenario and
           to allow for limiting the number of linked DataFiles.<br>
    <b>WARNING:</b> The read/write/avail/etc. methods are NOT synchronized for performance
    reasons. If reading or writing from multiple threads, caller MUST synchronize on this
    object. The attach/detach methods are synchronized since they generally are NOT called
    many times. <br>
    <br>
    <b>WARNING:</b> This class permits extension only for the purposes of supporting
    X-Midas inter-operability (see nxm.sys.libxm.XmPipe) and should not be
    extended within user-defined option trees.

    @author Jeff Schoen
 */
public class Pipe implements Chainable {
  /** Is specialized pipe debug enabled? */
  private static boolean DEBUG = false;

  /** Outlet number indicating that the pipe has detached. */
  public static final int DISCON = -2;

  /** default max number of outlets (pipe readers) */
  static final int DEFAULT_MAX_OUTLETS = 8;

  /**
   * Default max number of old "previous" linked DataFiles references to keep around (usually from pipe restarts).
   * This is merely a guess as to how far back to support for normal usage and for the case of PLOT/Layer2D raster
   * requiring time information that may be 2+ restarts ago. Since 3.7.0 as part of BUG 2991.
   */
  static final int DEFAULT_MAX_HISTORY = 10;

  // NOTE: A number of the field definitions are a bit "silly" from a pure-Java
  //       standpoint (e.g. inlet as int not boolean) but were intentionally
  //       chosen in the early days of NeXtMidas to mirror the definitions used
  //       in X-Midas (and the BLUE header). One advantage of this approach is
  //       that it simplified some of the work required to map X-Midas pipes
  //       into NeXtMidas (e.g. XmPipe for NeXtOpt).

  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getInlet()}. this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    int        inlet     = 0;       // is the inlet attached (1) or not (0)?
  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getOutlets()}. this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    int        outlets   = 0;       // number of currently attached readers
  /** @deprecated Since NeXtMidas 3.3.2, this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    int        outMask   = 0;       // bit mask of readers
  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getMaxOutlets()}. this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    int        maxOut    = 8;       // maximum readers (outlets)
  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getSize()}. this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    int        size      = 131072;  // current pipe size in bytes
  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getInByte()}. this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public volatile double inPtr   = 0;       // Inlet pointer in bytes (last write byte pointer SHOULD be >= outPtr[0]). added volatile in 3.5.1 for BUG 1511
  /** @deprecated Since NeXtMidas 3.3.2, use {@link #getOutByte(int)}this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    double[]    outPtr;             // Outlet pointers in bytes (for readers) (0=InletTotal, N=SingleOutlet)
  /** @deprecated Since NeXtMidas 3.3.2, this field should be private or package-private */ @Deprecated // TODO change in 3.7+
  public    byte[]      buffer;             // The data buffer
  /** Current Midas context (holds pipe references) */
  protected Midas      M;
  /** Pipe name. Use {@link #getName()} instead. */ // deprecated in 3.1.1 and became protected in 3.5.1
  protected String     name;
  /** The reference objects for the inlet/outlets (0=Inlet/writer, N=Outlet/reader). */
  protected Command[]  refs;
  /** Is a pipe restart needed? for the inlet/outlet (0=Inlet/writer, N=Outlet/reader) */
  protected boolean[]  restart;
  /** The data files (source e.g. WAVEFORM) for the inlet/outlets (0=Inlet/writer, N=Outlet/reader) */
  protected DataFile[] dfc;
  /** The data files (destination e.g. PLOT) for the inlet/outlets (0=Inlet/writer, N=Outlet/reader) */
  protected DataFile[] dfx;
  /** Time when rate was last checked (see getRate()) */
  protected double     timeLast  = 0;
  /** Write position when rate was last checked (see getRate()) */
  protected double     inPtrLast = 0;
  /** Last modified time (J1950) (initialized to time this object was created). since 3.3.0 */
  protected double     lastModified;

  /** Max number of history for old "linked" DataFiles to keep around (e.g. after a pipe restart). Since 3.7.0 for BUG 2991 */
  private int maxHistory = DEFAULT_MAX_HISTORY;
  /** Count of history (linked DataFile at this outlet). Since 3.7.0 for BUG 2991 */
  private int[] historyCount;
  /** Reference to tail of history for quick trimming of linked DataFile history (e.g. after pipe restart). Since 3.7.0 for BUG 2991 */
  private DataFile[] historyTail;

  /** Creates a new instance.
      @param M      Current Midas context (holds pipe references)
      @param name   Name of the pipe
   */
  public Pipe (Midas M, String name) {
    this(M, name, DEFAULT_MAX_OUTLETS);
  }

  /** Creates a new instance.
      @param M      Current Midas context (holds pipe references)
      @param name   Name of the pipe
      @param maxOut Max outlets (i.e. pipe readers)
   */
  public Pipe (Midas M, String name, int maxOut) {
    if (maxOut > 31) { // check add in 3.3.2 for BUG 2689
      // prior to 3.3.2, PipeResource used the constructor with default of 8 readers, but now allows the {MAXOUTLET=}
      // qualifier to increase limit, BUT 31 is current max b/c of outMask (32-bit int bitmask, where 0=inlet (writer))
      throw new IllegalArgumentException("Cannot set max outlets (pipe readers) > 31, given "+maxOut);
    }
    this.M            = M;
    this.name         = name;
    this.maxOut       = maxOut;
    this.outPtr       = new double[maxOut+1];
    this.refs         = new Command[maxOut+1];
    this.restart      = new boolean[maxOut+1];
    this.dfc          = new DataFile[maxOut+1];
    this.dfx          = new DataFile[maxOut+1];
    this.lastModified = Time.current();         // since 3.3.0 for BUG 2484
    this.historyCount = new int[maxOut+1];      // since 3.7.0 for BUG 2991
    this.historyTail  = new DataFile[maxOut+1]; // since 3.7.0 for BUG 2991
  }

  @Override
  public String toString () {
    return "Raw Pipe Name="+name;
  }

  /** Attach to pipe.
      @param dir Direction (one of {@link IOResource#DIR_INPUT},
                 {@link IOResource#DIR_OUTPUT}, {@link IOResource#DIR_INOUT}, etc.)
      @param df  Associated data file.
      @return Outlet number (1..N)
   */
  public synchronized int attach (int dir, DataFile df) {
    int outlet = 0;
    // if (dir>=0) {                                        // prior to 2.3.0
    if (dir>0 || (dir==0 && df!=null && df.cmd==refs[0])) { // since 2.3.0 to support HEADERMOD on pipes (writer to pipe)
      inlet      = 1;
      df.setDataStart(inPtr);
      if (dfc[0]!=null) { dfc[0] = linkDataFiles(dfc[0], df, -1); dfc[0].close(); }
      if (dfx[0]!=null) { dfx[0] = linkDataFiles(dfx[0], df, 0); }
      dfc[0]     = df; // current inlet (writer to pipe)
      dfx[0]     = df;
      refs[0]    = df.cmd;
      restart[0] = false;
    }
    else {                                                  // reader of pipe
      for (int i=1; i<=maxOut && outlet==0; i++) {
        // we have N-readers in maxOut-slots and we need to loop through to see if:
        //    1. this is a reconnect and we already have this primitive as a reader
        //       if so, we re-use that slot (RESTART) -- KNOWN DR: this logic
        //       can get confused if a primitive has 2+ readers for the same pipe
        //       (e.g. "MYMUX _out _in _in") and/or the pipe switches input
        //       slots (e.g. "MYMUX _out _in1 _in2" before restart and
        //       "MYMUX _out _in2 _in1" after restart).
        //    2. find the first available slot to attach to
        int mask = 1<<(i-1);
        if (restart[i] && (outMask&mask)!=0 && df.cmd==refs[i]) {  // CASE #1
          outlet      = i;
          dfx[outlet] = linkDataFiles(dfx[outlet], df, outlet);
        }
        else if (!restart[i] && (outMask&mask)==0) {               // CASE #2
          outlet         = i;
          outMask        = outMask | mask;
          outlets        = Math.max(outlets,outlet);
          outPtr[outlet] = inPtr;
          dfc[outlet]    = dfc[0];
        }
      }
      if (outlet>0) {
        refs[outlet]    = df.cmd;
        restart[outlet] = false;
        dfx[outlet]     = df;
      }
    }
    if (DEBUG) debug("Attach flags="+df.getFlagsString()+" dir="+IOResource.dirToString(dir),df,outlet);

    return outlet;
  }

  /** Links the old data file to the new one. Note that in the case of PLOT and
      RMIF this will be the same instance of DataFile (see diagram at top of
      class) requiring us to make a copy to use as the "old" file.
      @param old The old DataFile instance used (pre restart) or null for the first connect to the pipe.
      @param df  The new DataFile instance to use (post restart).
      @param historyIndex index into {@link #historyCount} and {@link #historyTail} to keep track of max
                 number of pipe history (DataFile links) during Pipe restarts (-1 to not trim history)
      @return The "old" instance to store.
   */
  private synchronized DataFile linkDataFiles (DataFile old, DataFile df, final int historyIndex) { // add synchronized in 3.7.0 as part of Bug 2991
    if (old != null) {
      if (old == df) {      // <-- Special case (for PLOT and RMIF): DataFile is reused on restart
        old = df.copy();                                 // clone current for old DF reference
        DataFile prevDF = old.getPreviousDataFile();
        if (prevDF != null) prevDF.setNextDataFile(old); // since 3.7.0: adjust next DF link as part of BUG 2991
        if (DEBUG) {
          // get unique IDs (similar to memory address) for DEBUG
          String id_df  = Integer.toHexString(System.identityHashCode(df));
          String id_old = Integer.toHexString(System.identityHashCode(old));
          debug("\nlinkDataFiles: Special RESTART index=" + historyIndex + " old is same as new=" + id_df + " old clone="+id_old + "  <---- RESTART");
        }
      }
      old.setNextDataFile(df);
      df.setPreviousDataFile(old);

      if (DEBUG) {
        // get unique IDs (similar to memory address) for DEBUG
        String id_df = Integer.toHexString(System.identityHashCode(df));
        String id_old = Integer.toHexString(System.identityHashCode(old));
        String tmp = (historyIndex < 0)? "" : " count="+historyCount[historyIndex];
        debug("linkDataFiles: index=" + historyIndex + " old=" + id_old + " new=" + id_df + tmp);
      }

      if (historyIndex >= 0) {              // Trim old previous DataFile references to address
        trimHistory(old, df, historyIndex); // resource/memory leak. Since 3.7.0 for BUG 2991
      }
    }
    return old;
  }

  /**
   * Trim old previous DataFile references after maxHistory count to address resource/memory leak.
   * This is important specifically when there are lots of pipe restarts. For BUG 2991.
   * @param old The old DataFile instance used (pre restart) (should NOT be null)
   * @param df  The new DataFile instance to use (post restart) (should NOT be null.
   * @param historyIndex index into {@link #historyCount} and {@link #historyTail} to keep track of max
   *            number of pipe history (DataFile links) during Pipe restarts (negative to not trim history - do nothing)
   * @since NeXtMidas 3.7.0
   */
  private synchronized void trimHistory (DataFile old, DataFile df, final int historyIndex) {
    if (historyIndex >= 0) {
      DataFile newTail = null;
      DataFile tail = historyTail[historyIndex];
      int count = historyCount[historyIndex];

      if (count < maxHistory || tail == null ) {
        if (tail == null) {                  // no reference to tail of linked DataFiles
          historyCount[historyIndex] = 1;    // reset count
          newTail = old;                     // old will become the new tail
          if (count > 1) warn("trimHistory: tail is null but count(" + count +") > 1! index="+ historyIndex+ " <=== WARN", df, historyIndex);
        }
        else {
          historyCount[historyIndex]++;      // increment count, tail remains the same
        }
      }
      else {                                 // *** need to trim history ***
        newTail = tail.getNextDataFile();    // current tail's next DataFile will become new tail
        tail.setNextDataFile(null);          // cut next DF link - to prevent resource leak

        if (DEBUG) {
          // get unique ID (similar to memory address) for DEBUG
          String id_tail    = Integer.toHexString(System.identityHashCode(tail));
          String id_newTail = Integer.toHexString(System.identityHashCode(newTail));
          debug("trimHistory: TRIM history: index="+ historyIndex+ " tail=" + id_tail + " newTail=" + id_newTail+ "\n");
        }

        if (newTail != null) {
          newTail.setPreviousDataFile(null); // cut prev DF link - to prevent resource leak
        }
        else {
          newTail = old;                     // newTail is null, so look through linked DataFiles starting at old
          int ii;
          for (ii=1; ii<maxHistory; ii++) {
            DataFile prevDF = df.getPreviousDataFile();
            if (prevDF == null) break;
            else newTail = prevDF;
          }
          // get unique IDs (similar to memory address) for DEBUG
          String id_old = Integer.toHexString(System.identityHashCode(old));
          String id_newTail = Integer.toHexString(System.identityHashCode(newTail));
          warn("trimHistory: WARN: unable to cut previous DataFile link since new tail is null! index="+ historyIndex
              + " setting tail=" +  id_newTail + " ii="+(ii+1) + " old=" + id_old + " <=== WARN", df, historyIndex);
        }
      }

      // adjust tail to point to new tail
      if (newTail != null) historyTail[historyIndex] = newTail;
    }
  }

  /** Detach from a pipe.
      @param outlet The (Pipe reader) outlet number (1..N) or 0 for inlet (Pipe writer).
      @param df     Associated data file.
      @return The new outlet number (usually {@link #DISCON}).
   */
  public synchronized int detach (int outlet, DataFile df) {
    return detach(outlet, df, ((df.flags&DataFile.REOPEN)!=0));
  }

  /** Detach from a pipe.
      @param outlet The (Pipe reader) outlet number (1..N) or 0 for inlet (Pipe writer).
      @param df     Associated data file.
      @param reopen true is this pipe is being reopened (e.g. after a Pipe restart)
      @return The new outlet number (usually {@link #DISCON}).
   */
  private synchronized int detach (int outlet, DataFile df, boolean reopen) {
    if (outlet==DISCON) return outlet; // already detached
    DataFile next = dfc[outlet].getNextDataFile();
    restart[outlet] = reopen || (next!=null);
    if (DEBUG) debug("Detach flags="+df.getFlagsString()+" reopen="+reopen+" next="+next,df,outlet);

    if (outlet==0) {
      if (!restart[outlet]) inlet = 0; // set inlet "flag" to zero to signal that writer is detached
      restart[outlet] = true;
    }
    else {
      int i=outlet-1;
      if (next != null) {
        dfc[outlet]     = next;
        outPtr[outlet]  = next.getDataStart();
        restart[outlet] = true;
      }
      else if (restart[outlet]) {
        // the input did not chain
      }
      else {
        outMask &= ~(1<<i);
      }
      for (i=outlets-1; i>=0 && (outMask&(1<<i))==0; i--) { }
      outlets = i+1;
      outlet = DISCON;
    }

    return outlet;
  }

  /** Update the combined read pointers and return the new value. Note that this recomputes the
      out pointers and detaches any readers that failed to reconnect to the pipe on restart. Due
      to this, most users should use {@link #getOutByte()} and should NOT call this method.
      @return new pointer value after read pointers are combined
   */
  public double lastOut () {
    double out = inPtr;
    for (int i=0; i<outlets; i++) {
      if ((outMask&(1<<i)) != 0 && outPtr[i+1]<out) {
        // Disconnect readers that did not reconnect, see %TEST_RESTART_BLOCKING
        if (restart[i+1]) {
          synchronized(this) {  // sync done to prevent mods to REFs while detaching (since 3.4.0 - Bug 2755)
            if (refs[i+1].state != Command.RESTART) {
              int flags = dfx[i+1].getFlags();
              if ((flags & BaseFile.ALWAYSREOPEN) == 0) { // Only check those that aren't marked as AlwaysReopen
                // Command restarted, but did not reconnect to this pipe after restart.
                M.info("PIPE: "+Convert.ref2String(refs[i+1])+" restarted, but did not reconnect to "+name);
                detach(i+1, dfx[i+1], false);
              }
            }
          }
        }
        out=outPtr[i+1];
      }
    }
    outPtr[0] = out;
    return out;
  }

  /** Read the pipe header.
      @param hb     Buffer to write header to.
      @param outlet Outlet number of outlet to copy the header from.
      @return Number of bytes copied (0 if outlet is disconnected).
   */
  public int readHeader (byte[] hb, int outlet) {
    if (outlet==DISCON) return 0;
    int bytes = Math.min(hb.length, dfc[outlet].hb.length);
    System.arraycopy(dfc[outlet].hb,0, hb,0, bytes);
    return bytes;
  }

  /** Write the pipe header.
      @param hb     Buffer to read header from.
      @param outlet Outlet number of outlet to copy the header to.
      @return Number of bytes copied.
   */
  public int writeHeader (byte[] hb, int outlet) {
    // its already there in dfc
    int bytes = Math.min(hb.length, dfc[outlet].hb.length);
    System.arraycopy(hb,0, dfc[outlet].hb,0, bytes);
    lastModified = Time.current(); // since 3.3.0
    return bytes;
  }

  /** Find bytes available for this outlet.
      @param outlet The (Pipe writer) outlet number (1..N) or 0 for inlet (Pipe writer).
      @return Number of bytes available (when negative see: {@link CoreIO#CLOSED}, {@link CoreIO#REOPEN}, {@link CoreIO#ERROR})
   */
  public double avail (int outlet) {
    double status;
    double in = inPtr;
    if (outlet==0) status = size - (in-lastOut());
    else if (outlet<0 || outlet>maxOut) status=CoreIO.ERROR; // invalid outlet number
    else {
      DataFile df = dfc[outlet];
      // if cascaded then the rest of this tube is available
      if (df.getNextDataFile()!=null) in = df.getDataStop();
      status = in-outPtr[outlet];
      if (status>0) { } // OK
      else if (status<0) status=0;
      else if (inlet==0) status=CoreIO.CLOSED;
      else if (df.getNextDataFile()!=null) status=CoreIO.REOPEN;
    }
    if (status>size) {
      // This should be impossible, if seen it suggests there is an error in the inPtr and outPtr
      // accounting or that the write was allowed to continue even with a full pipe. It does NOT
      // mean that we wrote outside of the buffer (that would be an ArrayIndexOutOfBoundsException).
      if (outlet==0) M.warning("Pipe overflow detected for "+name+" on pipe writer. avail="+status+" > size="+size+"! inPtr="+inPtr+" in="+in+" outPtr="+getOutByte(outlet)+ " outlet="+outlet);
      else M.warning("Pipe overflow detected for "+name+" closing pipe reader. avail="+status+" > size="+size+"! inPtr="+inPtr+" in="+in+" outPtr="+getOutByte(outlet)+ " outlet="+outlet);
      status = CoreIO.ERROR;
    }
    return status;
  }

  /** Check to see if the specified number of bytes are available for reading.
      @param outlet   The (Pipe reader) outlet number (1..N).
      @param numBytes The number of bytes to check for reading.
      @return true if the number of bytes specified are available for reading or if the pipe
              ends/restarts before the specified number of bytes will be available; false if
              not near the end of the pipe and the specified bytes are not available yet.
      @throws MidasException if the number of bytes specified exceeds the size of the pipe or if
                             the outlet number is invalid.
   */
  public boolean isReady (int outlet, int numBytes) {
    if (numBytes > size) {
      throw new MidasException("Transfer length of "+numBytes+" bytes exceeds pipe size of "+size+" bytes");
    }
    if ((outlet <= 0) || (outlet > maxOut)) {
      throw new MidasException("Invalid outlet number "+outlet);
    }
    // The code below is largely based on the avail(..) method
    double in = inPtr;
    DataFile df = dfc[outlet];
    if (df.getNextDataFile()!=null) in = df.getDataStop();
    double avail = in-outPtr[outlet];

    return (avail >= numBytes)           // ready - corrected in 3.8.1
        || (inlet==0)                    // CLOSED
        || (df.getNextDataFile()!=null); // REOPEN
  }

  /** Write to pipe section.
      @param buf    The data buffer.
      @param lbuf   Native buffer location (0=Disable).
      @param boff   The buffer start offset.
      @param bytes  Number of bytes to write.
      @param outlet (not used, keeps signature same as read)
      @return Number of bytes written; -1 if not in pipe section or pipe is stopped;
              0 if pipe is paused or there no available "free" buffer to write data.
   */
  public int write (byte[] buf, long lbuf, int boff, int bytes, int outlet) {
    if (M.pipeMode==Midas.PPAUSE) return 0;
    if (M.pipeMode==Midas.PSTOP || M.pipeMode==Midas.POFF) return -1;
    double out = lastOut();
    int poff = (int)(inPtr % (double)size);
    int bmax = size-poff;
    if (bmax<bytes) bytes=bmax;
    int pmax = size - (int)(inPtr-out);
    if (pmax<bytes) bytes=pmax;
    if (bytes>0) {
      try { // try catch added in 3.5.4 for Bug 2920
        if (lbuf!=0) Native.p2ja (lbuf,boff,buffer,poff,bytes);
        else System.arraycopy (buf,boff,buffer,poff,bytes);
      }
      catch (Exception e) {
        // since 3.7.0: always print detailed warning message to help debug the cause of the Exception
        //         1234567890123456789012345678901234567890123456789012345678901234567 <-- 67
        StringBuilder sb = new StringBuilder(1400); // the below warning msg is ~1324 chars
        sb.append("***************************************************************\n"); // adjusted for "WARN: " prefix
        sb.append("** Pipe.write: Exception details ************************************\n");
        sb.append("**                                                                 **\n");
        sb.append(StringUtil.padRight("**     Pipe name: " + getName(), 67)        + "**\n");
        sb.append(StringUtil.padRight("**     buffer: " + buf, 67)                 + "**\n");
        if (buf !=null) sb.append(StringUtil.padRight("**     buffer length: " + buf.length, 67) + "**\n");
        sb.append(StringUtil.padRight("**     native buffer location: " + lbuf, 67)+ "**\n");
        sb.append(StringUtil.padRight("**     buffer offset: " + boff, 67)         + "**\n");
        sb.append(StringUtil.padRight("**     number bytes to write: " + bytes, 67)+ "**\n");
        sb.append(StringUtil.padRight("**     outlet: " + outlet, 67)              + "**\n");
        String pipeModeInfo = Parser.get(Midas.pipeModes,M.pipeMode,-1) + " (" + M.pipeMode + ")";
        sb.append(StringUtil.padRight("**     pipeMode: " + pipeModeInfo, 67)      + "**\n");
        sb.append(StringUtil.padRight("**     out/lastOut(): " + out, 67)          + "**\n");
        sb.append(StringUtil.padRight("**     poff: " + poff, 67)                  + "**\n");
        sb.append(StringUtil.padRight("**     bmax: " + bmax, 67)                  + "**\n");
        sb.append(StringUtil.padRight("**     pmax: " + pmax, 67)                  + "**\n");
        sb.append("**                                                                 **\n");
        sb.append("** END Pipe.write: Exception details ********************************\n");
        sb.append("*********************************************************************");
        M.warning(sb);

        if (M.io.isOptionSet(IOOptions.WriteNoAbort)) {
          // Since 3.5.4: "ENV SET IOOptions +WriteNoAbort" keeps the write from aborting (Bug 2920)
          M.printStackTrace("Pipe.write() exception suppressed with: 'ENV SET IOOptions +WriteNoAbort'\n"
                          + "Pipe.write() Exception/stack trace info for " + getName(), e, true, Midas.WARNING);
        }
        else {
          // throw new MidasException(e); // since 3.5.4 for BUG 2920 - TODO remove this line in 3.9+
          throw e; // since 3.7.0: re-throw original exception for BUG 2920
        }
      }
      inPtr += bytes;
      lastModified = Time.current(); // since 3.3.0
    }
    else {
      bytes=0;
    }
    return bytes;
  }

  /** Read from pipe section.
      @param buf    The data buffer.
      @param lbuf   Native buffer location (0=Disable).
      @param boff   The buffer start offset.
      @param bytes  Number of bytes to read.
      @param outlet The (Pipe reader) outlet number the reading is being done on behalf of (1..N).
      @return Number of bytes written.
   */
  public int read (byte[] buf, long lbuf, int boff, int bytes, int outlet) {
    if (M.pipeMode==Midas.PPAUSE) return 0;
    if (M.pipeMode==Midas.PSTOP || M.pipeMode==Midas.POFF) return CoreIO.CLOSED;
    DataFile df = dfc[outlet];
    double out = outPtr[outlet];
    int poff = (int)(out % (double)size);
    int bmax = size-poff;
    if (bmax<bytes) bytes=bmax;
    double in = inPtr;
    if (df.getNextDataFile()!=null) in = df.getDataStop();
    int pmax = (int)(in-out);
    if (pmax<bytes) bytes=pmax;   // make sure we DO NOT read more than available for our outlet
    if (bytes>0) {
      if (bytes>size) bytes=size;
      if (lbuf!=0) Native.ja2p (buffer,poff,lbuf,boff,bytes);
      else System.arraycopy (buffer,poff,buf,boff,bytes);
      outPtr[outlet] = out + bytes;
    }
    else if (df.getNextDataFile()!=null) {
      return CoreIO.REOPEN;
    }
    else if (inlet==0) {
      return CoreIO.CLOSED;
    }
    else {
      bytes=0;
    }
    return bytes;
  }

  /** Get the inlet connection flag
   *  @return the inlet connection flag (0 if none). */
  public short getInlet () { return (short)inlet; }

  /** Get the number of outlets
   *  @return the number of outlets. */
  public short getOutlets () { return (short)outlets; }

  /** Get the outlet mask mapping connected outlets.
      deprecated Since NeXtMidas 3.3.2, this may become protected in 3.7+
      @return the outlet mask mapping connected outlets
   */
  @InternalUseOnly("Since NeXtMidas 3.3.2")
  public int getOutMask () { return outMask; }

  /** Indicates if a given outlet is in use.
      @param outlet The outlet number (1..N) or 0 for inlet.
      @return true if a given outlet is in use.
      @since NeXtMidas 2.5.0
   */
  public boolean isAttached (int outlet) {
    if (outlet == 0) {
      return (dfc[0] != null);
    }
    else {
      int mask = 1<<(outlet-1);
      return ((outMask&mask) != 0);
    }
  }

//  /** Gets the maximum number of outlets (readers) allowed for this pipe.
//      @since NeXtMidas 2.5.0
//      @deprecated Since NeXtMidas 3.3.2, use {@link #getMaxOutlets()}
//   */ @Deprecated // TODO: remove in 3.7+
//  public int getMaxOut () {
//    return maxOut;
//  }

  /** Gets the maximum number of outlets (readers) allowed for this pipe.
      @return maximum number of outlets
      @since NeXtMidas 3.3.2
   */
  public int getMaxOutlets () {
    return maxOut;
  }

  /** Get the size of the pipe in bytes
   *  @return the size of the pipe in bytes. */
  public int getSize () { return size; }

  /** Get index of the next byte to be written
   *  @return index of the next byte to be written. */
  public double getInByte () { return inPtr; }

  /** Get over-all outlet byte count
   *  @return over-all outlet byte count. Same as <code>getOutByte(0)</code>. */
  public double getOutByte () { return outPtr[0]; }

  /** Get index of the next byte to be read by a given outlet
   *  @return index of the next byte to be read by a given outlet.
   *  @param i Outlet number (1..N) or 0 for inlet (inlet uses limit for all outlets).
   */
  public double getOutByte (int i) { return outPtr[i]; }

  /** Get the number of bytes available waiting to be read
   *  @return the number of bytes available waiting to be read. */
  public double getAvail () { return (inPtr-outPtr[0]); }

  /** Set the approximate data rate for the pipe
   *  @return the approximate data rate for the pipe (bytes/sec). */
  public double getRate () {
    double time = Time.current();
    double xfer = (inPtr-inPtrLast);
    double rate = xfer/(time-timeLast);
    timeLast = time;
    inPtrLast = inPtr;
    return rate;
  }

  /** Get the data file for the inlet
   *  @return the data file for the inlet (i.e. Pipe writer). Same as <code>getDataFile(0)</code>. */
  public DataFile getDataFile () {
    return dfc[0];
  }

  /** Gets the data file for a given inlet/outlet.
      @param outlet (Pipe reader) Outlet number (1..N) or 0 for inlet (Pipe writer).
      @return data file for given inlet/outlet
   */
  public DataFile getDataFile (int outlet) {
    return (outlet!=DISCON) ? dfc[outlet] : null;
  }

  /** Gets the owner for a given inlet/outlet.
      @param outlet (Pipe reader) Outlet number (1..N) or 0 for inlet (Pipe writer).
      @return owner of inlet/outlet
   */
  public Object getOwner (int outlet) {
    return (outlet!=DISCON) ? refs[outlet] : null;
  }

  /** Sets the size of the pipe. This allows dynamic setting of the pipe size
      for tuning. These are the steps done by this method:
      <pre>
        (1) Set the shared pipe mode to {@link Midas#PPAUSE}.
        (2) Discard the current pipe buffer (discards any unread data).
        (3) Create a new pipe buffer with the new size.
        (4) Restore the shared pipe mode.
      </pre>
      @param value The new size of the pipe (in bytes).
   */
  public void setSize (int value) {
    if (buffer==null || value!=size) {
      int mode = M.pipeMode;
      if (mode==Midas.PRUN) { M.pipeMode=Midas.PPAUSE; Time.sleep(0.2); }
      buffer = new byte[value];
      size = value;
      if (M.pipeMode!=mode) M.pipeMode=mode;
    }
  }

  /** Set index of the next byte to be written.
      @param value index of the next byte to be written*/
  public void setInByte ( double value) { inPtr=value; }

  /** Set over-all outlet byte count. Same as <code>setOutByte(0,value)</code>.
      @param value outlet byte count */
  public void setOutByte (double value) { outPtr[0]=value; }

  /** Set index of the next byte to be read by a given outlet.
      @param i     Outlet number (1..N) or 0 for inlet.
      @param value The new index.
   */
  public void setOutByte (int i, double value) { outPtr[i]=value; }

  /** The pipe is always chained to the data file used by the inlet.
      Same as <code>getDataFile()</code>
      @return get chained DataFile
   */
  @Override
  public Object getNextLink () { return dfc[0]; }

  /** There is no previous link for a pipe, always returns null.
      @return null, as there is no previous link for a pipe */
  @Override
  public Object getPrevLink () { return null; }

  /** method used for signaled (vs polled) operation */
  public synchronized void awaken() { notifyAll(); }

  /** Get this pipe's name
   *  @return this pipe's name
   *  @since NeXtMidas 3.1.1
   */
  public String getName () {
    return name;
  }

  /** Get last modified time in seconds since J1950
   *  @return Last modified time in seconds since J1950.
   *  @since NeXtMidas 3.3.0
   */
  public double getLastModified () {
    return lastModified;
  }

  /** Set max number of history for old "linked" DataFiles to keep around
   *  @return Max number of history for old "linked" DataFiles to keep around (e.g. after a pipe restart).
   *  @since NeXtMidas 3.7.0
   */
  public int getMaxHistory () {
    return maxHistory;
  }

  /** Get max number of history for old "linked" DataFiles to keep around
   *  @param newMax Max number of history for old "linked" DataFiles to keep around (e.g. after a pipe restart). MUST be greater than 2.
   *  @since NeXtMidas 3.7.0
   */
  @InternalUseOnly("Since added")
  public synchronized void setMaxHistory (int newMax) {
    if (newMax < 2) throw new IllegalArgumentException("Cannot set max history < 2, given "+newMax);
    maxHistory = newMax;
  }

  /** Get if specialized pipe debug is enabled
   *  @return true if specialized pipe debug is enabled, otherwise false */
  @InternalUseOnly("Since added in NeXtMidas 3.7.0")
  public static boolean isDebug () { // may change to debug mask in future
    return DEBUG;
  }

  /** Set specialized pipe debug
   *  @param newDebug true to turn on specialized pipe debug, false to turn it off. */
  @InternalUseOnly("Since added in NeXtMidas 3.7.0")
  public static void setDebug (boolean newDebug) { // may change to debug mask in future
    DEBUG = newDebug;
  }

  /** Prints a message with this pipe's name for either debug (as INFO) or warning severity.
   *  If M is null, uses Shell.warning(..) if severity is WARNING, otherwise uses Shell.info(..).
   *  @param severity The severity type of the message (either {@link Midas#INFO} or {@link Midas#WARNING})
   *  @since NeXtMidas 3.7.0 (extracted from debug(String msg) method)
   */
  private void msg (int severity, String msg) {
    String message = "Pipe "+name+": "+msg;
    if (M != null) {
      M.message(message, severity);
    }
    else if (severity == Midas.WARNING) {
      Shell.warning(msg);
    }
    else {
      Shell.info(message);
    }
  }

  /** Prints a debug message with this pipe's name. */
  private void debug (String msg) {
    msg(Midas.INFO, msg);
  }

  /** Prints a debug message with this pipe's name along with df and outlet. */
  private void debug (String msg, DataFile df, int outlet) {
    String id = (df == null)? null : Convert.ref2String(df.cmd);

    debug(msg+" owner="+id+" outlet="+outlet);
  }

  /** Prints a warning message with this pipe's name along with df and outlet. Since 3.7.0 */
  private void warn (String msg, DataFile df, int outlet) {
    String id = (df == null)? null : Convert.ref2String(df.cmd);
    msg(Midas.WARNING, msg + " owner=" + id + " outlet=" + outlet);
  }
}
