* writer each time it is called.
   *
   * <p>The caller is responsible for ensuring that the returned writer is closed.
   *
   * @throws IOException if an I/O error occurs while opening the writer
   */
  public abstract Writer openStream() throws IOException;

  /**
   * Opens a new buffered {@link Writer} for writing to this sink. The returned stream is not

    // NPE on null listener, so we propagate that contract up into the add method as well.
    checkNotNull(runnable, "Runnable was null.");
    checkNotNull(executor, "Executor was null.");

    // Lock while we check state. We must maintain the lock while adding the new pair so that
    // another thread can't run the list out from under us. We only add to the list if we have not
    // yet started execution.
    synchronized (this) {
      if (!executed) {

        reader.setScrollSize(1000);
        int count = 0;
        Set<String> valueSet = Collections.synchronizedSet(new HashSet<>());
        Map<String, Object> source;
        while ((source = reader.read()) != null) {
            assertTrue(source.get("field1").toString().startsWith("test"));
            valueSet.add(source.get("field1").toString());
            count++;
        }

    // Now process directly from the rest of the input buffer
    while (readBuffer.remaining() >= chunkSize) {
      process(readBuffer);
    }

    // Finally stick the remainder back in our usual buffer
    buffer.put(readBuffer);
    return this;
  }

  /*
   * Note: hashString(CharSequence, Charset) is intentionally not overridden.
   *

 *   <li>It is easy for the user to ensure that listeners are never invoked while holding locks.
 * </ul>
 *
 * The last point is subtle. Often the observable object will be managing its own internal state
 * using a lock, however it is dangerous to dispatch listeners while holding a lock because they
 * might run on the {@code directExecutor()} or be otherwise re-entrant (call back into your

   * limits.
   */
  fun encode(string: String): String? {
    var pos = 0
    val limit = string.length
    val result = Buffer()

    while (pos < limit) {
      var dot = string.indexOf('.', startIndex = pos)
      if (dot == -1) dot = limit

      if (!encodeLabel(string, pos, dot, result)) {
        // If we couldn't encode the label, give up.

        if (low < 0) {
            low = 0;
        }
        if (hi >= ctype2.length) {
            hi = ctype2.length - 1;
        }
        while (low <= hi) {
            ctype2[low++] |= CT_ALPHA;
        }
    }

    /**
     * Sets a character as a word character.
     *
     * @param ctype2
     *            The array of character types.

 */
internal fun mergeAdjacentDeltaMappedRanges(ranges: MutableList<MappedRange>): MutableList<MappedRange> {
  var i = 0
  while (i < ranges.size) {
    val curr = ranges[i]
    if (curr is MappedRange.InlineDelta) {
      val j = i + 1
      mergeAdjacent@ while (j < ranges.size) {
        val next = ranges[j]
        if (next is MappedRange.InlineDelta &&
          curr.codepointDelta == next.codepointDelta

    // NPE on null listener, so we propagate that contract up into the add method as well.
    checkNotNull(runnable, "Runnable was null.");
    checkNotNull(executor, "Executor was null.");

    // Lock while we check state. We must maintain the lock while adding the new pair so that
    // another thread can't run the list out from under us. We only add to the list if we have not
    // yet started execution.
    synchronized (this) {
      if (!executed) {

   * independent stream each time it is called.
   *
   * <p>The caller is responsible for ensuring that the returned stream is closed.
   *
   * @throws IOException if an I/O error occurs while opening the stream
   */
  public abstract OutputStream openStream() throws IOException;

  /**
   * Opens a new buffered {@link OutputStream} for writing to this sink. The returned stream is not