package nxm.sys.lib;

import java.io.FilenameFilter;

import nxm.sys.lib.CoreIO.IOOptions;
import nxm.sys.lib.TimeLine.NotifyInvalidEntry;

/**
  Resource for interacting with a NeXtMidas Pipe.

  @author Jeff Schoen
*/
public class PipeResource extends IOResource {

  /** the Pipe entity */
  public Pipe pipe;
  /** 0=writer, 1:N for readers */
  public int outlet;
  /** key in pipe table */
  public String name;
  /** parent DataFile */
  public DataFile df;
  /** default pipe size (in bytes) */
  public int pipeSize;
  /** default pipe pause time (in seconds)*/
  public double pause=0.0125;
  /** Toggles polling on and off */
  public static boolean POLLING = true;
  private double diffInReconnectOffsetFromInitialConnect = 0; // used on reconnects to get number of inlet elements since connect

  /** initial reader's seek position into pipe (set on first connect in {@link #initTimecode()}. */
  private double initialConnectSeekPos = 0;

  /** PipeResource Default Constructor */
  public PipeResource () { }

  /** return if this resource is a stream resources */
  public boolean isStream () { return true; }

  /** return if this resource is currently streaming */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  public boolean isStreaming () {
    if (pipe==null) return true; // not connected yet
    DataFile df = pipe.getDataFile(outlet);
    if (df==null) return true; // in transition
    if (df.isOpen) return true; // still writing
    else if (df.getNextDataFile()!=null) return false; // no input, but chained
    else if (pipe.inlet==0) return false; // all done
    else return true;	// no input, but not ready to chain
  }

  /** test existence of the resource */
  public boolean exists () {
    name = urlstr.substring(4).toUpperCase();
    return M.pipes.containsKey(name);
  }

  /** open the resource */
  public synchronized boolean open () { // since 3.1.1 added synchronized to address BUG 2457
    // 2010-09-13 JWB: The line below was MOVE from the two blocks below to here because if there
    // Since 2.9.0:    is a restart of an output pipe then the dir could be >0 and
    //                 the pipe could be null but still exist in the pipes table. See BUG 1166.
    pipe = (Pipe) M.pipes.get(name);

    if (dir <= DIR_INOUT) { // if dir is for INOUT(0), INPUT(-1), INPUT_TRUNC(-2), etc.
      //pipe = (Pipe) M.pipes.get(name);  // MOVED above - See 2010-09-13 comment above
      outlet = pipe.attach(dir,df);
      initTimecode();
    }
    else if (pipe==null) {  // this happens when dir is for OUPUT(1) or OUTPUT_TRUNC(2)
      int maxOutlets = Pipe.DEFAULT_MAX_OUTLETS;
      if (qualTable != null) {
        maxOutlets = qualTable.getL("MAXOUTLETS", maxOutlets); // since 3.3.2 to allow setting max number of readers on a pipe
      }
      pipe = new Pipe(M, name, maxOutlets); // since 3.3.2: pass maxOutlet to allow setting max number of readers
      pipe.setSize(pipeSize);
      outlet = pipe.attach(dir,df);
      // Get rid of any timeline keywords that have been propagated from input to output,
      // as PipeResource will handle timeline propagation taking into account the different BPE
      // (important with TC keywords) - experimental for 3.8.1
      if (!M.io.isOptionSet(IOOptions.DoublePropagateTimelineKeywordsInPipe)) df.deleteTimeLineKeywords();
      M.pipes.put (name,pipe);
    }
    else {                 // have existing pipe (on a RESTART)
      //pipe = (Pipe) M.pipes.get(name);  // MOVED above - See 2010-09-13 comment above
      outlet = pipe.attach(dir,df);
      M.pipes.put (name,pipe);
    }
    pause = M.pause;
    if (M.macro!=null) pause = M.macro.lines[1].getD("/POLL",pause);
    isOpen = (outlet >= 0);
    return isOpen;
  }

  /** close the resource */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  public synchronized boolean close () { // since 3.1.1 added synchronized to address BUG 2457
    if (pipe==null) return true;
    pipe.detach(outlet,df);
    if (outlet==0 && pipe.inlet==0) M.pipes.remove(name);
    return true;
  }

  /** delete the resource */
  public boolean delete () {
    M.pipes.remove(name);
    return true;
  }

  /** get length of resource */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  public long getLength () {
    return (long)pipe.inPtr;
  }

  /** seek to the specified offset in file */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  public boolean seek (long offset) {
    if (outlet>0) pipe.outPtr[outlet]=(double)offset;
    this.offset = offset;
    return true;
  }

  /** get the current seek offset */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  public long seek () {
    if (outlet>0) return (long)pipe.outPtr[outlet];
    else return (long)pipe.inPtr;
  }

  /** get currently available length of resource */
  public long avail() {
//System.out.println("avail = "+(long)pipe.avail(outlet));
    return (long)pipe.avail(outlet);
  }

  @Override
  public boolean isReady (int numBytes) {
    return pipe.isReady(outlet, numBytes);
  }

  /** read data from resource */
  public int read (byte[] buf, int boff, int bytes) {
    return read(buf,0L,boff,bytes);
  }

  /** write data to resource */
  public int write (byte[] buf, int boff, int bytes) {
    return write(buf,0L,boff,bytes);
  }

  /** read data from resource */
  public int read (long lbuf, int boff, int bytes) {
    return read(null,lbuf,boff,bytes);
  }

  /** write data to resource */
  public int write (long lbuf, int boff, int bytes) {
    return write(null,lbuf,boff,bytes);
  }

  /** read data from resource */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  private int read (byte[] buf, long lbuf, int boff, int bytes) {
    int status=0, off=boff;
    while (bytes>0) {
      status = pipe.read(buf,lbuf,off,bytes,outlet);
      if (status>0) { bytes-=status; off+=status; }
      else if (status<0) break;
      else if ((df.cmd != null) && (df.cmd.isStateChanged())) break; // Added in 2.5.0
      else if (POLLING) Time.sleep(pause);
      else try { pipe.wait(); } catch (Exception e) { };
    }
    offset = (long)pipe.outPtr[outlet];
    if (!POLLING) pipe.awaken();
    if (status<0 && off==boff) return status;
    propagateTimecode();  // JC4: Added in 2.5.0 - propagate timecode for the pipe
    return off-boff;
  }

  /** write data to resource */
  @SuppressWarnings("deprecation") // for usage of pipe fields what should become package-private
  private int write (byte[] buf, long lbuf, int boff, int bytes) {
    int status=0, off=boff;
    while (bytes>0) {
      status = pipe.write(buf,lbuf,off,bytes,outlet);
      if (status>0) { bytes-=status; off+=status; }
      else if (status<0) break;
      //else if ((df.cmd != null) && (df.cmd.isStateChanged())) break; // Added in 2.5.0 <-- JC4: I think we want this, but I don't have a test case for it.
      else if (POLLING) Time.sleep(pause);
      else try { pipe.wait(); } catch (Exception e) { };
    }
    offset = (long)pipe.inPtr;
    if (!POLLING) pipe.awaken();
    if (status<0 && off==boff) return status;
    return off-boff;
  }

  /** get List of container entries */
  public String[] getEntryList (FilenameFilter filter) {
    return filterEntries(filter, M.pipes.getKeys());
  }

  /** get entry from Container */
  public IOResource getEntry (String name) { return null; }

  /** @since NeXtMidas 3.3.0 */
  @Override
  public Time lastModifiedTime () {
    return new Time(pipe.lastModified);
  }

  /** @since NeXtMidas 3.3.0 */
  @Override
  public long lastModified () {
    return (long) Time.toJ1970(pipe.lastModified); // convert J1950 to J1970
  }

  /** implement connect function (attach/detach from pipe)
   *  @param flag attach if greater than 0, detach if less than 0
   */
  public void connect (int flag) {
    if (flag>0 && outlet==Pipe.DISCON) outlet = pipe.attach(dir,df);
    if (flag<0 && outlet>0) outlet = pipe.detach(outlet,df);
  }

  /** Calls {@link Pipe#readHeader(byte[], int)} to read the pipe header
      @param hb buffer to write header to
      @return number of bytes copied (0 if outlet is disconnected)
   */
  public int readHeader (byte[] hb) {
    return pipe.readHeader (hb,outlet);
  }

  /** Calls {@link Pipe#writeHeader(byte[], int)} to write the pipe header
      @param hb buffer to read header from
      @return number of bytes copied
   */
  public int writeHeader (byte[] hb) {
    return pipe.writeHeader (hb,outlet);
  }

  /** Initialize the timeline in a pipe reader. This is only called by {@link #open()} for a pipe reader
      AFTER the call to attach().
   */
  private void initTimecode () {
    DataFile previous = df.getPreviousDataFile();

    if (previous == null) { // Outlet on first connect
      df.newTimeLineFromDF(pipe.getDataFile(outlet));
      initialConnectSeekPos = pipe.getDataFile(outlet).seek(); // 2010-12-20 NTN: for BUG 2265 // used to get position at connect - used for warm start
      propagateTimecode();

    } else {                // Outlet or Inlet on reconnect
      // Gets the time where we left off on the previous DataFile and adds delta to get the starttime
      // This is pretty good, but does not cover the case where there is a TimeLine entry for the next element
      Time stopTime  = previous.getTimeAt(previous.seek(), null);
      Time startTime = stopTime.addSec(previous.getDelta());

      // This just creates an empty TimeLine with the right properties (ex: tolerance)
      df.newTimeLineFromDF(previous);
      // set first value of this timeline based the time of the last entry of the previous file
      // (obtained via getTimeAt) and adding the delta - the "pretty good" time estimate (temporarily at least)
      df.setTimeAt(0.0, startTime, NotifyInvalidEntry.WARNING);

      // Figure out how far we have progressed in the inlet file since the connect
      // On a reset, df.seek() is 0 because it is a new file, so propagateTimecode needs to know
      // this value in order to grab the correct inlet offset's time
      if (!M.io.isOptionSet(IOOptions.DisablePipeRestartTimeLinesFix) && df.isInput) {
        DataFile in = pipe.getDataFile(outlet);
        if (in != null) {
          TimeLine inTimeLine = in.getTimeLineHandler();
          if (inTimeLine != null && in.getBPE() != 0) {
            diffInReconnectOffsetFromInitialConnect = Math.floor(in.getTimeLineHandler().getOffsetAt(startTime)/in.getBPE()) - initialConnectSeekPos;
          }
          else if (inTimeLine == null) { // since 3.9.1
            diffInReconnectOffsetFromInitialConnect = Math.floor((startTime.getSec()- in.getTimeAt(initialConnectSeekPos))/in.getDelta());
          }
        }
      }
    }
  }

  /** Propagate the timecode from pipe writer to pipe reader. */
  private void propagateTimecode () {
    DataFile in = pipe.getDataFile(outlet); // Inlet to the pipe
    double seekBytes = df.seek() * df.getBPE();             // Outlet's seek in bytes (this reader of the pipe)
    double seekElem  = Math.floor(seekBytes / in.getBPE()); // Inlet's position in elements (the writer of the pipe)

    Time time;
    if (!M.io.isOptionSet(IOOptions.DisablePipeRestartTimeLinesFix)) {
      // in grabbing the inlet position time, compensate for pipe resets by taking into account the offset where the reset occurred
      time = in.getTimeAt(diffInReconnectOffsetFromInitialConnect+initialConnectSeekPos+seekElem, null); // since 3.8.1
    }
    else {
      // 3.8.0 and earlier behavior
      time = in.getTimeAt(initialConnectSeekPos+seekElem, null); // Time at current seek() + initialConnectSeekPos (2010-12-20 for BUG 2265)
    }

    // NOTE (added in 3.8.0): This will cause a warning in the following situation:
    // 1) noop in a pipe
    // 2) input file to noop has timeline keywords
    // This causes a TimeLine object to be created based on the propagated keywords and then
    // (starting with offset=0) this call to setTimeAt to try to add to that timeline
    // There is an experimental way to avoid the timeline keywords being propagated using a switch to noop

    if ((df.timeLine==null) || (df.timeLine.getSize()==0)) {
      // new TimeLine ... propagate TimeLine Format in a pipe NXM-3639 3.9.4
      if (in.sourceTimeLineFormat != null) df.setTimeLineFormat(in.sourceTimeLineFormat);
      if (in.sourceTimeLineTolr   != 0)    df.setTimeLineTolerance(in.sourceTimeLineTolr);
      if (df.timeLine.format != TimeLine.Format.MICROSECOND_PRECISION) {
        df.timeLine.setAlloc(TimeLine.getHighPrecisionAllocDefault());
      }
    }

    df.setTimeAt(time, NotifyInvalidEntry.WARNING);

  }
}
