commonlib/src/main/java/com/streamsets/pipeline/lib/io/FileContext.java
public void releaseReader(boolean inErrorDiscardReader) throws IOException {
    Utils.checkState(open, "FileContext is closed");
    // update starting offsets for next invocation either cold (no reader) or hot (reader)
    boolean hasNext;
    try {
      hasNext = reader != null && reader.hasNext();
    } catch (IOException ex) {
      IOUtils.closeQuietly(reader);
      reader = null;
      hasNext = false;
    }
    boolean doneWithFile = !hasNext || inErrorDiscardReader;
    if (doneWithFile) {
      IOUtils.closeQuietly(reader);
      reader = null;
      // Using Long.MAX_VALUE to signal we reach the end of the file and next iteration should get the next file.
      setStartingCurrentFileName(currentFile);
      setStartingOffset(Long.MAX_VALUE);

      // If we failed to open the file in first place, it will be null and hence we won't do anything with it.
      if(currentFile == null) {
        return;
      }

      // File end event
      LiveFile file = currentFile.refresh();

      if (inErrorDiscardReader) {
        LOG.warn("Processing file '{}' produced an error, skipping '{}' post processing on that file",
                 file, postProcessing);
        eventPublisher.publish(new FileEvent(file, FileEvent.Action.ERROR));
      } else {
        eventPublisher.publish(new FileEvent(file, FileEvent.Action.END));
        switch (postProcessing) {
          case NONE:
            LOG.debug("File '{}' processing completed, post processing action 'NONE'", file);
            break;
          case DELETE:
            if(!inPreviewMode) {
              try {
                Files.delete(file.getPath());
                LOG.debug("File '{}' processing completed, post processing action 'DELETED'", file);
              } catch (IOException ex) {
                throw new IOException(Utils.format("Could not delete '{}': {}", file, ex.toString()), ex);
              }
            }
            break;
          case ARCHIVE:
            if(!inPreviewMode) {
              Path fileArchive = Paths.get(archiveDir, file.getPath().toString());
              if (fileArchive == null) {
                throw new IOException("Could not find archive file");
              }
              try {
                Files.createDirectories(fileArchive.getParent());
                Files.move(file.getPath(), fileArchive);
                LOG.debug("File '{}' processing completed, post processing action 'ARCHIVED' as", file);
              } catch (IOException ex) {
                throw new IOException(Utils.format("Could not archive '{}': {}", file, ex.toString()), ex);
              }
            }
            break;
        }
      }
    } else {
      setStartingCurrentFileName(currentFile);
      setStartingOffset(getReader().getOffset());
    }
  }
public void releaseReader(boolean inErrorDiscardReader) throws IOException {
    Utils.checkState(open, "FileContext is closed");
    // update starting offsets for next invocation either cold (no reader) or hot (reader)
    boolean hasNext;
    try {
      hasNext = reader != null && reader.hasNext();
    } catch (IOException ex) {
      IOUtils.closeQuietly(reader);
      reader = null;
      hasNext = false;
    }
    boolean doneWithFile = !hasNext || inErrorDiscardReader;
    if (doneWithFile) {
      IOUtils.closeQuietly(reader);
      reader = null;
      // Using Long.MAX_VALUE to signal we reach the end of the file and next iteration should get the next file.
      setStartingCurrentFileName(currentFile);
      setStartingOffset(Long.MAX_VALUE);

      // If we failed to open the file in first place, it will be null and hence we won't do anything with it.
      if(currentFile == null) {
        return;
      }

      // File end event
      LiveFile file = currentFile.refresh();

      if (inErrorDiscardReader) {
        LOG.warn("Processing file '{}' produced an error, skipping '{}' post processing on that file",
                 file, postProcessing);
        eventPublisher.publish(new FileEvent(file, FileEvent.Action.ERROR));
      } else {
        eventPublisher.publish(new FileEvent(file, FileEvent.Action.END));
        switch (postProcessing) {
          case NONE:
            LOG.debug("File '{}' processing completed, post processing action 'NONE'", file);
            break;
          case DELETE:
            if(!inPreviewMode) {
              try {
                Files.delete(file.getPath());
                LOG.debug("File '{}' processing completed, post processing action 'DELETED'", file);
              } catch (IOException ex) {
                throw new IOException(Utils.format("Could not delete '{}': {}", file, ex.toString()), ex);
              }
            }
            break;
          case ARCHIVE:
            if(!inPreviewMode) {
              Path fileArchive = Paths.get(archiveDir, file.getPath().toString());
              if (fileArchive == null) {
                throw new IOException("Could not find archive file");
              }
              try {
                Files.createDirectories(fileArchive.getParent());
                Files.move(file.getPath(), fileArchive);
                LOG.debug("File '{}' processing completed, post processing action 'ARCHIVED' as", file);
              } catch (IOException ex) {
                throw new IOException(Utils.format("Could not archive '{}': {}", file, ex.toString()), ex);
              }
            }
            break;
        }
      }
    } else {
      setStartingCurrentFileName(currentFile);
      setStartingOffset(getReader().getOffset());
    }
  }
public void releaseReader ( boolean inErrorDiscardReader ) throws IOException { Utils . checkState ( open , "FileContext is closed" ) ; // update starting offsets for next invocation either cold (no reader) or hot (reader) boolean hasNext ; try { hasNext = reader != null && reader . hasNext ( ) ; } catch ( IOException ex ) { IOUtils . closeQuietly ( reader ) ; reader = null ; hasNext = false ; } boolean doneWithFile = ! hasNext || inErrorDiscardReader ; if ( doneWithFile ) { IOUtils . closeQuietly ( reader ) ; reader = null ; // Using Long.MAX_VALUE to signal we reach the end of the file and next iteration should get the next file. setStartingCurrentFileName ( currentFile ) ; setStartingOffset ( Long . MAX_VALUE ) ; // If we failed to open the file in first place, it will be null and hence we won't do anything with it. if ( currentFile == null ) { return ; } // File end event LiveFile file = currentFile . refresh ( ) ; if ( inErrorDiscardReader ) { LOG . warn ( "Processing file '{}' produced an error, skipping '{}' post processing on that file" , file , postProcessing ) ; eventPublisher . publish ( new FileEvent ( file , FileEvent . Action . ERROR ) ) ; } else { eventPublisher . publish ( new FileEvent ( file , FileEvent . Action . END ) ) ; switch ( postProcessing ) { case NONE : LOG . debug ( "File '{}' processing completed, post processing action 'NONE'" , file ) ; break ; case DELETE : if ( ! inPreviewMode ) { try { Files . delete ( file . getPath ( ) ) ; LOG . debug ( "File '{}' processing completed, post processing action 'DELETED'" , file ) ; } catch ( IOException ex ) { throw new IOException ( Utils . format ( "Could not delete '{}': {}" , file , ex . toString ( ) ) , ex ) ; } } break ; case ARCHIVE : if ( ! inPreviewMode ) { Path fileArchive = Paths . get ( archiveDir , file . getPath ( ) . toString ( ) ) ; if ( fileArchive == null ) { throw new IOException ( "Could not find archive file" ) ; } try { Files . createDirectories ( fileArchive . getParent ( ) ) ; Files . move ( file . getPath ( ) , fileArchive ) ; LOG . debug ( "File '{}' processing completed, post processing action 'ARCHIVED' as" , file ) ; } catch ( IOException ex ) { throw new IOException ( Utils . format ( "Could not archive '{}': {}" , file , ex . toString ( ) ) , ex ) ; } } break ; } } } else { setStartingCurrentFileName ( currentFile ) ; setStartingOffset ( getReader ( ) . getOffset ( ) ) ; } }
updates reader and offsets after a read.
updates reader and offsets after a read .