/* * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved. * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER. * * This code is free software; you can redistribute it and/or modify it * under the terms of the GNU General Public License version 2 only, as * published by the Free Software Foundation. Oracle designates this * particular file as subject to the "Classpath" exception as provided * by Oracle in the LICENSE file that accompanied this code. * * This code is distributed in the hope that it will be useful, but WITHOUT * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License * version 2 for more details (a copy is included in the LICENSE file that * accompanied this code). * * You should have received a copy of the GNU General Public License version * 2 along with this work; if not, write to the Free Software Foundation, * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA. * * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA * or visit www.oracle.com if you need additional information or have any * questions. */ package java.util.stream; import java.util.ArrayList; import java.util.Arrays; import java.util.Comparator; import java.util.Objects; import java.util.Spliterator; import java.util.concurrent.ForkJoinTask; import java.util.function.IntFunction; /** * Factory methods for transforming streams into sorted streams. * * @since 1.8 */ final class SortedOps { private SortedOps() { } /** * Appends a "sorted" operation to the provided stream. * * @param the type of both input and output elements * @param upstream a reference stream with element type T */ static Stream makeRef(AbstractPipeline upstream) { return new OfRef<>(upstream); } /** * Appends a "sorted" operation to the provided stream. * * @param the type of both input and output elements * @param upstream a reference stream with element type T * @param comparator the comparator to order elements by */ static Stream makeRef(AbstractPipeline upstream, Comparator comparator) { return new OfRef<>(upstream, comparator); } /** * Appends a "sorted" operation to the provided stream. * * @param the type of both input and output elements * @param upstream a reference stream with element type T */ static IntStream makeInt(AbstractPipeline upstream) { return new OfInt(upstream); } /** * Appends a "sorted" operation to the provided stream. * * @param the type of both input and output elements * @param upstream a reference stream with element type T */ static LongStream makeLong(AbstractPipeline upstream) { return new OfLong(upstream); } /** * Appends a "sorted" operation to the provided stream. * * @param the type of both input and output elements * @param upstream a reference stream with element type T */ static DoubleStream makeDouble(AbstractPipeline upstream) { return new OfDouble(upstream); } /** * Specialized subtype for sorting reference streams */ private static final class OfRef extends ReferencePipeline.StatefulOp { /** * Comparator used for sorting */ private final boolean isNaturalSort; private final Comparator comparator; /** * Sort using natural order of {@literal } which must be * {@code Comparable}. */ OfRef(AbstractPipeline upstream) { super(upstream, StreamShape.REFERENCE, StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SORTED); this.isNaturalSort = true; // Will throw CCE when we try to sort if T is not Comparable this.comparator = (Comparator) Comparator.naturalOrder(); } /** * Sort using the provided comparator. * * @param comparator The comparator to be used to evaluate ordering. */ OfRef(AbstractPipeline upstream, Comparator comparator) { super(upstream, StreamShape.REFERENCE, StreamOpFlag.IS_ORDERED | StreamOpFlag.NOT_SORTED); this.isNaturalSort = false; this.comparator = Objects.requireNonNull(comparator); } @Override public Sink opWrapSink(int flags, Sink sink) { Objects.requireNonNull(sink); // If the input is already naturally sorted and this operation // also naturally sorted then this is a no-op if (StreamOpFlag.SORTED.isKnown(flags) && isNaturalSort) return sink; else if (StreamOpFlag.SIZED.isKnown(flags)) return new SizedRefSortingSink<>(sink, comparator); else return new RefSortingSink<>(sink, comparator); } @Override public Node opEvaluateParallel(PipelineHelper helper, Spliterator spliterator, IntFunction 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); } } } /** * Specialized subtype for sorting int streams. */ private static final class OfInt extends IntPipeline.StatefulOp { OfInt(AbstractPipeline upstream) { super(upstream, StreamShape.INT_VALUE, StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SORTED); } @Override public Sink opWrapSink(int flags, Sink sink) { Objects.requireNonNull(sink); if (StreamOpFlag.SORTED.isKnown(flags)) return sink; else if (StreamOpFlag.SIZED.isKnown(flags)) return new SizedIntSortingSink(sink); else return new IntSortingSink(sink); } @Override public Node opEvaluateParallel(PipelineHelper helper, Spliterator spliterator, IntFunction 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.asIntArray(); Arrays.parallelSort(content); return Nodes.node(content); } } } /** * Specialized subtype for sorting long streams. */ private static final class OfLong extends LongPipeline.StatefulOp { OfLong(AbstractPipeline upstream) { super(upstream, StreamShape.LONG_VALUE, StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SORTED); } @Override public Sink opWrapSink(int flags, Sink sink) { Objects.requireNonNull(sink); if (StreamOpFlag.SORTED.isKnown(flags)) return sink; else if (StreamOpFlag.SIZED.isKnown(flags)) return new SizedLongSortingSink(sink); else return new LongSortingSink(sink); } @Override public Node opEvaluateParallel(PipelineHelper helper, Spliterator spliterator, IntFunction 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.asLongArray(); Arrays.parallelSort(content); return Nodes.node(content); } } } /** * Specialized subtype for sorting double streams. */ private static final class OfDouble extends DoublePipeline.StatefulOp { OfDouble(AbstractPipeline upstream) { super(upstream, StreamShape.DOUBLE_VALUE, StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SORTED); } @Override public Sink opWrapSink(int flags, Sink sink) { Objects.requireNonNull(sink); if (StreamOpFlag.SORTED.isKnown(flags)) return sink; else if (StreamOpFlag.SIZED.isKnown(flags)) return new SizedDoubleSortingSink(sink); else return new DoubleSortingSink(sink); } @Override public Node opEvaluateParallel(PipelineHelper helper, Spliterator spliterator, IntFunction generator) { if (StreamOpFlag.SORTED.isKnown(helper.getStreamAndOpFlags())) { return helper.evaluate(spliterator, false, generator); } else { Node.OfDouble n = (Node.OfDouble) helper.evaluate(spliterator, true, generator); double[] content = n.asDoubleArray(); Arrays.parallelSort(content); return Nodes.node(content); } } } /** * {@link ForkJoinTask} for implementing sort on SIZED reference streams. */ private static final class SizedRefSortingSink extends Sink.ChainedReference { private final Comparator comparator; private T[] array; private int offset; SizedRefSortingSink(Sink sink, Comparator comparator) { super(sink); this.comparator = comparator; } @Override public void begin(long size) { if (size >= Nodes.MAX_ARRAY_SIZE) throw new IllegalArgumentException("Stream size exceeds max array size"); array = (T[]) new Object[(int) size]; } @Override public void end() { // Need to use offset rather than array.length since the downstream // many be short-circuiting // @@@ A better approach is to know if the downstream short-circuits // and check sink.cancellationRequested Arrays.sort(array, 0, offset, comparator); downstream.begin(offset); for (int i = 0; i < offset; i++) downstream.accept(array[i]); downstream.end(); array = null; } @Override public void accept(T t) { array[offset++] = t; } } /** * {@link Sink} for implementing sort on reference streams. */ private static final class RefSortingSink extends Sink.ChainedReference { private final Comparator comparator; private ArrayList list; RefSortingSink(Sink sink, Comparator comparator) { super(sink); this.comparator = comparator; } @Override public void begin(long size) { list = (size >= 0) ? new ArrayList((int) size) : new ArrayList(); } @Override public void end() { list.sort(comparator); downstream.begin(list.size()); list.forEach(downstream::accept); downstream.end(); list = null; } @Override public void accept(T t) { list.add(t); } } /** * {@link Sink} for implementing sort on SIZED int streams. */ private static final class SizedIntSortingSink extends Sink.ChainedInt { private int[] array; private int offset; SizedIntSortingSink(Sink downstream) { super(downstream); } @Override public void begin(long size) { if (size >= Nodes.MAX_ARRAY_SIZE) throw new IllegalArgumentException("Stream size exceeds max array size"); array = new int[(int) size]; } @Override public void end() { Arrays.sort(array, 0, offset); downstream.begin(offset); for (int i = 0; i < offset; i++) downstream.accept(array[i]); downstream.end(); array = null; } @Override public void accept(int t) { array[offset++] = t; } } /** * {@link Sink} for implementing sort on int streams. */ private static final class IntSortingSink extends Sink.ChainedInt { private SpinedBuffer.OfInt b; IntSortingSink(Sink sink) { super(sink); } @Override public void begin(long size) { b = (size > 0) ? new SpinedBuffer.OfInt((int) size) : new SpinedBuffer.OfInt(); } @Override public void end() { int[] ints = b.asIntArray(); Arrays.sort(ints); downstream.begin(ints.length); for (int anInt : ints) downstream.accept(anInt); downstream.end(); } @Override public void accept(int t) { b.accept(t); } } /** * {@link Sink} for implementing sort on SIZED long streams. */ private static final class SizedLongSortingSink extends Sink.ChainedLong { private long[] array; private int offset; SizedLongSortingSink(Sink downstream) { super(downstream); } @Override public void begin(long size) { if (size >= Nodes.MAX_ARRAY_SIZE) throw new IllegalArgumentException("Stream size exceeds max array size"); array = new long[(int) size]; } @Override public void end() { Arrays.sort(array, 0, offset); downstream.begin(offset); for (int i = 0; i < offset; i++) downstream.accept(array[i]); downstream.end(); array = null; } @Override public void accept(long t) { array[offset++] = t; } } /** * {@link Sink} for implementing sort on long streams. */ private static final class LongSortingSink extends Sink.ChainedLong { private SpinedBuffer.OfLong b; LongSortingSink(Sink sink) { super(sink); } @Override public void begin(long size) { b = (size > 0) ? new SpinedBuffer.OfLong((int) size) : new SpinedBuffer.OfLong(); } @Override public void end() { long[] longs = b.asLongArray(); Arrays.sort(longs); downstream.begin(longs.length); for (long aLong : longs) downstream.accept(aLong); downstream.end(); } @Override public void accept(long t) { b.accept(t); } } /** * {@link Sink} for implementing sort on SIZED double streams. */ private static final class SizedDoubleSortingSink extends Sink.ChainedDouble { private double[] array; private int offset; SizedDoubleSortingSink(Sink downstream) { super(downstream); } @Override public void begin(long size) { if (size >= Nodes.MAX_ARRAY_SIZE) throw new IllegalArgumentException("Stream size exceeds max array size"); array = new double[(int) size]; } @Override public void end() { Arrays.sort(array, 0, offset); downstream.begin(offset); for (int i = 0; i < offset; i++) downstream.accept(array[i]); downstream.end(); array = null; } @Override public void accept(double t) { array[offset++] = t; } } /** * {@link Sink} for implementing sort on double streams. */ private static final class DoubleSortingSink extends Sink.ChainedDouble { private SpinedBuffer.OfDouble b; DoubleSortingSink(Sink sink) { super(sink); } @Override public void begin(long size) { b = (size > 0) ? new SpinedBuffer.OfDouble((int) size) : new SpinedBuffer.OfDouble(); } @Override public void end() { double[] doubles = b.asDoubleArray(); Arrays.sort(doubles); downstream.begin(doubles.length); for (double aDouble : doubles) downstream.accept(aDouble); downstream.end(); } @Override public void accept(double t) { b.accept(t); } } }