jdk/src/share/classes/java/util/stream/ReduceOps.java
author briangoetz
Fri, 28 Jun 2013 16:26:54 -0400
changeset 19214 e5901820c3c1
parent 17182 b786c0de868c
child 19218 8e7212b90b81
permissions -rw-r--r--
8015318: Extend Collector with 'finish' operation Reviewed-by: mduigou Contributed-by: brian.goetz@oracle.com
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
17182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     1
/*
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     2
 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     3
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     4
 *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     5
 * This code is free software; you can redistribute it and/or modify it
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     6
 * under the terms of the GNU General Public License version 2 only, as
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     7
 * published by the Free Software Foundation.  Oracle designates this
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     8
 * particular file as subject to the "Classpath" exception as provided
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
     9
 * by Oracle in the LICENSE file that accompanied this code.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    10
 *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    11
 * This code is distributed in the hope that it will be useful, but WITHOUT
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    12
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    13
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    14
 * version 2 for more details (a copy is included in the LICENSE file that
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    15
 * accompanied this code).
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    16
 *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    17
 * You should have received a copy of the GNU General Public License version
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    18
 * 2 along with this work; if not, write to the Free Software Foundation,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    19
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    20
 *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    21
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    22
 * or visit www.oracle.com if you need additional information or have any
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    23
 * questions.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    24
 */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    25
package java.util.stream;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    26
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    27
import java.util.Objects;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    28
import java.util.Optional;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    29
import java.util.OptionalDouble;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    30
import java.util.OptionalInt;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    31
import java.util.OptionalLong;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    32
import java.util.Spliterator;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    33
import java.util.concurrent.CountedCompleter;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    34
import java.util.function.BiConsumer;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    35
import java.util.function.BiFunction;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    36
import java.util.function.BinaryOperator;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    37
import java.util.function.DoubleBinaryOperator;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    38
import java.util.function.IntBinaryOperator;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    39
import java.util.function.LongBinaryOperator;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    40
import java.util.function.ObjDoubleConsumer;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    41
import java.util.function.ObjIntConsumer;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    42
import java.util.function.ObjLongConsumer;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    43
import java.util.function.Supplier;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    44
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    45
/**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    46
 * Factory for the creating instances of {@code TerminalOp) that implement
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    47
 * reductions.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    48
 *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    49
 * @since 1.8
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    50
 */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    51
final class ReduceOps {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    52
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    53
    private ReduceOps() { }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    54
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    55
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    56
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    57
     * reference values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    58
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    59
     * @param <T> the type of the input elements
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    60
     * @param <U> the type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    61
     * @param seed the identity element for the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    62
     * @param reducer the accumulating function that incorporates an additional
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    63
     *        input element into the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    64
     * @param combiner the combining function that combines two intermediate
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    65
     *        results
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    66
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    67
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    68
    public static <T, U> TerminalOp<T, U>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    69
    makeRef(U seed, BiFunction<U, ? super T, U> reducer, BinaryOperator<U> combiner) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    70
        Objects.requireNonNull(reducer);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    71
        Objects.requireNonNull(combiner);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    72
        class ReducingSink extends Box<U> implements AccumulatingSink<T, U, ReducingSink> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    73
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    74
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    75
                state = seed;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    76
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    77
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    78
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    79
            public void accept(T t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    80
                state = reducer.apply(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    81
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    82
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    83
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    84
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    85
                state = combiner.apply(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    86
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    87
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    88
        return new ReduceOp<T, U, ReducingSink>(StreamShape.REFERENCE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    89
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    90
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    91
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    92
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    93
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    94
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    95
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    96
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    97
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    98
     * reference values producing an optional reference result.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
    99
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   100
     * @param <T> The type of the input elements, and the type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   101
     * @param operator The reducing function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   102
     * @return A {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   103
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   104
    public static <T> TerminalOp<T, Optional<T>>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   105
    makeRef(BinaryOperator<T> operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   106
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   107
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   108
                implements AccumulatingSink<T, Optional<T>, ReducingSink> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   109
            private boolean empty;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   110
            private T state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   111
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   112
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   113
                empty = true;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   114
                state = null;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   115
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   116
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   117
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   118
            public void accept(T t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   119
                if (empty) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   120
                    empty = false;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   121
                    state = t;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   122
                } else {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   123
                    state = operator.apply(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   124
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   125
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   126
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   127
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   128
            public Optional<T> get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   129
                return empty ? Optional.empty() : Optional.of(state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   130
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   131
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   132
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   133
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   134
                if (!other.empty)
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   135
                    accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   136
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   137
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   138
        return new ReduceOp<T, Optional<T>, ReducingSink>(StreamShape.REFERENCE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   139
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   140
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   141
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   142
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   143
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   144
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   145
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   146
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   147
     * Constructs a {@code TerminalOp} that implements a mutable reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   148
     * reference values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   149
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   150
     * @param <T> the type of the input elements
19214
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   151
     * @param <I> the type of the intermediate reduction result
17182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   152
     * @param collector a {@code Collector} defining the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   153
     * @return a {@code ReduceOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   154
     */
19214
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   155
    public static <T, I> TerminalOp<T, I>
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   156
    makeRef(Collector<? super T, I, ?> collector) {
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   157
        Supplier<I> supplier = Objects.requireNonNull(collector).supplier();
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   158
        BiConsumer<I, ? super T> accumulator = collector.accumulator();
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   159
        BinaryOperator<I> combiner = collector.combiner();
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   160
        class ReducingSink extends Box<I>
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   161
                implements AccumulatingSink<T, I, ReducingSink> {
17182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   162
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   163
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   164
                state = supplier.get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   165
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   166
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   167
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   168
            public void accept(T t) {
19214
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   169
                accumulator.accept(state, t);
17182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   170
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   171
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   172
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   173
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   174
                state = combiner.apply(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   175
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   176
        }
19214
e5901820c3c1 8015318: Extend Collector with 'finish' operation
briangoetz
parents: 17182
diff changeset
   177
        return new ReduceOp<T, I, ReducingSink>(StreamShape.REFERENCE) {
17182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   178
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   179
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   180
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   181
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   182
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   183
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   184
            public int getOpFlags() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   185
                return collector.characteristics().contains(Collector.Characteristics.UNORDERED)
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   186
                       ? StreamOpFlag.NOT_ORDERED
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   187
                       : 0;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   188
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   189
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   190
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   191
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   192
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   193
     * Constructs a {@code TerminalOp} that implements a mutable reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   194
     * reference values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   195
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   196
     * @param <T> the type of the input elements
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   197
     * @param <R> the type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   198
     * @param seedFactory a factory to produce a new base accumulator
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   199
     * @param accumulator a function to incorporate an element into an
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   200
     *        accumulator
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   201
     * @param reducer a function to combine an accumulator into another
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   202
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   203
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   204
    public static <T, R> TerminalOp<T, R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   205
    makeRef(Supplier<R> seedFactory,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   206
            BiConsumer<R, ? super T> accumulator,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   207
            BiConsumer<R,R> reducer) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   208
        Objects.requireNonNull(seedFactory);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   209
        Objects.requireNonNull(accumulator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   210
        Objects.requireNonNull(reducer);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   211
        class ReducingSink extends Box<R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   212
                implements AccumulatingSink<T, R, ReducingSink> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   213
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   214
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   215
                state = seedFactory.get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   216
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   217
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   218
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   219
            public void accept(T t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   220
                accumulator.accept(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   221
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   222
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   223
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   224
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   225
                reducer.accept(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   226
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   227
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   228
        return new ReduceOp<T, R, ReducingSink>(StreamShape.REFERENCE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   229
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   230
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   231
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   232
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   233
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   234
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   235
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   236
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   237
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   238
     * {@code int} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   239
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   240
     * @param identity the identity for the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   241
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   242
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   243
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   244
    public static TerminalOp<Integer, Integer>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   245
    makeInt(int identity, IntBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   246
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   247
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   248
                implements AccumulatingSink<Integer, Integer, ReducingSink>, Sink.OfInt {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   249
            private int state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   250
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   251
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   252
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   253
                state = identity;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   254
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   255
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   256
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   257
            public void accept(int t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   258
                state = operator.applyAsInt(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   259
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   260
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   261
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   262
            public Integer get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   263
                return state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   264
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   265
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   266
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   267
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   268
                accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   269
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   270
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   271
        return new ReduceOp<Integer, Integer, ReducingSink>(StreamShape.INT_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   272
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   273
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   274
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   275
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   276
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   277
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   278
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   279
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   280
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   281
     * {@code int} values, producing an optional integer result.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   282
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   283
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   284
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   285
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   286
    public static TerminalOp<Integer, OptionalInt>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   287
    makeInt(IntBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   288
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   289
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   290
                implements AccumulatingSink<Integer, OptionalInt, ReducingSink>, Sink.OfInt {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   291
            private boolean empty;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   292
            private int state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   293
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   294
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   295
                empty = true;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   296
                state = 0;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   297
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   298
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   299
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   300
            public void accept(int t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   301
                if (empty) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   302
                    empty = false;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   303
                    state = t;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   304
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   305
                else {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   306
                    state = operator.applyAsInt(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   307
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   308
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   309
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   310
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   311
            public OptionalInt get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   312
                return empty ? OptionalInt.empty() : OptionalInt.of(state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   313
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   314
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   315
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   316
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   317
                if (!other.empty)
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   318
                    accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   319
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   320
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   321
        return new ReduceOp<Integer, OptionalInt, ReducingSink>(StreamShape.INT_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   322
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   323
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   324
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   325
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   326
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   327
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   328
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   329
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   330
     * Constructs a {@code TerminalOp} that implements a mutable reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   331
     * {@code int} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   332
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   333
     * @param <R> The type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   334
     * @param supplier a factory to produce a new accumulator of the result type
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   335
     * @param accumulator a function to incorporate an int into an
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   336
     *        accumulator
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   337
     * @param combiner a function to combine an accumulator into another
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   338
     * @return A {@code ReduceOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   339
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   340
    public static <R> TerminalOp<Integer, R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   341
    makeInt(Supplier<R> supplier,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   342
            ObjIntConsumer<R> accumulator,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   343
            BinaryOperator<R> combiner) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   344
        Objects.requireNonNull(supplier);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   345
        Objects.requireNonNull(accumulator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   346
        Objects.requireNonNull(combiner);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   347
        class ReducingSink extends Box<R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   348
                implements AccumulatingSink<Integer, R, ReducingSink>, Sink.OfInt {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   349
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   350
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   351
                state = supplier.get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   352
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   353
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   354
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   355
            public void accept(int t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   356
                accumulator.accept(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   357
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   358
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   359
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   360
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   361
                state = combiner.apply(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   362
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   363
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   364
        return new ReduceOp<Integer, R, ReducingSink>(StreamShape.INT_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   365
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   366
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   367
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   368
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   369
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   370
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   371
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   372
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   373
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   374
     * {@code long} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   375
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   376
     * @param identity the identity for the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   377
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   378
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   379
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   380
    public static TerminalOp<Long, Long>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   381
    makeLong(long identity, LongBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   382
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   383
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   384
                implements AccumulatingSink<Long, Long, ReducingSink>, Sink.OfLong {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   385
            private long state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   386
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   387
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   388
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   389
                state = identity;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   390
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   391
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   392
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   393
            public void accept(long t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   394
                state = operator.applyAsLong(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   395
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   396
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   397
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   398
            public Long get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   399
                return state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   400
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   401
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   402
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   403
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   404
                accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   405
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   406
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   407
        return new ReduceOp<Long, Long, ReducingSink>(StreamShape.LONG_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   408
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   409
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   410
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   411
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   412
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   413
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   414
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   415
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   416
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   417
     * {@code long} values, producing an optional long result.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   418
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   419
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   420
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   421
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   422
    public static TerminalOp<Long, OptionalLong>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   423
    makeLong(LongBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   424
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   425
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   426
                implements AccumulatingSink<Long, OptionalLong, ReducingSink>, Sink.OfLong {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   427
            private boolean empty;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   428
            private long state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   429
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   430
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   431
                empty = true;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   432
                state = 0;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   433
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   434
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   435
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   436
            public void accept(long t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   437
                if (empty) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   438
                    empty = false;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   439
                    state = t;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   440
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   441
                else {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   442
                    state = operator.applyAsLong(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   443
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   444
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   445
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   446
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   447
            public OptionalLong get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   448
                return empty ? OptionalLong.empty() : OptionalLong.of(state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   449
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   450
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   451
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   452
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   453
                if (!other.empty)
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   454
                    accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   455
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   456
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   457
        return new ReduceOp<Long, OptionalLong, ReducingSink>(StreamShape.LONG_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   458
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   459
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   460
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   461
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   462
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   463
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   464
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   465
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   466
     * Constructs a {@code TerminalOp} that implements a mutable reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   467
     * {@code long} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   468
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   469
     * @param <R> the type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   470
     * @param supplier a factory to produce a new accumulator of the result type
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   471
     * @param accumulator a function to incorporate an int into an
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   472
     *        accumulator
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   473
     * @param combiner a function to combine an accumulator into another
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   474
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   475
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   476
    public static <R> TerminalOp<Long, R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   477
    makeLong(Supplier<R> supplier,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   478
             ObjLongConsumer<R> accumulator,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   479
             BinaryOperator<R> combiner) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   480
        Objects.requireNonNull(supplier);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   481
        Objects.requireNonNull(accumulator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   482
        Objects.requireNonNull(combiner);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   483
        class ReducingSink extends Box<R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   484
                implements AccumulatingSink<Long, R, ReducingSink>, Sink.OfLong {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   485
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   486
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   487
                state = supplier.get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   488
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   489
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   490
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   491
            public void accept(long t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   492
                accumulator.accept(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   493
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   494
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   495
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   496
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   497
                state = combiner.apply(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   498
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   499
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   500
        return new ReduceOp<Long, R, ReducingSink>(StreamShape.LONG_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   501
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   502
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   503
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   504
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   505
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   506
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   507
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   508
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   509
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   510
     * {@code double} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   511
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   512
     * @param identity the identity for the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   513
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   514
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   515
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   516
    public static TerminalOp<Double, Double>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   517
    makeDouble(double identity, DoubleBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   518
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   519
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   520
                implements AccumulatingSink<Double, Double, ReducingSink>, Sink.OfDouble {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   521
            private double state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   522
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   523
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   524
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   525
                state = identity;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   526
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   527
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   528
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   529
            public void accept(double t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   530
                state = operator.applyAsDouble(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   531
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   532
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   533
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   534
            public Double get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   535
                return state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   536
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   537
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   538
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   539
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   540
                accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   541
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   542
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   543
        return new ReduceOp<Double, Double, ReducingSink>(StreamShape.DOUBLE_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   544
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   545
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   546
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   547
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   548
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   549
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   550
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   551
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   552
     * Constructs a {@code TerminalOp} that implements a functional reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   553
     * {@code double} values, producing an optional double result.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   554
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   555
     * @param operator the combining function
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   556
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   557
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   558
    public static TerminalOp<Double, OptionalDouble>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   559
    makeDouble(DoubleBinaryOperator operator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   560
        Objects.requireNonNull(operator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   561
        class ReducingSink
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   562
                implements AccumulatingSink<Double, OptionalDouble, ReducingSink>, Sink.OfDouble {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   563
            private boolean empty;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   564
            private double state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   565
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   566
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   567
                empty = true;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   568
                state = 0;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   569
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   570
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   571
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   572
            public void accept(double t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   573
                if (empty) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   574
                    empty = false;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   575
                    state = t;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   576
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   577
                else {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   578
                    state = operator.applyAsDouble(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   579
                }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   580
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   581
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   582
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   583
            public OptionalDouble get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   584
                return empty ? OptionalDouble.empty() : OptionalDouble.of(state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   585
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   586
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   587
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   588
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   589
                if (!other.empty)
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   590
                    accept(other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   591
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   592
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   593
        return new ReduceOp<Double, OptionalDouble, ReducingSink>(StreamShape.DOUBLE_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   594
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   595
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   596
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   597
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   598
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   599
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   600
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   601
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   602
     * Constructs a {@code TerminalOp} that implements a mutable reduce on
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   603
     * {@code double} values.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   604
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   605
     * @param <R> the type of the result
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   606
     * @param supplier a factory to produce a new accumulator of the result type
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   607
     * @param accumulator a function to incorporate an int into an
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   608
     *        accumulator
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   609
     * @param combiner a function to combine an accumulator into another
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   610
     * @return a {@code TerminalOp} implementing the reduction
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   611
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   612
    public static <R> TerminalOp<Double, R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   613
    makeDouble(Supplier<R> supplier,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   614
               ObjDoubleConsumer<R> accumulator,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   615
               BinaryOperator<R> combiner) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   616
        Objects.requireNonNull(supplier);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   617
        Objects.requireNonNull(accumulator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   618
        Objects.requireNonNull(combiner);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   619
        class ReducingSink extends Box<R>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   620
                implements AccumulatingSink<Double, R, ReducingSink>, Sink.OfDouble {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   621
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   622
            public void begin(long size) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   623
                state = supplier.get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   624
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   625
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   626
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   627
            public void accept(double t) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   628
                accumulator.accept(state, t);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   629
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   630
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   631
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   632
            public void combine(ReducingSink other) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   633
                state = combiner.apply(state, other.state);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   634
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   635
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   636
        return new ReduceOp<Double, R, ReducingSink>(StreamShape.DOUBLE_VALUE) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   637
            @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   638
            public ReducingSink makeSink() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   639
                return new ReducingSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   640
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   641
        };
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   642
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   643
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   644
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   645
     * A type of {@code TerminalSink} that implements an associative reducing
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   646
     * operation on elements of type {@code T} and producing a result of type
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   647
     * {@code R}.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   648
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   649
     * @param <T> the type of input element to the combining operation
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   650
     * @param <R> the result type
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   651
     * @param <K> the type of the {@code AccumulatingSink}.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   652
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   653
    private interface AccumulatingSink<T, R, K extends AccumulatingSink<T, R, K>>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   654
            extends TerminalSink<T, R> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   655
        public void combine(K other);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   656
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   657
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   658
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   659
     * State box for a single state element, used as a base class for
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   660
     * {@code AccumulatingSink} instances
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   661
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   662
     * @param <U> The type of the state element
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   663
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   664
    private static abstract class Box<U> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   665
        U state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   666
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   667
        Box() {} // Avoid creation of special accessor
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   668
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   669
        public U get() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   670
            return state;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   671
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   672
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   673
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   674
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   675
     * A {@code TerminalOp} that evaluates a stream pipeline and sends the
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   676
     * output into an {@code AccumulatingSink}, which performs a reduce
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   677
     * operation. The {@code AccumulatingSink} must represent an associative
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   678
     * reducing operation.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   679
     *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   680
     * @param <T> the output type of the stream pipeline
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   681
     * @param <R> the result type of the reducing operation
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   682
     * @param <S> the type of the {@code AccumulatingSink}
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   683
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   684
    private static abstract class ReduceOp<T, R, S extends AccumulatingSink<T, R, S>>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   685
            implements TerminalOp<T, R> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   686
        private final StreamShape inputShape;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   687
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   688
        /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   689
         * Create a {@code ReduceOp} of the specified stream shape which uses
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   690
         * the specified {@code Supplier} to create accumulating sinks.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   691
         *
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   692
         * @param shape The shape of the stream pipeline
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   693
         */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   694
        ReduceOp(StreamShape shape) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   695
            inputShape = shape;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   696
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   697
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   698
        public abstract S makeSink();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   699
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   700
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   701
        public StreamShape inputShape() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   702
            return inputShape;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   703
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   704
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   705
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   706
        public <P_IN> R evaluateSequential(PipelineHelper<T> helper,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   707
                                           Spliterator<P_IN> spliterator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   708
            return helper.wrapAndCopyInto(makeSink(), spliterator).get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   709
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   710
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   711
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   712
        public <P_IN> R evaluateParallel(PipelineHelper<T> helper,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   713
                                         Spliterator<P_IN> spliterator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   714
            return new ReduceTask<>(this, helper, spliterator).invoke().get();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   715
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   716
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   717
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   718
    /**
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   719
     * A {@code ForkJoinTask} for performing a parallel reduce operation.
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   720
     */
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   721
    private static final class ReduceTask<P_IN, P_OUT, R,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   722
                                          S extends AccumulatingSink<P_OUT, R, S>>
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   723
            extends AbstractTask<P_IN, P_OUT, S, ReduceTask<P_IN, P_OUT, R, S>> {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   724
        private final ReduceOp<P_OUT, R, S> op;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   725
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   726
        ReduceTask(ReduceOp<P_OUT, R, S> op,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   727
                   PipelineHelper<P_OUT> helper,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   728
                   Spliterator<P_IN> spliterator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   729
            super(helper, spliterator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   730
            this.op = op;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   731
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   732
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   733
        ReduceTask(ReduceTask<P_IN, P_OUT, R, S> parent,
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   734
                   Spliterator<P_IN> spliterator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   735
            super(parent, spliterator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   736
            this.op = parent.op;
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   737
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   738
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   739
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   740
        protected ReduceTask<P_IN, P_OUT, R, S> makeChild(Spliterator<P_IN> spliterator) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   741
            return new ReduceTask<>(this, spliterator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   742
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   743
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   744
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   745
        protected S doLeaf() {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   746
            return helper.wrapAndCopyInto(op.makeSink(), spliterator);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   747
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   748
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   749
        @Override
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   750
        public void onCompletion(CountedCompleter caller) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   751
            if (!isLeaf()) {
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   752
                S leftResult = leftChild.getLocalResult();
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   753
                leftResult.combine(rightChild.getLocalResult());
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   754
                setLocalResult(leftResult);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   755
            }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   756
            // GC spliterator, left and right child
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   757
            super.onCompletion(caller);
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   758
        }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   759
    }
b786c0de868c 8011920: Main streams implementation
mduigou
parents:
diff changeset
   760
}