close extractor read stream
update high watermark
972984
@Override
  public void close() {
    log.info("Updating the current state high water mark with " + this.highWatermark);
    this.workUnitState.setActualHighWatermark(new LongWatermark(this.highWatermark));
    try {
      this.closeConnection();
    } catch (Exception e) {
      log.error("Failed to close the extractor", e);
    }
  }
18854
private static File extract(String name, InputStream is) throws IOException {
        byte[] buf = new byte[4096];
        int len;

        File lib = File.createTempFile(name, "lib");
        FileOutputStream os = new FileOutputStream(lib);

        try {
            while ((len = is.read(buf)) > 0) {
                os.write(buf, 0, len);
            }
        } catch (IOException e) {
            lib.delete();
            throw e;
        } finally {
            os.close();
            is.close();
        }

        return lib;
    }
1185984
def muted(*streams):
  
  devnull = open(os.devnull, 'w')
  try:
    old_streams = [os.dup(s.fileno()) for s in streams]
    for s in streams:
      os.dup2(devnull.fileno(), s.fileno())
    yield
  finally:
    for o,n in zip(old_streams, streams):
      os.dup2(o, n.fileno())
    devnull.close()
975652
private static int getUpdatedInterval(int inputInterval, ExtractType extractType, WatermarkType watermarkType) {
    LOG.debug("Getting updated interval");
    if ((extractType == ExtractType.SNAPSHOT && watermarkType == WatermarkType.DATE)) {
      return inputInterval * 24;
    } else if (extractType == ExtractType.APPEND_DAILY) {
      return (inputInterval < 1 ? 1 : inputInterval) * 24;
    } else {
      return inputInterval;
    }
  }
1043818
def WriteFileChunk(self, chunk):
    """"""
    precondition.AssertType(chunk, bytes)

    if not self._stream:
      raise ArchiveAlreadyClosedError(
          "Attempting to write to a ZIP archive that was already closed.")

    self.cur_file_size += len(chunk)
    # TODO(user):pytype: crc32 is not visible outside of zipfile.
    # pytype: disable=module-attr
    self.cur_crc = zipfile.crc32(chunk, self.cur_crc) & 0xffffffff
    # pytype: enable=module-attr

    if self.cur_cmpr:
      chunk = self.cur_cmpr.compress(chunk)
      self.cur_compress_size += len(chunk)

    self._stream.write(chunk)
    return self._stream.GetValueAndReset()
693318
public Extractor<S, D> build() throws ExtractPrepareException {
    String watermarkColumn = this.workUnitState.getProp(ConfigurationKeys.EXTRACT_DELTA_FIELDS_KEY);
    long lwm = partition.getLowWatermark();
    long hwm = partition.getHighWatermark();
    log.info("Low water mark: " + lwm + "; and High water mark: " + hwm);

    WatermarkType watermarkType;
    if (StringUtils.isBlank(this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_WATERMARK_TYPE))) {
      watermarkType = null;
    } else {
      watermarkType = WatermarkType
          .valueOf(this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_WATERMARK_TYPE).toUpperCase());
    }

    log.info("Source Entity is " + this.entity);
    try {
      this.setTimeOut(
          this.workUnitState.getPropAsInt(ConfigurationKeys.SOURCE_CONN_TIMEOUT, ConfigurationKeys.DEFAULT_CONN_TIMEOUT));
      this.extractMetadata(this.schema, this.entity, this.workUnit);

      if (StringUtils.isNotBlank(watermarkColumn)) {
        if (partition.isLastPartition()) {
          // Get a more accurate high watermark from the source
          long adjustedHighWatermark = this.getLatestWatermark(watermarkColumn, watermarkType, lwm, hwm);
          log.info("High water mark from source: " + adjustedHighWatermark);
          // If the source reports a finer high watermark, then consider the same as runtime high watermark.
          // Else, consider the low watermark as high water mark(with no delta).i.e, don't move the pointer
          if (adjustedHighWatermark == ConfigurationKeys.DEFAULT_WATERMARK_VALUE) {
            adjustedHighWatermark = getLowWatermarkWithNoDelta(lwm);
          }
          this.highWatermark = adjustedHighWatermark;
        } else {
          this.highWatermark = hwm;
        }

        log.info("High water mark for the current run: " + highWatermark);
        this.setRangePredicates(watermarkColumn, watermarkType, lwm, highWatermark);
      }

      // if it is set to true, skip count calculation and set source count to -1
      if (!Boolean.valueOf(this.workUnitState.getProp(ConfigurationKeys.SOURCE_QUERYBASED_SKIP_COUNT_CALC))) {
        this.sourceRecordCount = this.getSourceCount(this.schema, this.entity, this.workUnit, this.predicateList);
      } else {
        log.info("Skip count calculation");
        this.sourceRecordCount = -1;
      }

      if (this.sourceRecordCount == 0) {
        log.info("Record count is 0; Setting fetch status to false to skip readRecord()");
        this.setFetchStatus(false);
      }
    } catch (SchemaException e) {
      throw new ExtractPrepareException("Failed to get schema for this object; error - " + e.getMessage(), e);
    } catch (HighWatermarkException e) {
      throw new ExtractPrepareException("Failed to get high watermark; error - " + e.getMessage(), e);
    } catch (RecordCountException e) {
      throw new ExtractPrepareException("Failed to get record count; error - " + e.getMessage(), e);
    } catch (Exception e) {
      throw new ExtractPrepareException("Failed to prepare the extract build; error - " + e.getMessage(), e);
    }
    return this;
  }
1215624
def close(self):
    """"""
    self.acquire()
    try:
      self.flush()
      try:
        # Do not close the stream if it's sys.stderr|stdout. They may be
        # redirected or overridden to files, which should be managed by users
        # explicitly.
        if self.stream not in (sys.stderr, sys.stdout) and (
            not hasattr(self.stream, 'isatty') or not self.stream.isatty()):
          self.stream.close()
      except ValueError:
        # A ValueError is thrown if we try to run isatty() on a closed file.
        pass
      super(PythonHandler, self).close()
    finally:
      self.release()
1036437
void cleanup(Http2Stream stream, EmbeddedChannel compressor) {
        if (compressor.finish()) {
            for (;;) {
                final ByteBuf buf = compressor.readOutbound();
                if (buf == null) {
                    break;
                }

                buf.release();
            }
        }
        stream.removeProperty(propertyKey);
    }
633768
def _updateStream(self, xref=0, stream=None, new=0):
        """"""
        if self.isClosed or self.isEncrypted:
            raise ValueError("operation illegal for closed / encrypted doc")

        return _fitz.Document__updateStream(self, xref, stream, new)
716216
function Parser_findDCTDecodeInlineStreamEnd(stream) {
      var startPos = stream.pos, foundEOI = false, b, markerLength, length;
      while ((b = stream.getByte()) !== -1) {
        if (b !== 0xFF) { // Not a valid marker.
          continue;
        }
        switch (stream.getByte()) {
          case 0x00: // Byte stuffing.
            // 0xFF00 appears to be a very common byte sequence in JPEG images.
            break;

          case 0xFF: // Fill byte.
            // Avoid skipping a valid marker, resetting the stream position.
            stream.skip(-1);
            break;

          case 0xD9: // EOI
            foundEOI = true;
            break;

          case 0xC0: // SOF0
          case 0xC1: // SOF1
          case 0xC2: // SOF2
          case 0xC3: // SOF3

          case 0xC5: // SOF5
          case 0xC6: // SOF6
          case 0xC7: // SOF7

          case 0xC9: // SOF9
          case 0xCA: // SOF10
          case 0xCB: // SOF11

          case 0xCD: // SOF13
          case 0xCE: // SOF14
          case 0xCF: // SOF15

          case 0xC4: // DHT
          case 0xCC: // DAC

          case 0xDA: // SOS
          case 0xDB: // DQT
          case 0xDC: // DNL
          case 0xDD: // DRI
          case 0xDE: // DHP
          case 0xDF: // EXP

          case 0xE0: // APP0
          case 0xE1: // APP1
          case 0xE2: // APP2
          case 0xE3: // APP3
          case 0xE4: // APP4
          case 0xE5: // APP5
          case 0xE6: // APP6
          case 0xE7: // APP7
          case 0xE8: // APP8
          case 0xE9: // APP9
          case 0xEA: // APP10
          case 0xEB: // APP11
          case 0xEC: // APP12
          case 0xED: // APP13
          case 0xEE: // APP14
          case 0xEF: // APP15

          case 0xFE: // COM
            // The marker should be followed by the length of the segment.
            markerLength = stream.getUint16();
            if (markerLength > 2) {
              // |markerLength| contains the byte length of the marker segment,
              // including its own length (2 bytes) and excluding the marker.
              stream.skip(markerLength - 2); // Jump to the next marker.
            } else {
              // The marker length is invalid, resetting the stream position.
              stream.skip(-2);
            }
            break;
        }
        if (foundEOI) {
          break;
        }
      }
      length = stream.pos - startPos;
      if (b === -1) {
        warn('Inline DCTDecode image stream: ' +
             'EOI marker not found, searching for /EI/ instead.');
        stream.skip(-length); // Reset the stream position.
        return this.findDefaultInlineStreamEnd(stream);
      }
      this.inlineStreamSkipEI(stream);
      return length;
    }
1238734
function ChunkStream(options) {
  stream.Stream.call(this);
  this.writable = this.readable = true;

  if (!options) {
    options = {};
  }

  this._highWaterMark = options.highWaterMark || bufferSize;

  this._paused = undefined; //True/false is the external status from users.

  this._isStreamOpened = false;
  this._offset = 0;
  this._allocator = options.allocator;
  this._streamEnded = false;
  this._md5hash = null;
  this._buffer = null;
  this._internalBufferSize = 0;
  this._outputLengthLimit = 0;
  this._md5sum = undefined;

  if (options.calcContentMd5) {
    this._md5hash = new Md5Wrapper().createMd5Hash();
  }
}
1121577
private void copy(InputStream input, OutputStream output) throws IOException {
    try {
      byte[] buffer = new byte[BUFFER_SIZE];
      int bytesRead = input.read(buffer);
      while (bytesRead != -1) {
        output.write(buffer, 0, bytesRead);
        bytesRead = input.read(buffer);
      }
    } finally {
      input.close();
      output.close();
    }
  }
920682
def mutate_stream (instream, outstream):
    """"""

    chunk = None
    key = None
    data = None
    misclines = []

    for fullline in instream:
        line = fullline.split ('#', 1)[0]

        m = sectionre.match (line)
        if m is not None:
            # New chunk
            if chunk is not None:
                if key is not None:
                    chunk.data.set_one (key, data.strip ().decode ('utf8'))
                    key = data = None
                yield chunk
                chunk.emit (outstream)

            chunk = FileChunk ()
            for miscline in misclines:
                chunk._addLine (miscline, None)
            misclines = []
            chunk.data.section = m.group (1)
            chunk._addLine (fullline, None)
            continue

        if len (line.strip ()) == 0:
            if key is not None:
                chunk.data.set_one (key, data.strip ().decode ('utf8'))
                key = data = None
            if chunk is not None:
                chunk._addLine (fullline, None)
            else:
                misclines.append (fullline)
            continue

        m = escre.match (fullline)
        if m is not None:
            if chunk is None:
                raise InifileError ('key seen without section!')
            if key is not None:
                chunk.data.set_one (key, data.strip ().decode ('utf8'))
            key = m.group (1)
            data = m.group (2).replace (r'\"', '"').replace (r'\n', '\n').replace (r'\\', '\\')
            chunk.data.set_one (key, data.decode ('utf8'))
            chunk._addLine (fullline, key)
            key = data = None
            continue

        m = keyre.match (line)
        if m is not None:
            if chunk is None:
                raise InifileError ('key seen without section!')
            if key is not None:
                chunk.data.set_one (key, data.strip ().decode ('utf8'))
            key = m.group (1)
            data = m.group (2)
            if not data[-1].isspace ():
                data += ' '
            chunk._addLine (fullline, key)
            continue

        if line[0].isspace () and key is not None:
            data += line.strip () + ' '
            chunk._addLine (fullline, key)
            continue

        raise InifileError ('unparsable line: ' + line[:-1])

    if chunk is not None:
        if key is not None:
            chunk.data.set_one (key, data.strip ().decode ('utf8'))
        yield chunk
        chunk.emit (outstream)
860347
private long getAppendWatermarkCutoff(ExtractType extractType) {
    LOG.debug("Getting append water mark cutoff");
    long highWatermark = ConfigurationKeys.DEFAULT_WATERMARK_VALUE;
    String timeZone = this.state.getProp(ConfigurationKeys.SOURCE_TIMEZONE);
    AppendMaxLimitType limitType = getAppendLimitType(extractType,
        this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT));
    if (limitType == null) {
      LOG.debug("Limit type is not found");
      return highWatermark;
    }
    int limitDelta =
        getAppendLimitDelta(this.state.getProp(ConfigurationKeys.SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT));
    // if it is CURRENTDATE or CURRENTHOUR then high water mark is current time
    if (limitDelta == 0) {
      highWatermark = Long.parseLong(Utils.dateTimeToString(getCurrentTime(timeZone), WATERMARKTIMEFORMAT, timeZone));
    }
    // if CURRENTDATE or CURRENTHOUR has offset then high water mark is end of day of the given offset
    else {
      int seconds = 3599; // x:59:59
      String format = null;

      switch (limitType) {
        case CURRENTDATE:
          format = "yyyyMMdd";
          limitDelta = limitDelta * 24 * 60 * 60;
          seconds = 86399; // 23:59:59
          break;
        case CURRENTHOUR:
          format = "yyyyMMddHH";
          limitDelta = limitDelta * 60 * 60;
          seconds = 3599; // x:59:59
          break;
        case CURRENTMINUTE:
          format = "yyyyMMddHHmm";
          limitDelta = limitDelta * 60;
          seconds = 59;
          break;
        case CURRENTSECOND:
          format = "yyyyMMddHHmmss";
          seconds = 0;
          break;
        default:
          break;
      }

      DateTime deltaTime = getCurrentTime(timeZone).minusSeconds(limitDelta);
      DateTime previousTime =
          Utils.toDateTime(Utils.dateTimeToString(deltaTime, format, timeZone), format, timeZone).plusSeconds(seconds);
      highWatermark = Long.parseLong(Utils.dateTimeToString(previousTime, WATERMARKTIMEFORMAT, timeZone));

      // User specifies SOURCE_QUERYBASED_APPEND_MAX_WATERMARK_LIMIT
      hasUserSpecifiedHighWatermark = true;
    }
    return highWatermark;
  }
1014085
function copyFile(source, target, copyCallback) {
    const readStream = fs.createReadStream(source);
    const writeStream = fs.createWriteStream(target);
    let done = false;

    readStream.on('error', copyDone);
    writeStream.on('error', copyDone);

    writeStream.on('close', function onWriteCb() {
        copyDone(null);
    });

    readStream.pipe(writeStream);

    /**
     * Finish copying. Reports error when needed
     * @param [error] optional error
     */
    function copyDone(error) {
        if (!done) {
            done = true;
            copyCallback(error);
        }
    }
}
1279913
def flush(self):
    """"""
    self.acquire()
    try:
      self.stream.flush()
    except (EnvironmentError, ValueError):
      # A ValueError is thrown if we try to flush a closed file.
      pass
    finally:
      self.release()
1342799
function extract(src, dst, opts) {
    var extractor;
    var promise;

    opts = opts || {};
    extractor = getExtractor(src);

    // Try to get extractor from mime type
    if (!extractor && opts.mimeType) {
        extractor = getExtractor(opts.mimeType);
    }

    // If extractor is null, then the archive type is unknown
    if (!extractor) {
        return Q.reject(createError('File ' + src + ' is not a known archive', 'ENOTARCHIVE'));
    }

    // Check archive file size
    promise = Q.nfcall(fs.stat, src)
    .then(function (stat) {
        if (stat.size <= 8) {
            throw createError('File ' + src + ' is an invalid archive', 'ENOTARCHIVE');
        }

        // Extract archive
        return extractor(src, dst);
    });

    // TODO: There's an issue here if the src and dst are the same and
    //       The zip name is the same as some of the zip file contents
    //       Maybe create a temp directory inside dst, unzip it there,
    //       unlink zip and then move contents

    // Remove archive
    if (!opts.keepArchive) {
        promise = promise
        .then(function () {
            return Q.nfcall(fs.unlink, src);
        });
    }

    // Move contents if a single directory was extracted
    if (!opts.keepStructure) {
        promise = promise
        .then(function () {
            return isSingleDir(dst);
        })
        .then(function (singleDir) {
            return singleDir ? moveSingleDirContents(singleDir) : null;
        });
    }

    // Resolve promise to the dst dir
    return promise.then(function () {
        return dst;
    });
}