--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/jdk/src/share/classes/java/util/stream/DistinctOps.java Wed Apr 24 16:15:47 2013 -0700
@@ -0,0 +1,162 @@
+/*
+ * 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.HashSet;
+import java.util.LinkedHashSet;
+import java.util.Objects;
+import java.util.Set;
+import java.util.Spliterator;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.IntFunction;
+
+/**
+ * Factory methods for transforming streams into duplicate-free streams, using
+ * {@link Object#equals(Object)} to determine equality.
+ *
+ * @since 1.8
+ */
+final class DistinctOps {
+
+ private DistinctOps() { }
+
+ /**
+ * Appends a "distinct" operation to the provided stream, and returns the
+ * new stream.
+ *
+ * @param <T> the type of both input and output elements
+ * @param upstream a reference stream with element type T
+ * @return the new stream
+ */
+ static <T> ReferencePipeline<T, T> makeRef(AbstractPipeline<?, T, ?> upstream) {
+ return new ReferencePipeline.StatefulOp<T, T>(upstream, StreamShape.REFERENCE,
+ StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {
+ @Override
+ <P_IN> Node<T> opEvaluateParallel(PipelineHelper<T> helper,
+ Spliterator<P_IN> spliterator,
+ IntFunction<T[]> generator) {
+ if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {
+ // No-op
+ return helper.evaluate(spliterator, false, generator);
+ }
+ else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {
+ // If the stream is SORTED then it should also be ORDERED so the following will also
+ // preserve the sort order
+ TerminalOp<T, LinkedHashSet<T>> reduceOp
+ = ReduceOps.<T, LinkedHashSet<T>>makeRef(LinkedHashSet::new, LinkedHashSet::add,
+ LinkedHashSet::addAll);
+ return Nodes.node(reduceOp.evaluateParallel(helper, spliterator));
+ }
+ else {
+ // Holder of null state since ConcurrentHashMap does not support null values
+ AtomicBoolean seenNull = new AtomicBoolean(false);
+ ConcurrentHashMap<T, Boolean> map = new ConcurrentHashMap<>();
+ TerminalOp<T, Void> forEachOp = ForEachOps.makeRef(t -> {
+ if (t == null)
+ seenNull.set(true);
+ else
+ map.putIfAbsent(t, Boolean.TRUE);
+ }, false);
+ forEachOp.evaluateParallel(helper, spliterator);
+
+ // If null has been seen then copy the key set into a HashSet that supports null values
+ // and add null
+ Set<T> keys = map.keySet();
+ if (seenNull.get()) {
+ // TODO Implement a more efficient set-union view, rather than copying
+ keys = new HashSet<>(keys);
+ keys.add(null);
+ }
+ return Nodes.node(keys);
+ }
+ }
+
+ @Override
+ Sink<T> opWrapSink(int flags, Sink<T> sink) {
+ Objects.requireNonNull(sink);
+
+ if (StreamOpFlag.DISTINCT.isKnown(flags)) {
+ return sink;
+ } else if (StreamOpFlag.SORTED.isKnown(flags)) {
+ return new Sink.ChainedReference<T>(sink) {
+ boolean seenNull;
+ T lastSeen;
+
+ @Override
+ public void begin(long size) {
+ seenNull = false;
+ lastSeen = null;
+ downstream.begin(-1);
+ }
+
+ @Override
+ public void end() {
+ seenNull = false;
+ lastSeen = null;
+ downstream.end();
+ }
+
+ @Override
+ public void accept(T t) {
+ if (t == null) {
+ if (!seenNull) {
+ seenNull = true;
+ downstream.accept(lastSeen = null);
+ }
+ } else if (lastSeen == null || !t.equals(lastSeen)) {
+ downstream.accept(lastSeen = t);
+ }
+ }
+ };
+ } else {
+ return new Sink.ChainedReference<T>(sink) {
+ Set<T> seen;
+
+ @Override
+ public void begin(long size) {
+ seen = new HashSet<>();
+ downstream.begin(-1);
+ }
+
+ @Override
+ public void end() {
+ seen = null;
+ downstream.end();
+ }
+
+ @Override
+ public void accept(T t) {
+ if (!seen.contains(t)) {
+ seen.add(t);
+ downstream.accept(t);
+ }
+ }
+ };
+ }
+ }
+ };
+ }
+}