1 /*
   2  * Copyright (c) 2013, Oracle and/or its affiliates. All rights reserved.
   3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   4  *
   5  * This code is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License version 2 only, as
   7  * published by the Free Software Foundation.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle in the LICENSE file that accompanied this code.
  10  *
  11  * This code is distributed in the hope that it will be useful, but WITHOUT
  12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14  * version 2 for more details (a copy is included in the LICENSE file that
  15  * accompanied this code).
  16  *
  17  * You should have received a copy of the GNU General Public License version
  18  * 2 along with this work; if not, write to the Free Software Foundation,
  19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  20  *
  21  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * questions.
  24  */
  25 package java.util.stream;
  26 
  27 import java.util.LongSummaryStatistics;
  28 import java.util.Objects;
  29 import java.util.OptionalDouble;
  30 import java.util.OptionalLong;
  31 import java.util.PrimitiveIterator;
  32 import java.util.Spliterator;
  33 import java.util.Spliterators;
  34 import java.util.function.BiConsumer;
  35 import java.util.function.BinaryOperator;
  36 import java.util.function.IntFunction;
  37 import java.util.function.LongBinaryOperator;
  38 import java.util.function.LongConsumer;
  39 import java.util.function.LongFunction;
  40 import java.util.function.LongPredicate;
  41 import java.util.function.LongToDoubleFunction;
  42 import java.util.function.LongToIntFunction;
  43 import java.util.function.LongUnaryOperator;
  44 import java.util.function.ObjLongConsumer;
  45 import java.util.function.Supplier;
  46 
  47 /**
  48  * Abstract base class for an intermediate pipeline stage or pipeline source
  49  * stage implementing whose elements are of type {@code long}.
  50  *
  51  * @param <E_IN> type of elements in the upstream source
  52  * @since 1.8
  53  */
  54 abstract class LongPipeline<E_IN>
  55         extends AbstractPipeline<E_IN, Long, LongStream>
  56         implements LongStream {
  57 
  58     /**
  59      * Constructor for the head of a stream pipeline.
  60      *
  61      * @param source {@code Supplier<Spliterator>} describing the stream source
  62      * @param sourceFlags the source flags for the stream source, described in
  63      *        {@link StreamOpFlag}
  64      * @param parallel {@code true} if the pipeline is parallel
  65      */
  66     LongPipeline(Supplier<? extends Spliterator<Long>> source,
  67                  int sourceFlags, boolean parallel) {
  68         super(source, sourceFlags, parallel);
  69     }
  70 
  71     /**
  72      * Constructor for the head of a stream pipeline.
  73      *
  74      * @param source {@code Spliterator} describing the stream source
  75      * @param sourceFlags the source flags for the stream source, described in
  76      *        {@link StreamOpFlag}
  77      * @param parallel {@code true} if the pipeline is parallel
  78      */
  79     LongPipeline(Spliterator<Long> source,
  80                  int sourceFlags, boolean parallel) {
  81         super(source, sourceFlags, parallel);
  82     }
  83 
  84     /**
  85      * Constructor for appending an intermediate operation onto an existing pipeline.
  86      *
  87      * @param upstream the upstream element source.
  88      * @param opFlags the operation flags
  89      */
  90     LongPipeline(AbstractPipeline<?, E_IN, ?> upstream, int opFlags) {
  91         super(upstream, opFlags);
  92     }
  93 
  94     /**
  95      * Adapt a {@code Sink<Long> to an {@code LongConsumer}, ideally simply
  96      * by casting.
  97      */
  98     private static LongConsumer adapt(Sink<Long> sink) {
  99         if (sink instanceof LongConsumer) {
 100             return (LongConsumer) sink;
 101         } else {
 102             if (Tripwire.ENABLED)
 103                 Tripwire.trip(AbstractPipeline.class,
 104                               "using LongStream.adapt(Sink<Long> s)");
 105             return sink::accept;
 106         }
 107     }
 108 
 109     /**
 110      * Adapt a {@code Spliterator<Long>} to a {@code Spliterator.OfLong}.
 111      *
 112      * @implNote
 113      * The implementation attempts to cast to a Spliterator.OfLong, and throws
 114      * an exception if this cast is not possible.
 115      */
 116     private static Spliterator.OfLong adapt(Spliterator<Long> s) {
 117         if (s instanceof Spliterator.OfLong) {
 118             return (Spliterator.OfLong) s;
 119         } else {
 120             if (Tripwire.ENABLED)
 121                 Tripwire.trip(AbstractPipeline.class,
 122                               "using LongStream.adapt(Spliterator<Long> s)");
 123             throw new UnsupportedOperationException("LongStream.adapt(Spliterator<Long> s)");
 124         }
 125     }
 126 
 127 
 128     // Shape-specific methods
 129 
 130     @Override
 131     final StreamShape getOutputShape() {
 132         return StreamShape.LONG_VALUE;
 133     }
 134 
 135     @Override
 136     final <P_IN> Node<Long> evaluateToNode(PipelineHelper<Long> helper,
 137                                            Spliterator<P_IN> spliterator,
 138                                            boolean flattenTree,
 139                                            IntFunction<Long[]> generator) {
 140         return Nodes.collectLong(helper, spliterator, flattenTree);
 141     }
 142 
 143     @Override
 144     final <P_IN> Spliterator<Long> wrap(PipelineHelper<Long> ph,
 145                                         Supplier<Spliterator<P_IN>> supplier,
 146                                         boolean isParallel) {
 147         return new StreamSpliterators.LongWrappingSpliterator<>(ph, supplier, isParallel);
 148     }
 149 
 150     @Override
 151     @SuppressWarnings("unchecked")
 152     final Spliterator.OfLong lazySpliterator(Supplier<? extends Spliterator<Long>> supplier) {
 153         return new StreamSpliterators.DelegatingSpliterator.OfLong((Supplier<Spliterator.OfLong>) supplier);
 154     }
 155 
 156     @Override
 157     final void forEachWithCancel(Spliterator<Long> spliterator, Sink<Long> sink) {
 158         Spliterator.OfLong spl = adapt(spliterator);
 159         LongConsumer adaptedSink =  adapt(sink);
 160         do { } while (!sink.cancellationRequested() && spl.tryAdvance(adaptedSink));
 161     }
 162 
 163     @Override
 164     final Node.Builder<Long> makeNodeBuilder(long exactSizeIfKnown, IntFunction<Long[]> generator) {
 165         return Nodes.longBuilder(exactSizeIfKnown);
 166     }
 167 
 168 
 169     // LongStream
 170 
 171     @Override
 172     public final PrimitiveIterator.OfLong iterator() {
 173         return Spliterators.iterator(spliterator());
 174     }
 175 
 176     @Override
 177     public final Spliterator.OfLong spliterator() {
 178         return adapt(super.spliterator());
 179     }
 180 
 181     // Stateless intermediate ops from LongStream
 182 
 183     @Override
 184     public final DoubleStream asDoubleStream() {
 185         return new DoublePipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 186                                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
 187             @Override
 188             Sink<Long> opWrapSink(int flags, Sink<Double> sink) {
 189                 return new Sink.ChainedLong<Double>(sink) {
 190                     @Override
 191                     public void accept(long t) {
 192                         downstream.accept((double) t);
 193                     }
 194                 };
 195             }
 196         };
 197     }
 198 
 199     @Override
 200     public final Stream<Long> boxed() {
 201         return mapToObj(Long::valueOf);
 202     }
 203 
 204     @Override
 205     public final LongStream map(LongUnaryOperator mapper) {
 206         Objects.requireNonNull(mapper);
 207         return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 208                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
 209             @Override
 210             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
 211                 return new Sink.ChainedLong<Long>(sink) {
 212                     @Override
 213                     public void accept(long t) {
 214                         downstream.accept(mapper.applyAsLong(t));
 215                     }
 216                 };
 217             }
 218         };
 219     }
 220 
 221     @Override
 222     public final <U> Stream<U> mapToObj(LongFunction<? extends U> mapper) {
 223         Objects.requireNonNull(mapper);
 224         return new ReferencePipeline.StatelessOp<Long, U>(this, StreamShape.LONG_VALUE,
 225                                                           StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
 226             @Override
 227             Sink<Long> opWrapSink(int flags, Sink<U> sink) {
 228                 return new Sink.ChainedLong<U>(sink) {
 229                     @Override
 230                     public void accept(long t) {
 231                         downstream.accept(mapper.apply(t));
 232                     }
 233                 };
 234             }
 235         };
 236     }
 237 
 238     @Override
 239     public final IntStream mapToInt(LongToIntFunction mapper) {
 240         Objects.requireNonNull(mapper);
 241         return new IntPipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 242                                                  StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
 243             @Override
 244             Sink<Long> opWrapSink(int flags, Sink<Integer> sink) {
 245                 return new Sink.ChainedLong<Integer>(sink) {
 246                     @Override
 247                     public void accept(long t) {
 248                         downstream.accept(mapper.applyAsInt(t));
 249                     }
 250                 };
 251             }
 252         };
 253     }
 254 
 255     @Override
 256     public final DoubleStream mapToDouble(LongToDoubleFunction mapper) {
 257         Objects.requireNonNull(mapper);
 258         return new DoublePipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 259                                                     StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
 260             @Override
 261             Sink<Long> opWrapSink(int flags, Sink<Double> sink) {
 262                 return new Sink.ChainedLong<Double>(sink) {
 263                     @Override
 264                     public void accept(long t) {
 265                         downstream.accept(mapper.applyAsDouble(t));
 266                     }
 267                 };
 268             }
 269         };
 270     }
 271 
 272     @Override
 273     public final LongStream flatMap(LongFunction<? extends LongStream> mapper) {
 274         return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 275                                      StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT | StreamOpFlag.NOT_SIZED) {
 276             @Override
 277             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
 278                 return new Sink.ChainedLong<Long>(sink) {
 279                     @Override
 280                     public void begin(long size) {
 281                         downstream.begin(-1);
 282                     }
 283 
 284                     @Override
 285                     public void accept(long t) {
 286                         try (LongStream result = mapper.apply(t)) {
 287                             // We can do better that this too; optimize for depth=0 case and just grab spliterator and forEach it
 288                             if (result != null)
 289                                 result.sequential().forEach(i -> downstream.accept(i));
 290                         }
 291                     }
 292                 };
 293             }
 294         };
 295     }
 296 
 297     @Override
 298     public LongStream unordered() {
 299         if (!isOrdered())
 300             return this;
 301         return new StatelessOp<Long>(this, StreamShape.LONG_VALUE, StreamOpFlag.NOT_ORDERED) {
 302             @Override
 303             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
 304                 return sink;
 305             }
 306         };
 307     }
 308 
 309     @Override
 310     public final LongStream filter(LongPredicate predicate) {
 311         Objects.requireNonNull(predicate);
 312         return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 313                                      StreamOpFlag.NOT_SIZED) {
 314             @Override
 315             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
 316                 return new Sink.ChainedLong<Long>(sink) {
 317                     @Override
 318                     public void begin(long size) {
 319                         downstream.begin(-1);
 320                     }
 321 
 322                     @Override
 323                     public void accept(long t) {
 324                         if (predicate.test(t))
 325                             downstream.accept(t);
 326                     }
 327                 };
 328             }
 329         };
 330     }
 331 
 332     @Override
 333     public final LongStream peek(LongConsumer consumer) {
 334         Objects.requireNonNull(consumer);
 335         return new StatelessOp<Long>(this, StreamShape.LONG_VALUE,
 336                                      0) {
 337             @Override
 338             Sink<Long> opWrapSink(int flags, Sink<Long> sink) {
 339                 return new Sink.ChainedLong<Long>(sink) {
 340                     @Override
 341                     public void accept(long t) {
 342                         consumer.accept(t);
 343                         downstream.accept(t);
 344                     }
 345                 };
 346             }
 347         };
 348     }
 349 
 350     // Stateful intermediate ops from LongStream
 351 
 352     private LongStream slice(long skip, long limit) {
 353         return SliceOps.makeLong(this, skip, limit);
 354     }
 355 
 356     @Override
 357     public final LongStream limit(long maxSize) {
 358         if (maxSize < 0)
 359             throw new IllegalArgumentException(Long.toString(maxSize));
 360         return slice(0, maxSize);
 361     }
 362 
 363     @Override
 364     public final LongStream substream(long startingOffset) {
 365         if (startingOffset < 0)
 366             throw new IllegalArgumentException(Long.toString(startingOffset));
 367         if (startingOffset == 0)
 368             return this;
 369         else
 370             return slice(startingOffset, -1);
 371     }
 372 
 373     @Override
 374     public final LongStream substream(long startingOffset, long endingOffset) {
 375         if (startingOffset < 0 || endingOffset < startingOffset)
 376             throw new IllegalArgumentException(String.format("substream(%d, %d)", startingOffset, endingOffset));
 377         return slice(startingOffset, endingOffset - startingOffset);
 378     }
 379 
 380     @Override
 381     public final LongStream sorted() {
 382         return SortedOps.makeLong(this);
 383     }
 384 
 385     @Override
 386     public final LongStream distinct() {
 387         // While functional and quick to implement, this approach is not very efficient.
 388         // An efficient version requires a long-specific map/set implementation.
 389         return boxed().distinct().mapToLong(i -> (long) i);
 390     }
 391 
 392     // Terminal ops from LongStream
 393 
 394     @Override
 395     public void forEach(LongConsumer action) {
 396         evaluate(ForEachOps.makeLong(action, false));
 397     }
 398 
 399     @Override
 400     public void forEachOrdered(LongConsumer action) {
 401         evaluate(ForEachOps.makeLong(action, true));
 402     }
 403 
 404     @Override
 405     public final long sum() {
 406         // use better algorithm to compensate for intermediate overflow?
 407         return reduce(0, Long::sum);
 408     }
 409 
 410     @Override
 411     public final OptionalLong min() {
 412         return reduce(Math::min);
 413     }
 414 
 415     @Override
 416     public final OptionalLong max() {
 417         return reduce(Math::max);
 418     }
 419 
 420     @Override
 421     public final OptionalDouble average() {
 422         long[] avg = collect(() -> new long[2],
 423                              (ll, i) -> {
 424                                  ll[0]++;
 425                                  ll[1] += i;
 426                              },
 427                              (ll, rr) -> {
 428                                  ll[0] += rr[0];
 429                                  ll[1] += rr[1];
 430                              });
 431         return avg[0] > 0
 432                ? OptionalDouble.of((double) avg[1] / avg[0])
 433                : OptionalDouble.empty();
 434     }
 435 
 436     @Override
 437     public final long count() {
 438         return map(e -> 1L).sum();
 439     }
 440 
 441     @Override
 442     public final LongSummaryStatistics summaryStatistics() {
 443         return collect(LongSummaryStatistics::new, LongSummaryStatistics::accept,
 444                        LongSummaryStatistics::combine);
 445     }
 446 
 447     @Override
 448     public final long reduce(long identity, LongBinaryOperator op) {
 449         return evaluate(ReduceOps.makeLong(identity, op));
 450     }
 451 
 452     @Override
 453     public final OptionalLong reduce(LongBinaryOperator op) {
 454         return evaluate(ReduceOps.makeLong(op));
 455     }
 456 
 457     @Override
 458     public final <R> R collect(Supplier<R> resultFactory,
 459                                ObjLongConsumer<R> accumulator,
 460                                BiConsumer<R, R> combiner) {
 461         BinaryOperator<R> operator = (left, right) -> {
 462             combiner.accept(left, right);
 463             return left;
 464         };
 465         return evaluate(ReduceOps.makeLong(resultFactory, accumulator, operator));
 466     }
 467 
 468     @Override
 469     public final boolean anyMatch(LongPredicate predicate) {
 470         return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.ANY));
 471     }
 472 
 473     @Override
 474     public final boolean allMatch(LongPredicate predicate) {
 475         return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.ALL));
 476     }
 477 
 478     @Override
 479     public final boolean noneMatch(LongPredicate predicate) {
 480         return evaluate(MatchOps.makeLong(predicate, MatchOps.MatchKind.NONE));
 481     }
 482 
 483     @Override
 484     public final OptionalLong findFirst() {
 485         return evaluate(FindOps.makeLong(true));
 486     }
 487 
 488     @Override
 489     public final OptionalLong findAny() {
 490         return evaluate(FindOps.makeLong(false));
 491     }
 492 
 493     @Override
 494     public final long[] toArray() {
 495         return Nodes.flattenLong((Node.OfLong) evaluateToArrayNode(Long[]::new))
 496                 .asPrimitiveArray();
 497     }
 498 
 499 
 500     //
 501 
 502     /**
 503      * Source stage of a LongPipeline.
 504      *
 505      * @param <E_IN> type of elements in the upstream source
 506      * @since 1.8
 507      */
 508     static class Head<E_IN> extends LongPipeline<E_IN> {
 509         /**
 510          * Constructor for the source stage of a LongStream.
 511          *
 512          * @param source {@code Supplier<Spliterator>} describing the stream
 513          *               source
 514          * @param sourceFlags the source flags for the stream source, described
 515          *                    in {@link StreamOpFlag}
 516          * @param parallel {@code true} if the pipeline is parallel
 517          */
 518         Head(Supplier<? extends Spliterator<Long>> source,
 519              int sourceFlags, boolean parallel) {
 520             super(source, sourceFlags, parallel);
 521         }
 522 
 523         /**
 524          * Constructor for the source stage of a LongStream.
 525          *
 526          * @param source {@code Spliterator} describing the stream source
 527          * @param sourceFlags the source flags for the stream source, described
 528          *                    in {@link StreamOpFlag}
 529          * @param parallel {@code true} if the pipeline is parallel
 530          */
 531         Head(Spliterator<Long> source,
 532              int sourceFlags, boolean parallel) {
 533             super(source, sourceFlags, parallel);
 534         }
 535 
 536         @Override
 537         final boolean opIsStateful() {
 538             throw new UnsupportedOperationException();
 539         }
 540 
 541         @Override
 542         final Sink<E_IN> opWrapSink(int flags, Sink<Long> sink) {
 543             throw new UnsupportedOperationException();
 544         }
 545 
 546         // Optimized sequential terminal operations for the head of the pipeline
 547 
 548         @Override
 549         public void forEach(LongConsumer action) {
 550             if (!isParallel()) {
 551                 adapt(sourceStageSpliterator()).forEachRemaining(action);
 552             } else {
 553                 super.forEach(action);
 554             }
 555         }
 556 
 557         @Override
 558         public void forEachOrdered(LongConsumer action) {
 559             if (!isParallel()) {
 560                 adapt(sourceStageSpliterator()).forEachRemaining(action);
 561             } else {
 562                 super.forEachOrdered(action);
 563             }
 564         }
 565     }
 566 
 567     /** Base class for a stateless intermediate stage of a LongStream.
 568      *
 569      * @param <E_IN> type of elements in the upstream source
 570      * @since 1.8
 571      */
 572     abstract static class StatelessOp<E_IN> extends LongPipeline<E_IN> {
 573         /**
 574          * Construct a new LongStream by appending a stateless intermediate
 575          * operation to an existing stream.
 576          * @param upstream The upstream pipeline stage
 577          * @param inputShape The stream shape for the upstream pipeline stage
 578          * @param opFlags Operation flags for the new stage
 579          */
 580         StatelessOp(AbstractPipeline<?, E_IN, ?> upstream,
 581                     StreamShape inputShape,
 582                     int opFlags) {
 583             super(upstream, opFlags);
 584             assert upstream.getOutputShape() == inputShape;
 585         }
 586 
 587         @Override
 588         final boolean opIsStateful() {
 589             return false;
 590         }
 591     }
 592 
 593     /**
 594      * Base class for a stateful intermediate stage of a LongStream.
 595      *
 596      * @param <E_IN> type of elements in the upstream source
 597      * @since 1.8
 598      */
 599     abstract static class StatefulOp<E_IN> extends LongPipeline<E_IN> {
 600         /**
 601          * Construct a new LongStream by appending a stateful intermediate
 602          * operation to an existing stream.
 603          * @param upstream The upstream pipeline stage
 604          * @param inputShape The stream shape for the upstream pipeline stage
 605          * @param opFlags Operation flags for the new stage
 606          */
 607         StatefulOp(AbstractPipeline<?, E_IN, ?> upstream,
 608                    StreamShape inputShape,
 609                    int opFlags) {
 610             super(upstream, opFlags);
 611             assert upstream.getOutputShape() == inputShape;
 612         }
 613 
 614         @Override
 615         final boolean opIsStateful() {
 616             return true;
 617         }
 618 
 619         @Override
 620         abstract <P_IN> Node<Long> opEvaluateParallel(PipelineHelper<Long> helper,
 621                                                       Spliterator<P_IN> spliterator,
 622                                                       IntFunction<Long[]> generator);
 623     }
 624 }