@Test
public void readToByteBuffer() throws IOException {
  for (IntFunction<ByteBuffer> constructor : BUFFER_CONSTRUCTORS) {
    for (int streamLength = 1; streamLength <= 10; streamLength++) {
      for (int bufferCapacity = 0; bufferCapacity <= 10; bufferCapacity++) {
        testReadToByteBuffer(constructor, streamLength, bufferCapacity, 0);
        if (bufferCapacity > 1) {
          testReadToByteBuffer(constructor, streamLength, bufferCapacity, 1);
          if (bufferCapacity > 2) {
            testReadToByteBuffer(constructor, streamLength, bufferCapacity,
                bufferCapacity - 1);
          }
        }
        testReadToByteBuffer(constructor, streamLength, bufferCapacity,
            bufferCapacity);
      }
    }
  }
}
 

/**
 * Collect the elements output from the pipeline stage.
 *
 * @param generator the array generator to be used to create array instances
 * @return a flat array-backed Node that holds the collected output elements
 */
@SuppressWarnings("unchecked")
final Node<E_OUT> evaluateToArrayNode(IntFunction<E_OUT[]> generator) {
    if (linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    linkedOrConsumed = true;

    // If the last intermediate operation is stateful then
    // evaluate directly to avoid an extra collection step
    if (isParallel() && previousStage != null && opIsStateful()) {
        // Set the depth of this, last, pipeline stage to zero to slice the
        // pipeline such that this operation will not be included in the
        // upstream slice and upstream operations will not be included
        // in this slice
        depth = 0;
        return opEvaluateParallel(previousStage, previousStage.sourceSpliterator(0), generator);
    }
    else {
        return evaluate(sourceSpliterator(0), true, generator);
    }
}
 

/**
 * Returns a mapping that shifts a given mapping's source by a given
 * offset.
 *
 * <p>For example, given {@code mapping} with sourceCount=2, targetCount=8,
 * and (source, target) entries {[0: 5], [1: 7]}, offsetSource(mapping, 3)
 * returns a mapping with sourceCount=5, targetCount=8,
 * and (source, target) entries {[3: 5], [4: 7]}.
 *
 * @param mapping     Input mapping
 * @param offset      Offset to be applied to each source
 * @param sourceCount New source count; must be at least {@code mapping}'s
 *                    source count plus {@code offset}
 * @return Shifted mapping
 */
public static TargetMapping offsetSource(
    final TargetMapping mapping, final int offset, final int sourceCount) {
  if (sourceCount < mapping.getSourceCount() + offset) {
    throw new IllegalArgumentException("new source count too low");
  }
  return target(
      (IntFunction<Integer>) source -> {
        int source2 = source - offset;
        return source2 < 0 || source2 >= mapping.getSourceCount()
            ? null
            : mapping.getTargetOpt(source2);
      },
      sourceCount,
      mapping.getTargetCount());
}
 

@Override
public <P_IN> Node<Long> opEvaluateParallel(PipelineHelper<Long> helper,
                                            Spliterator<P_IN> spliterator,
                                            IntFunction<Long[]> generator) {
    if (StreamOpFlag.SORTED.isKnown(helper.getStreamAndOpFlags())) {
        return helper.evaluate(spliterator, false, generator);
    }
    else {
        Node.OfLong n = (Node.OfLong) helper.evaluate(spliterator, true, generator);

        long[] content = n.asPrimitiveArray();
        Arrays.parallelSort(content);

        return Nodes.node(content);
    }
}
 

@Override
public final <U> Stream<U> mapToObj(IntFunction<? extends U> mapper) {
    Objects.requireNonNull(mapper);
    return new ReferencePipeline.StatelessOp<Integer, U>(this, StreamShape.INT_VALUE,
                                                         StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
        @Override
        Sink<Integer> opWrapSink(int flags, Sink<U> sink) {
            return new Sink.ChainedInt<U>(sink) {
                @Override
                public void accept(int t) {
                    downstream.accept(mapper.apply(t));
                }
            };
        }
    };
}
 

@Override
public <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,
                                         Spliterator<P_IN> spliterator,
                                         IntFunction<T[]> generator) {
    // If the input is already naturally sorted and this operation
    // naturally sorts then collect the output
    if (StreamOpFlag.SORTED.isKnown(helper.getStreamAndOpFlags()) && isNaturalSort) {
        return helper.evaluate(spliterator, false, generator);
    }
    else {
        // @@@ Weak two-pass parallel implementation; parallel collect, parallel sort
        T[] flattenedData = helper.evaluate(spliterator, true, generator).asArray(generator);
        Arrays.parallelSort(flattenedData, comparator);
        return Nodes.node(flattenedData);
    }
}
 

@Override
public <P_IN> Node<Integer> opEvaluateParallel(PipelineHelper<Integer> helper,
                                               Spliterator<P_IN> spliterator,
                                               IntFunction<Integer[]> generator) {
    if (StreamOpFlag.SORTED.isKnown(helper.getStreamAndOpFlags())) {
        return helper.evaluate(spliterator, false, generator);
    }
    else {
        Node.OfInt n = (Node.OfInt) helper.evaluate(spliterator, true, generator);

        int[] content = n.asPrimitiveArray();
        Arrays.parallelSort(content);

        return Nodes.node(content);
    }
}
 

@Test
public void readEmptyStreamToByteBuffer() throws IOException {
  for (IntFunction<ByteBuffer> constructor : BUFFER_CONSTRUCTORS) {
    final OzoneFSInputStream subject = createTestSubject(emptyStream());
    final ByteBuffer buf = constructor.apply(1);

    final int bytesRead = subject.read(buf);

    assertEquals(-1, bytesRead);
    assertEquals(0, buf.position());
  }
}
 

@Test
public void bufferPositionUnchangedOnEOF() throws IOException {
  for (IntFunction<ByteBuffer> constructor : BUFFER_CONSTRUCTORS) {
    final OzoneFSInputStream subject = createTestSubject(eofStream());
    final ByteBuffer buf = constructor.apply(123);

    final int bytesRead = subject.read(buf);

    assertEquals(-1, bytesRead);
    assertEquals(0, buf.position());
  }
}
 

@Test(dataProvider = "string")
public void testSetAllString(String name, int size, IntFunction<String> generator, String[] expected) {
    String[] result = new String[size];
    Arrays.setAll(result, generator);
    assertEquals(result, expected, "setAll(String[], IntFunction<String>) case " + name + " failed.");

    // ensure fresh array
    result = new String[size];
    Arrays.parallelSetAll(result, generator);
    assertEquals(result, expected, "parallelSetAll(String[], IntFunction<String>) case " + name + " failed.");
}
 

@Override
@SuppressWarnings("unchecked")
final <P_IN> Node<E_OUT> evaluate(Spliterator<P_IN> spliterator,
                                  boolean flatten,
                                  IntFunction<E_OUT[]> generator) {
    if (isParallel()) {
        // @@@ Optimize if op of this pipeline stage is a stateful op
        return evaluateToNode(this, spliterator, flatten, generator);
    }
    else {
        Node.Builder<E_OUT> nb = makeNodeBuilder(
                exactOutputSizeIfKnown(spliterator), generator);
        return wrapAndCopyInto(nb, spliterator).build();
    }
}
 

private PartitionedConsumption(
        int consumersCount,
        ListenableFuture<?> activator,
        Iterable<Integer> partitionNumbers,
        IntFunction<ListenableFuture<T>> loader,
        IntConsumer disposer)
{
    checkArgument(consumersCount > 0, "consumersCount must be positive");
    this.consumersCount = consumersCount;
    this.partitions = createPartitions(activator, partitionNumbers, loader, disposer);
}
 

public static <T extends Enum<T>> T constructAndAdd(Class<T> clazz, IntFunction<? extends T> constructor) {
	T[] values = clazz.getEnumConstants();
	T instance = constructor.apply(values.length);
	addToValues(values, instance);
	clearCachedValues(clazz);
	return instance;
}
 

@Test(dataProvider = "string")
public void testSetAllString(String name, int size, IntFunction<String> generator, String[] expected) {
    String[] result = new String[size];
    Arrays.setAll(result, generator);
    assertEquals(result, expected, "setAll(String[], IntFunction<String>) case " + name + " failed.");

    // ensure fresh array
    result = new String[size];
    Arrays.parallelSetAll(result, generator);
    assertEquals(result, expected, "parallelSetAll(String[], IntFunction<String>) case " + name + " failed.");
}
 

@Override
final <P_IN> Node<Integer> evaluateToNode(PipelineHelper<Integer> helper,
                                          Spliterator<P_IN> spliterator,
                                          boolean flattenTree,
                                          IntFunction<Integer[]> generator) {
    return Nodes.collectInt(helper, spliterator, flattenTree);
}
 

@Override
final <P_IN> Node<Long> evaluateToNode(PipelineHelper<Long> helper,
                                       Spliterator<P_IN> spliterator,
                                       boolean flattenTree,
                                       IntFunction<Long[]> generator) {
    return Nodes.collectLong(helper, spliterator, flattenTree);
}
 

private static DataTypeTest unicodeTests(IntFunction<DataType<String>> typeConstructor, ToIntFunction<String> stringLength, int maxSize)
{
    String unicodeText = "攻殻機動隊";
    String nonBmpCharacter = "\ud83d\ude02";
    int unicodeLength = stringLength.applyAsInt(unicodeText);
    int nonBmpLength = stringLength.applyAsInt(nonBmpCharacter);

    return DataTypeTest.create()
            .addRoundTrip(typeConstructor.apply(unicodeLength), unicodeText)
            .addRoundTrip(typeConstructor.apply(unicodeLength + 8), unicodeText)
            .addRoundTrip(typeConstructor.apply(maxSize), unicodeText)
            .addRoundTrip(typeConstructor.apply(nonBmpLength), nonBmpCharacter)
            .addRoundTrip(typeConstructor.apply(nonBmpLength + 5), nonBmpCharacter);
}
 

@Override
public <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,
                                         Spliterator<P_IN> spliterator,
                                         IntFunction<T[]> generator) {
    int flags = helper.getStreamAndOpFlags();

    Assert.assertTrue(StreamOpFlag.SIZED.isKnown(flags));
    return super.opEvaluateParallel(helper, spliterator, generator);
}
 

@Override
final <P_IN> Node<Integer> evaluateToNode(PipelineHelper<Integer> helper,
                                          Spliterator<P_IN> spliterator,
                                          boolean flattenTree,
                                          IntFunction<Integer[]> generator) {
    return Nodes.collectInt(helper, spliterator, flattenTree);
}
 

SliceTask(AbstractPipeline<P_OUT, P_OUT, ?> op,
          PipelineHelper<P_OUT> helper,
          Spliterator<P_IN> spliterator,
          IntFunction<P_OUT[]> generator,
          long offset, long size) {
    super(helper, spliterator);
    this.op = op;
    this.generator = generator;
    this.targetOffset = offset;
    this.targetSize = size;
}
 

@Test
public void testSingleFieldSerializer() {
	TEST_FIELD_SERIALIZERS.forEach(t -> {
		@SuppressWarnings("unchecked")
		TypeSerializer<Object>[] fieldSerializers = new TypeSerializer[] { t.f0 };
		List<Object>[] instances = Arrays.stream(t.f1)
			.map(Arrays::asList)
			.toArray((IntFunction<List<Object>[]>) List[]::new);
		runTests(t.f0.getLength(), fieldSerializers, instances);
	});
}
 

@NonNull
public Label buildExpectedLabel(IntFunction<Val<Integer>> numViolationsPerLine, int idx) {
    Label foo = new Label();
    foo.getStyleClass().addAll("num-violations-gutter-label");
    Val<Integer> num = numViolationsPerLine.apply(idx + 1);
    foo.textProperty().bind(num.map(Object::toString));
    foo.setTooltip(new Tooltip("Number of violations expected on this line"));
    foo.visibleProperty().bind(num.map(it -> it > 0));
    return foo;
}
 

@Override
final <P_IN> Node<Double> evaluateToNode(PipelineHelper<Double> helper,
                                         Spliterator<P_IN> spliterator,
                                         boolean flattenTree,
                                         IntFunction<Double[]> generator) {
    return Nodes.collectDouble(helper, spliterator, flattenTree);
}
 

@Override
public T[] asArray(IntFunction<T[]> generator) {
    long size = count();
    if (size >= MAX_ARRAY_SIZE)
        throw new IllegalArgumentException(BAD_SIZE);
    T[] array = generator.apply((int) size);
    copyInto(array, 0);
    return array;
}
 

@Override
final <P_IN> Node<P_OUT> evaluateToNode(PipelineHelper<P_OUT> helper,
                                    Spliterator<P_IN> spliterator,
                                    boolean flattenTree,
                                    IntFunction<P_OUT[]> generator) {
    return Nodes.collect(helper, spliterator, flattenTree, generator);
}
 

@Override
@SuppressWarnings("unchecked")
public final <A> A[] toArray(IntFunction<A[]> generator) {
    // Since A has no relation to U (not possible to declare that A is an upper bound of U)
    // there will be no static type checking.
    // Therefore use a raw type and assume A == U rather than propagating the separation of A and U
    // throughout the code-base.
    // The runtime type of U is never checked for equality with the component type of the runtime type of A[].
    // Runtime checking will be performed when an element is stored in A[], thus if A is not a
    // super type of U an ArrayStoreException will be thrown.
    @SuppressWarnings("rawtypes")
    IntFunction rawGenerator = (IntFunction) generator;
    return (A[]) Nodes.flatten(evaluateToArrayNode(rawGenerator), rawGenerator)
                          .asArray(rawGenerator);
}
 

@Override
public T[] asArray(IntFunction<T[]> generator) {
    long size = count();
    if (size >= MAX_ARRAY_SIZE)
        throw new IllegalArgumentException(BAD_SIZE);
    T[] array = generator.apply((int) size);
    copyInto(array, 0);
    return array;
}
 

/**
 * {@inheritDoc}
 *
 * @implSpec the default implementation invokes the generator to create
 * an instance of a boxed primitive array with a length of
 * {@link #count()} and then invokes {@link #copyInto(T[], int)} with
 * that array at an offset of 0.
 */
@Override
default T[] asArray(IntFunction<T[]> generator) {
    if (java.util.stream.Tripwire.ENABLED)
        java.util.stream.Tripwire.trip(getClass(), "{0} calling Node.OfPrimitive.asArray");

    long size = count();
    if (size >= Nodes.MAX_ARRAY_SIZE)
        throw new IllegalArgumentException(Nodes.BAD_SIZE);
    T[] boxed = generator.apply((int) count());
    copyInto(boxed, 0);
    return boxed;
}
 

/**
 * Create a new array using the specified array factory, and copy the
 * elements into it.
 */
public E[] asArray(IntFunction<E[]> arrayFactory) {
    long size = count();
    if (size >= Nodes.MAX_ARRAY_SIZE)
        throw new IllegalArgumentException(Nodes.BAD_SIZE);
    E[] result = arrayFactory.apply((int) size);
    copyInto(result, 0);
    return result;
}
 

@SuppressWarnings("unchecked")
ArrayNode(long size, IntFunction<T[]> generator) {
    if (size >= MAX_ARRAY_SIZE)
        throw new IllegalArgumentException(BAD_SIZE);
    this.array = generator.apply((int) size);
    this.curSize = 0;
}