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