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
433862
static <T> ReferencePipeline<T, T> makeRef(AbstractPipeline<?, T, ?> upstream) {
        return new ReferencePipeline.StatefulOp<T, T>(upstream, StreamShape.REFERENCE,
                                                      StreamOpFlag.IS_DISTINCT | StreamOpFlag.NOT_SIZED) {

            <P_IN> Node<T> reduce(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {
                // 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));
            }

            @Override
            public <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())) {
                    return reduce(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
            public <P_IN> Spliterator<T> opEvaluateParallelLazy(PipelineHelper<T> helper, Spliterator<P_IN> spliterator) {
                if (StreamOpFlag.DISTINCT.isKnown(helper.getStreamAndOpFlags())) {
                    // No-op
                    return helper.wrapSpliterator(spliterator);
                }
                else if (StreamOpFlag.ORDERED.isKnown(helper.getStreamAndOpFlags())) {
                    // Not lazy, barrier required to preserve order
                    return reduce(helper, spliterator).spliterator();
                }
                else {
                    // Lazy
                    return new StreamSpliterators.DistinctSpliterator<>(helper.wrapSpliterator(spliterator));
                }
            }

            @Override
            public 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, 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, 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);
                            }
                        }
                    };
                }
            }
        };
    }
170856
public <T, R> Stream<R> map(Stream<T> stream, Function<? super T, ? extends R> mapper) {
        Callable<Stream<R>> callable = () -> stream.map(mapper);
        return ThreadMethods.forkJoinExecution(callable, concurrencyConfiguration, stream.isParallel());
    }
1097179
public <R> Stream<R> map(Function<? super T, ? extends R> mapper) {
        List<R> list = new ArrayList<R>();
        while (iterator.hasNext()) {
            list.add(mapper.apply(iterator.next()));
        }
        return Stream.of(list);
    }
320281
public static <T, U> Stream<U> cast(final Stream<T> stream, final Class<? extends U> type) {
        return stream.map(type::cast);
    }
627262
@Override
  public StreamBuilderImpl<T,U> reduce(BinaryOperatorSync<T> op)
  {
    return new ReduceOpSync<>(this, op);
  }
70960
public static <T> Iterator<T> distinct(final Iterator<T> iterator) {
        requireNonNull(iterator);
        return new AbstractIterator<T>() {
            private final PeekingIterator<T> peekingIterator = peekingIterator(iterator);
            private T curr = null;

            @Override
            protected T computeNext() {
                while (peekingIterator.hasNext() && Objects.equals(curr, peekingIterator.peek())) {
                    peekingIterator.next();
                }
                if (!peekingIterator.hasNext())
                    return endOfData();

                curr = peekingIterator.next();
                return curr;
            }
        };
    }
605752
public static Pipe newPipe(final InputStream in)
    {
        final CodedInput codedInput = new CodedInput(in, false);
        return new Pipe()
        {
            protected Input begin(Pipe.Schema<?> pipeSchema) throws IOException
            {
                return codedInput;
            }
            protected void end(Pipe.Schema<?> pipeSchema, Input input, 
                    boolean cleanupOnly) throws IOException
            {
                if(cleanupOnly)
                    return;
                
                assert input == codedInput;
            }
        };
    }
1078062
public <K> Stream<T> distinctBy(final Function<? super T, K> keyMapper, final Predicate<? super Long> occurrencesFilter) {
        final Supplier<? extends Map<Keyed<K, T>, Long>> supplier = isParallel() ? Suppliers.<Keyed<K, T>, Long> ofConcurrentHashMap()
                : Suppliers.<Keyed<K, T>, Long> ofLinkedHashMap();

        return groupBy(Fn.<K, T> keyed(keyMapper), Collectors.counting(), supplier).filter(Fn.<Keyed<K, T>, Long> testByValue(occurrencesFilter))
                .map(Fn.<T, K, Long> kk());
    }
1183641
@Override
    public final DoubleStream asDoubleStream() {
        return new DoublePipeline.StatelessOp<Long>(this, StreamShape.LONG_VALUE,
                                                    StreamOpFlag.NOT_SORTED | StreamOpFlag.NOT_DISTINCT) {
            @Override
            public Sink<Long> opWrapSink(int flags, Sink<Double> sink) {
                return new Sink.ChainedLong<Double>(sink) {
                    @Override
                    public void accept(long t) {
                        downstream.accept((double) t);
                    }
                };
            }
        };
    }
686021
public void copyFrom(InputStream in) throws IOException, InterruptedException {
        try (OutputStream os = write()) {
            org.apache.commons.io.IOUtils.copy(in, os);
        }
    }
409881
@javax.annotation.Nonnull
  public PipedInputStream newInputStream() throws IOException {
    @javax.annotation.Nonnull final com.simiacryptus.util.io.TeeOutputStream outTee = this;
    @javax.annotation.Nonnull final AtomicReference<Runnable> onClose = new AtomicReference<>();
    @javax.annotation.Nonnull final PipedOutputStream outPipe = new PipedOutputStream();
    @javax.annotation.Nonnull final PipedInputStream in = new PipedInputStream() {
      @Override
      public void close() throws IOException {
        outPipe.close();
        super.close();
      }
    };
    outPipe.connect(in);
    @javax.annotation.Nonnull final OutputStream outAsync = new AsyncOutputStream(outPipe);
    new Thread(() -> {
      try {
        if (null != heapBuffer) {
          outAsync.write(heapBuffer.toByteArray());
          outAsync.flush();
        }
        outTee.branches.add(outAsync);
      } catch (@javax.annotation.Nonnull final IOException e) {
        e.printStackTrace();
      }
    }).start();
    onClose.set(() -> {
      outTee.branches.remove(outAsync);
      System.err.println("END HTTP Session");
    });
    return in;
  }
789024
public static <T> LazyIterable<T> distinct(Iterable<T> iterable)
    {
        return new DistinctIterable<T>(iterable);
    }
316262
public <T> DistinctBufferMaker<T> createDistinctBuffer( Serializer<T> distinctSerializer ) {
        return new MakeDistinctBuffer<T>("buffer-" + dbCounter.incrementAndGet(), distinctSerializer);
    }
488546
private static <T> Stream<T> generateStream(T seed, Predicate<? super T> hasNext, UnaryOperator<T> next) {
        final Spliterator<T> spliterator = Spliterators.spliteratorUnknownSize(new Iterator<T>() {
            private T last = seed;

            @Override
            public boolean hasNext() {
                return hasNext.test(last);
            }

            @Override
            public T next() {
                final T current = last;
                last = next.apply(last);
                return current;
            }
        }, Spliterator.ORDERED);

        return StreamSupport.stream(spliterator, false);
    }
382087
public Stream<Tag> stream() {
        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator(),
                Spliterator.ORDERED | Spliterator.DISTINCT | Spliterator.NONNULL | Spliterator.SORTED), false);
    }
771323
@Override
    public Stream<P_OUT> unordered() {
        if (!isOrdered())
            return this;
        return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE, StreamOpFlag.NOT_ORDERED) {
            @Override
            public Sink<P_OUT> opWrapSink(int flags, Sink<P_OUT> sink) {
                return sink;
            }
        };
    }
1150862
public static <K, V> EntryStream<K, V> of(Stream<? extends Entry<K, V>> stream) {
        if (stream instanceof AbstractStreamEx) {
            @SuppressWarnings("unchecked")
            AbstractStreamEx<Entry<K, V>, ?> ase = (AbstractStreamEx<Entry<K, V>, ?>) stream;
            if (ase.spliterator != null)
                return new EntryStream<>(ase.spliterator(), ase.context);
            return new EntryStream<>(ase.stream(), ase.context);
        }
        return new EntryStream<>(stream, StreamContext.of(stream));
    }