Java 类java.util.function.LongConsumer 实例源码

@Override
public boolean tryAdvance(LongConsumer consumer) {
    Objects.requireNonNull(consumer);

    final long i = from;
    if (i < upTo) {
        from++;
        consumer.accept(i);
        return true;
    }
    else if (last > 0) {
        last = 0;
        consumer.accept(i);
        return true;
    }
    return false;
}

@Override
public final LongStream peek(LongConsumer action) {
    Objects.requireNonNull(action);
    return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
                                 0) {
        @Override
        Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
            return new Sink.ChainedLong<Long>(sink) {
                @Override
                public void accept(long t) {
                    action.accept(t);
                    downstream.accept(t);
                }
            };
        }
    };
}

/**
 * Returns an infinite sequential ordered {@code LongStream} produced by iterative
 * application of a function {@code f} to an initial element {@code seed},
 * producing a {@code Stream} consisting of {@code seed}, {@code f(seed)},
 * {@code f(f(seed))}, etc.
 *
 * <p>The first element (position {@code 0}) in the {@code LongStream} will
 * be the provided {@code seed}.  For {@code n > 0}, the element at position
 * {@code n}, will be the result of applying the function {@code f} to the
 * element at position {@code n - 1}.
 *
 * <p>The action of applying {@code f} for one element
 * <a href="../concurrent/package-summary.html#MemoryVisibility"><i>happens-before</i></a>
 * the action of applying {@code f} for subsequent elements.  For any given
 * element the action may be performed in whatever thread the library
 * chooses.
 *
 * @param seed the initial element
 * @param f a function to be applied to the previous element to produce
 *          a new element
 * @return a new sequential {@code LongStream}
 */
public static LongStream iterate(final long seed, final LongUnaryOperator f) {
    Objects.requireNonNull(f);
    Spliterator.OfLong spliterator = new Spliterators.AbstractLongSpliterator(Long.MAX_VALUE,
           Spliterator.ORDERED | Spliterator.IMMUTABLE | Spliterator.NONNULL) {
        long prev;
        boolean started;

        @Override
        public boolean tryAdvance(LongConsumer action) {
            Objects.requireNonNull(action);
            long t;
            if (started)
                t = f.applyAsLong(prev);
            else {
                t = seed;
                started = true;
            }
            action.accept(prev = t);
            return true;
        }
    };
    return StreamSupport.longStream(spliterator, false);
}

@Override
public Array<T> forEachLong(LongConsumer consumer) {
    final int length = length();
    if (isParallel() && length > 0) {
        final int processors = Runtime.getRuntime().availableProcessors();
        final int splitThreshold = parallel ? Math.max(length() / processors, 10000) : Integer.MAX_VALUE;
        final ForEach action = new ForEach(0, length - 1, splitThreshold, consumer);
        ForkJoinPool.commonPool().invoke(action);
    } else {
        for (int i=0; i<length; ++i) {
            final long value = getLong(i);
            consumer.accept(value);
        }
    }
    return this;
}

@Override
protected final void computeUserRanking(final Set<RunFile> runs, final long userID,
        final LongSet candidateItems, final TreeMap<Double, LongSet> ranking) {

    candidateItems.forEach((LongConsumer) itemID -> {

        final MutableInt n = new MutableInt(0);
        final MutableDouble accum = new MutableDouble(0.0);

        runs.forEach(run -> {
            final double score = run.getScore(userID, itemID, Double.NaN);
            if (!Double.isNaN(score)) {
                n.increment();
                accum.add(score);
            }
        });

        if (n.get() > 0) {
            saveScore(ranking, itemID, computeScore(n.get(), accum.get()));
        }

    });

}

public void testLongForEachRemainingWithNull() {
    PrimitiveIterator.OfLong i = new PrimitiveIterator.OfLong() {
        @Override
        public long nextLong() {
            return 0;
        }

        @Override
        public boolean hasNext() {
            return false;
        }
    };

    executeAndCatch(() -> i.forEachRemaining((LongConsumer) null));
    executeAndCatch(() -> i.forEachRemaining((Consumer<Long>) null));
}

public void testLongForEachRemainingWithNull() {
    PrimitiveIterator.OfLong i = new PrimitiveIterator.OfLong() {
        @Override
        public long nextLong() {
            return 0;
        }

        @Override
        public boolean hasNext() {
            return false;
        }
    };

    assertThrowsNPE(() -> i.forEachRemaining((LongConsumer) null));
    assertThrowsNPE(() -> i.forEachRemaining((Consumer<Long>) null));
}

@Override
public boolean tryAdvance(LongConsumer consumer) {
    Objects.requireNonNull(consumer);

    final long i = from;
    if (i < upTo) {
        from++;
        consumer.accept(i);
        return true;
    }
    else if (last > 0) {
        last = 0;
        consumer.accept(i);
        return true;
    }
    return false;
}

@Override
public void forEachRemaining(LongConsumer consumer) {
    Objects.requireNonNull(consumer);

    long i = from;
    final long hUpTo = upTo;
    int hLast = last;
    from = upTo;
    last = 0;
    while (i < hUpTo) {
        consumer.accept(i++);
    }
    if (hLast > 0) {
        // Last element of closed range
        consumer.accept(i);
    }
}

@Override
public boolean tryAdvance(LongConsumer consumer) {
    Objects.requireNonNull(consumer);

    final long i = from;
    if (i < upTo) {
        from++;
        consumer.accept(i);
        return true;
    }
    else if (last > 0) {
        last = 0;
        consumer.accept(i);
        return true;
    }
    return false;
}

@Override
public final LongStream peek(LongConsumer action) {
    Objects.requireNonNull(action);
    return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
                                 0) {
        @Override
        Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
            return new Sink.ChainedLong<Long>(sink) {
                @Override
                public void accept(long t) {
                    action.accept(t);
                    downstream.accept(t);
                }
            };
        }
    };
}

@Test(dataProvider = "LongStreamTestData", dataProviderClass = LongStreamTestDataProvider.class)
public void testLongOps(String name, final TestData.OfLong data) {
    class RecordingConsumer extends AbstractRecordingConsumer<Long> implements LongConsumer {
        public void accept(long t) {
            list.add(t);
        }
    }
    final RecordingConsumer b = new RecordingConsumer();

    withData(data)
            .stream(s -> s.peek(b))
            .before(b::before)
            .after(b::after)
            .exercise();
}

@Override
public boolean tryAdvance(LongConsumer action) {
    Objects.requireNonNull(action);

    action.accept(s.getAsLong());
    return true;
}

/**
 * Fuse the specified combination of runs and print the result.
 *
 * @param fold
 *            the fold
 * @param runs
 *            the runs of the current combination
 * @param allUsers
 *            all the users in the current combination
 * @param outputFile
 *            the path to the output file
 */
protected final void fuseAndPrint(final int fold, final Set<RunFile> runs,
        final LongSet allUsers, final Path outputFile) {

    try (final PrintWriter writer = new PrintWriter(Files.newBufferedWriter(outputFile))) {

        final LongSet all = HashLongSets.newUpdatableSet(maxRank);
        runs.forEach(run -> {
            all.addAll(run.getItems());
        });

        allUsers.forEach((LongConsumer) userID -> {
            final TreeMap<Double, LongSet> ranking = new TreeMap<Double, LongSet>(
                    Collections.reverseOrder());

            final LongSet candidateItems = HashLongSets.newUpdatableSet(maxRank);
            runs.forEach(run -> {
                candidateItems.addAll(run.getRanking(userID).keySet());
            });

            computeUserRanking(runs, userID, candidateItems, ranking);
            printRanking(userID, ranking, writer);
        });

    } catch (final IOException e) {
        throw new RuntimeException(e);
    }

}

@Override
public boolean tryAdvance(LongConsumer action) {
    Objects.requireNonNull(action);

    if (count == -2) {
        action.accept(first);
        count = -1;
        return true;
    }
    else {
        return false;
    }
}

@Override
public void forEachRemaining(LongConsumer action) {
    Objects.requireNonNull(action);

    if (count == -2) {
        action.accept(first);
        count = -1;
    }
}

/**
 * Adapt a {@code Sink<Long> to an {@code LongConsumer}, ideally simply
 * by casting.
 */
private static LongConsumer adapt(Sink<Long> sink) {
    if (sink instanceof LongConsumer) {
        return (LongConsumer) sink;
    } else {
        if (Tripwire.ENABLED)
            Tripwire.trip(AbstractPipeline.class,
                          "using LongStream.adapt(Sink<Long> s)");
        return sink::accept;
    }
}

@Override
public void forEach(LongConsumer action) {
    if (!isParallel()) {
        adapt(sourceStageSpliterator()).forEachRemaining(action);
    } else {
        super.forEach(action);
    }
}

@Override
public void forEachOrdered(LongConsumer action) {
    if (!isParallel()) {
        adapt(sourceStageSpliterator()).forEachRemaining(action);
    } else {
        super.forEachOrdered(action);
    }
}

@Override
public void forEach(Consumer<? super Long> consumer) {
    if (consumer instanceof LongConsumer) {
        forEach((LongConsumer) consumer);
    }
    else {
        if (Tripwire.ENABLED)
            Tripwire.trip(getClass(), "{0} calling SpinedBuffer.OfLong.forEach(Consumer)");
        spliterator().forEachRemaining(consumer);
    }
}

@Override
protected void arrayForEach(long[] array,
                            int from, int to,
                            LongConsumer consumer) {
    for (int i = from; i < to; i++)
        consumer.accept(array[i]);
}

@Override
public boolean tryAdvance(LongConsumer consumer) {
    Objects.requireNonNull(consumer);
    boolean hasNext = doAdvance();
    if (hasNext)
        consumer.accept(buffer.get(nextToConsume));
    return hasNext;
}

@Override
public void forEachRemaining(LongConsumer consumer) {
    if (buffer == null && !finished) {
        Objects.requireNonNull(consumer);
        init();

        ph.wrapAndCopyInto((Sink.OfLong) consumer::accept, spliterator);
        finished = true;
    }
    else {
        do { } while (tryAdvance(consumer));
    }
}

@Override
public final LongStream flatMapToLong(Function<? super P_OUT, ? extends LongStream> mapper) {
    Objects.requireNonNull(mapper);
    // We can do better than this, by polling cancellationRequested when stream is infinite
    return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
                                               StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
            return new Sink.ChainedReference<P_OUT, Long>(sink) {
                LongConsumer downstreamAsLong = downstream::accept;
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    try (LongStream result = mapper.apply(u)) {
                        // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                        if (result != null)
                            result.sequential().forEach(downstreamAsLong);
                    }
                }
            };
        }
    };
}

@Override
public boolean tryAdvance(LongConsumer action) {
    Objects.requireNonNull(action);

    action.accept(s.getAsLong());
    return true;
}

@Test(dataProvider = "LongStreamTestData", dataProviderClass = LongStreamTestDataProvider.class)
public void testLongOps(String name, final TestData.OfLong data) {
    class RecordingConsumer extends AbstractRecordingConsumer<Long> implements LongConsumer {
        public void accept(long t) {
            list.add(t);
        }
    }
    final RecordingConsumer b = new RecordingConsumer();

    withData(data)
            .stream(s -> s.peek(b))
            .before(b::before)
            .after(b::after)
            .exercise();
}

/**
 * {@inheritDoc}
 *
 * @param consumer A {@code Consumer} that is to be invoked with each
 *        element in this {@code Node}.  If this is an
 *        {@code LongConsumer}, it is cast to {@code LongConsumer} so
 *        the elements may be processed without boxing.
 */
@Override
default void forEach(Consumer<? super Long> consumer) {
    if (consumer instanceof LongConsumer) {
        forEach((LongConsumer) consumer);
    }
    else {
        if (Tripwire.ENABLED)
            Tripwire.trip(getClass(), "{0} calling Node.OfLong.forEachRemaining(Consumer)");
        spliterator().forEachRemaining(consumer);
    }
}

public void testDiscardReturnBound() throws IOException, ClassNotFoundException {
    List<String> list = new ArrayList<>();
    Consumer<String> c = (Consumer<String> & Serializable) list::add;
    assertSerial(c, cc -> { assertTrue(cc instanceof Consumer); });

    AtomicLong a = new AtomicLong();
    LongConsumer lc = (LongConsumer & Serializable) a::addAndGet;
    assertSerial(lc, plc -> { plc.accept(3); });
}

public void forEachRemaining(LongConsumer consumer) {
    if (consumer == null) throw new NullPointerException();
    long i = index, f = fence;
    if (i < f) {
        index = f;
        Random r = rng;
        long o = origin, b = bound;
        do {
            consumer.accept(r.internalNextLong(o, b));
        } while (++i < f);
    }
}

public boolean tryAdvance(LongConsumer consumer) {
    if (consumer == null) throw new NullPointerException();
    long i = index, f = fence;
    if (i < f) {
        consumer.accept(ThreadLocalRandom.current().internalNextLong(origin, bound));
        index = i + 1;
        return true;
    }
    return false;
}

public void forEachRemaining(LongConsumer consumer) {
    if (consumer == null) throw new NullPointerException();
    long i = index, f = fence;
    if (i < f) {
        index = f;
        long o = origin, b = bound;
        ThreadLocalRandom rng = ThreadLocalRandom.current();
        do {
            consumer.accept(rng.internalNextLong(o, b));
        } while (++i < f);
    }
}

public boolean tryAdvance(LongConsumer consumer) {
    if (consumer == null) throw new NullPointerException();
    long i = index, f = fence;
    if (i < f) {
        consumer.accept(ThreadLocalRandom.current().internalNextLong(origin, bound));
        index = i + 1;
        return true;
    }
    return false;
}

/**
 * Creates an {@code PrimitiveIterator.OfLong} from a
 * {@code Spliterator.OfLong}.
 *
 * <p>Traversal of elements should be accomplished through the iterator.
 * The behaviour of traversal is undefined if the spliterator is operated
 * after the iterator is returned.
 *
 * @param spliterator The spliterator
 * @return An iterator
 * @throws NullPointerException if the given spliterator is {@code null}
 */
public static PrimitiveIterator.OfLong iterator(Spliterator.OfLong spliterator) {
    Objects.requireNonNull(spliterator);
    class Adapter implements PrimitiveIterator.OfLong, LongConsumer {
        boolean valueReady = false;
        long nextElement;

        @Override
        public void accept(long t) {
            valueReady = true;
            nextElement = t;
        }

        @Override
        public boolean hasNext() {
            if (!valueReady)
                spliterator.tryAdvance(this);
            return valueReady;
        }

        @Override
        public long nextLong() {
            if (!valueReady && !hasNext())
                throw new NoSuchElementException();
            else {
                valueReady = false;
                return nextElement;
            }
        }
    }

    return new Adapter();
}

@Override
final boolean forEachWithCancel(Spliterator<Long> spliterator, Sink<Long> sink) {
    Spliterator.OfLong spl = adapt(spliterator);
    LongConsumer adaptedSink =  adapt(sink);
    boolean cancelled;
    do { } while (!(cancelled = sink.cancellationRequested()) && spl.tryAdvance(adaptedSink));
    return cancelled;
}

@Override
public boolean tryAdvance(LongConsumer action) {
    Objects.requireNonNull(action);

    if (count == -2) {
        action.accept(first);
        count = -1;
        return true;
    }
    else {
        return false;
    }
}

public boolean tryAdvance(LongConsumer consumer) {
    if (consumer == null) throw new NullPointerException();
    long i = index, f = fence;
    if (i < f) {
        consumer.accept(rng.internalNextLong(origin, bound));
        index = i + 1;
        return true;
    }
    return false;
}

@Override
public final LongStream flatMapToLong(Function<? super P_OUT, ? extends LongStream> mapper) {
    Objects.requireNonNull(mapper);
    // We can do better than this, by polling cancellationRequested when stream is infinite
    return new LongPipeline.StatelessOp<P_OUT>(this, StreamShape.REFERENCE,
                                               StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
        @Override
        Sink<P_OUT> opWrapSink(int flags, Sink<Long> sink) {
            return new Sink.ChainedReference<P_OUT, Long>(sink) {
                LongConsumer downstreamAsLong = downstream::accept;
                @Override
                public void begin(long size) {
                    downstream.begin(-1);
                }

                @Override
                public void accept(P_OUT u) {
                    try (LongStream result = mapper.apply(u)) {
                        // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
                        if (result != null)
                            result.sequential().forEach(downstreamAsLong);
                    }
                }
            };
        }
    };
}

/**
 * Adapt a {@code Sink<Long> to an {@code LongConsumer}, ideally simply
 * by casting.
 */
private static LongConsumer adapt(Sink<Long> sink) {
    if (sink instanceof LongConsumer) {
        return (LongConsumer) sink;
    } else {
        if (Tripwire.ENABLED)
            Tripwire.trip(AbstractPipeline.class,
                          "using LongStream.adapt(Sink<Long> s)");
        return sink::accept;
    }
}

@Override
public void forEach(LongConsumer action) {
    if (!isParallel()) {
        adapt(sourceStageSpliterator()).forEachRemaining(action);
    } else {
        super.forEach(action);
    }
}

/**
 * Adapt a {@code Sink<Long> to an {@code LongConsumer}, ideally simply
 * by casting.
 */
private static LongConsumer adapt(Sink<Long> sink) {
    if (sink instanceof LongConsumer) {
        return (LongConsumer) sink;
    } else {
        if (Tripwire.ENABLED)
            Tripwire.trip(AbstractPipeline.class,
                          "using LongStream.adapt(Sink<Long> s)");
        return sink::accept;
    }
}