1 /*

     2  * Copyright (c) 2012, 2015, Oracle and/or its affiliates. All rights reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.

     8  *

     9  * This code is distributed in the hope that it will be useful, but WITHOUT

    10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    12  * version 2 for more details (a copy is included in the LICENSE file that

    13  * accompanied this code).

    14  *

    15  * You should have received a copy of the GNU General Public License version

    16  * 2 along with this work; if not, write to the Free Software Foundation,

    17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    18  *

    19  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

    20  * or visit www.oracle.com if you need additional information or have any

    21  * questions.

    22  */

    23 package org.openjdk.tests.java.util.stream;

    24 

    25 import java.util.ArrayList;

    26 import java.util.Arrays;

    27 import java.util.Collection;

    28 import java.util.Collections;

    29 import java.util.Comparator;

    30 import java.util.HashMap;

    31 import java.util.HashSet;

    32 import java.util.Iterator;

    33 import java.util.List;

    34 import java.util.Map;

    35 import java.util.Optional;

    36 import java.util.Set;

    37 import java.util.StringJoiner;

    38 import java.util.TreeMap;

    39 import java.util.concurrent.ConcurrentHashMap;

    40 import java.util.concurrent.ConcurrentSkipListMap;

    41 import java.util.concurrent.atomic.AtomicInteger;

    42 import java.util.function.BinaryOperator;

    43 import java.util.function.Function;

    44 import java.util.function.Predicate;

    45 import java.util.function.Supplier;

    46 import java.util.stream.Collector;

    47 import java.util.stream.Collectors;

    48 import java.util.stream.LambdaTestHelpers;

    49 import java.util.stream.OpTestCase;

    50 import java.util.stream.Stream;

    51 import java.util.stream.StreamOpFlagTestHelper;

    52 import java.util.stream.StreamTestDataProvider;

    53 import java.util.stream.TestData;

    54 

    55 import org.testng.annotations.Test;

    56 

    57 import static java.util.stream.Collectors.collectingAndThen;

    58 import static java.util.stream.Collectors.flatMapping;

    59 import static java.util.stream.Collectors.groupingBy;

    60 import static java.util.stream.Collectors.groupingByConcurrent;

    61 import static java.util.stream.Collectors.mapping;

    62 import static java.util.stream.Collectors.partitioningBy;

    63 import static java.util.stream.Collectors.reducing;

    64 import static java.util.stream.Collectors.toCollection;

    65 import static java.util.stream.Collectors.toConcurrentMap;

    66 import static java.util.stream.Collectors.toList;

    67 import static java.util.stream.Collectors.toMap;

    68 import static java.util.stream.Collectors.toSet;

    69 import static java.util.stream.LambdaTestHelpers.assertContents;

    70 import static java.util.stream.LambdaTestHelpers.assertContentsUnordered;

    71 import static java.util.stream.LambdaTestHelpers.mDoubler;

    72 

    73 /*

    74  * @test

    75  * @bug 8071600

    76  * @summary Test for collectors.

    77  */

    78 public class CollectorsTest extends OpTestCase {

    79 

    80     private static abstract class CollectorAssertion<T, U> {

    81         abstract void assertValue(U value,

    82                                   Supplier<Stream<T>> source,

    83                                   boolean ordered) throws ReflectiveOperationException;

    84     }

    85 

    86     static class MappingAssertion<T, V, R> extends CollectorAssertion<T, R> {

    87         private final Function<T, V> mapper;

    88         private final CollectorAssertion<V, R> downstream;

    89 

    90         MappingAssertion(Function<T, V> mapper, CollectorAssertion<V, R> downstream) {

    91             this.mapper = mapper;

    92             this.downstream = downstream;

    93         }

    94 

    95         @Override

    96         void assertValue(R value, Supplier<Stream<T>> source, boolean ordered) throws ReflectiveOperationException {

    97             downstream.assertValue(value,

    98                                    () -> source.get().map(mapper::apply),

    99                                    ordered);

   100         }

   101     }

   102 

   103     static class FlatMappingAssertion<T, V, R> extends CollectorAssertion<T, R> {

   104         private final Function<T, Stream<V>> mapper;

   105         private final CollectorAssertion<V, R> downstream;

   106 

   107         FlatMappingAssertion(Function<T, Stream<V>> mapper,

   108                              CollectorAssertion<V, R> downstream) {

   109             this.mapper = mapper;

   110             this.downstream = downstream;

   111         }

   112 

   113         @Override

   114         void assertValue(R value, Supplier<Stream<T>> source, boolean ordered) throws ReflectiveOperationException {

   115             downstream.assertValue(value,

   116                                    () -> source.get().flatMap(mapper::apply),

   117                                    ordered);

   118         }

   119     }

   120 

   121     static class GroupingByAssertion<T, K, V, M extends Map<K, ? extends V>> extends CollectorAssertion<T, M> {

   122         private final Class<? extends Map> clazz;

   123         private final Function<T, K> classifier;

   124         private final CollectorAssertion<T,V> downstream;

   125 

   126         GroupingByAssertion(Function<T, K> classifier, Class<? extends Map> clazz,

   127                             CollectorAssertion<T, V> downstream) {

   128             this.clazz = clazz;

   129             this.classifier = classifier;

   130             this.downstream = downstream;

   131         }

   132 

   133         @Override

   134         void assertValue(M map,

   135                          Supplier<Stream<T>> source,

   136                          boolean ordered) throws ReflectiveOperationException {

   137             if (!clazz.isAssignableFrom(map.getClass()))

   138                 fail(String.format("Class mismatch in GroupingByAssertion: %s, %s", clazz, map.getClass()));

   139             assertContentsUnordered(map.keySet(), source.get().map(classifier).collect(toSet()));

   140             for (Map.Entry<K, ? extends V> entry : map.entrySet()) {

   141                 K key = entry.getKey();

   142                 downstream.assertValue(entry.getValue(),

   143                                        () -> source.get().filter(e -> classifier.apply(e).equals(key)),

   144                                        ordered);

   145             }

   146         }

   147     }

   148 

   149     static class ToMapAssertion<T, K, V, M extends Map<K,V>> extends CollectorAssertion<T, M> {

   150         private final Class<? extends Map> clazz;

   151         private final Function<T, K> keyFn;

   152         private final Function<T, V> valueFn;

   153         private final BinaryOperator<V> mergeFn;

   154 

   155         ToMapAssertion(Function<T, K> keyFn,

   156                        Function<T, V> valueFn,

   157                        BinaryOperator<V> mergeFn,

   158                        Class<? extends Map> clazz) {

   159             this.clazz = clazz;

   160             this.keyFn = keyFn;

   161             this.valueFn = valueFn;

   162             this.mergeFn = mergeFn;

   163         }

   164 

   165         @Override

   166         void assertValue(M map, Supplier<Stream<T>> source, boolean ordered) throws ReflectiveOperationException {

   167             if (!clazz.isAssignableFrom(map.getClass()))

   168                 fail(String.format("Class mismatch in ToMapAssertion: %s, %s", clazz, map.getClass()));

   169             Set<K> uniqueKeys = source.get().map(keyFn).collect(toSet());

   170             assertEquals(uniqueKeys, map.keySet());

   171             source.get().forEach(t -> {

   172                 K key = keyFn.apply(t);

   173                 V v = source.get()

   174                             .filter(e -> key.equals(keyFn.apply(e)))

   175                             .map(valueFn)

   176                             .reduce(mergeFn)

   177                             .get();

   178                 assertEquals(map.get(key), v);

   179             });

   180         }

   181     }

   182 

   183     static class PartitioningByAssertion<T, D> extends CollectorAssertion<T, Map<Boolean,D>> {

   184         private final Predicate<T> predicate;

   185         private final CollectorAssertion<T,D> downstream;

   186 

   187         PartitioningByAssertion(Predicate<T> predicate, CollectorAssertion<T, D> downstream) {

   188             this.predicate = predicate;

   189             this.downstream = downstream;

   190         }

   191 

   192         @Override

   193         void assertValue(Map<Boolean, D> map,

   194                          Supplier<Stream<T>> source,

   195                          boolean ordered) throws ReflectiveOperationException {

   196             if (!Map.class.isAssignableFrom(map.getClass()))

   197                 fail(String.format("Class mismatch in PartitioningByAssertion: %s", map.getClass()));

   198             assertEquals(2, map.size());

   199             downstream.assertValue(map.get(true), () -> source.get().filter(predicate), ordered);

   200             downstream.assertValue(map.get(false), () -> source.get().filter(predicate.negate()), ordered);

   201         }

   202     }

   203 

   204     static class ToListAssertion<T> extends CollectorAssertion<T, List<T>> {

   205         @Override

   206         void assertValue(List<T> value, Supplier<Stream<T>> source, boolean ordered)

   207                 throws ReflectiveOperationException {

   208             if (!List.class.isAssignableFrom(value.getClass()))

   209                 fail(String.format("Class mismatch in ToListAssertion: %s", value.getClass()));

   210             Stream<T> stream = source.get();

   211             List<T> result = new ArrayList<>();

   212             for (Iterator<T> it = stream.iterator(); it.hasNext(); ) // avoid capturing result::add

   213                 result.add(it.next());

   214             if (StreamOpFlagTestHelper.isStreamOrdered(stream) && ordered)

   215                 assertContents(value, result);

   216             else

   217                 assertContentsUnordered(value, result);

   218         }

   219     }

   220 

   221     static class ToCollectionAssertion<T> extends CollectorAssertion<T, Collection<T>> {

   222         private final Class<? extends Collection> clazz;

   223         private final boolean targetOrdered;

   224 

   225         ToCollectionAssertion(Class<? extends Collection> clazz, boolean targetOrdered) {

   226             this.clazz = clazz;

   227             this.targetOrdered = targetOrdered;

   228         }

   229 

   230         @Override

   231         void assertValue(Collection<T> value, Supplier<Stream<T>> source, boolean ordered)

   232                 throws ReflectiveOperationException {

   233             if (!clazz.isAssignableFrom(value.getClass()))

   234                 fail(String.format("Class mismatch in ToCollectionAssertion: %s, %s", clazz, value.getClass()));

   235             Stream<T> stream = source.get();

   236             Collection<T> result = clazz.newInstance();

   237             for (Iterator<T> it = stream.iterator(); it.hasNext(); ) // avoid capturing result::add

   238                 result.add(it.next());

   239             if (StreamOpFlagTestHelper.isStreamOrdered(stream) && targetOrdered && ordered)

   240                 assertContents(value, result);

   241             else

   242                 assertContentsUnordered(value, result);

   243         }

   244     }

   245 

   246     static class ReducingAssertion<T, U> extends CollectorAssertion<T, U> {

   247         private final U identity;

   248         private final Function<T, U> mapper;

   249         private final BinaryOperator<U> reducer;

   250 

   251         ReducingAssertion(U identity, Function<T, U> mapper, BinaryOperator<U> reducer) {

   252             this.identity = identity;

   253             this.mapper = mapper;

   254             this.reducer = reducer;

   255         }

   256 

   257         @Override

   258         void assertValue(U value, Supplier<Stream<T>> source, boolean ordered)

   259                 throws ReflectiveOperationException {

   260             Optional<U> reduced = source.get().map(mapper).reduce(reducer);

   261             if (value == null)

   262                 assertTrue(!reduced.isPresent());

   263             else if (!reduced.isPresent()) {

   264                 assertEquals(value, identity);

   265             }

   266             else {

   267                 assertEquals(value, reduced.get());

   268             }

   269         }

   270     }

   271 

   272     private <T> ResultAsserter<T> mapTabulationAsserter(boolean ordered) {

   273         return (act, exp, ord, par) -> {

   274             if (par && (!ordered || !ord)) {

   275                 CollectorsTest.nestedMapEqualityAssertion(act, exp);

   276             }

   277             else {

   278                 LambdaTestHelpers.assertContentsEqual(act, exp);

   279             }

   280         };

   281     }

   282 

   283     private<T, M extends Map>

   284     void exerciseMapCollection(TestData<T, Stream<T>> data,

   285                                Collector<T, ?, ? extends M> collector,

   286                                CollectorAssertion<T, M> assertion)

   287             throws ReflectiveOperationException {

   288         boolean ordered = !collector.characteristics().contains(Collector.Characteristics.UNORDERED);

   289 

   290         M m = withData(data)

   291                 .terminal(s -> s.collect(collector))

   292                 .resultAsserter(mapTabulationAsserter(ordered))

   293                 .exercise();

   294         assertion.assertValue(m, () -> data.stream(), ordered);

   295 

   296         m = withData(data)

   297                 .terminal(s -> s.unordered().collect(collector))

   298                 .resultAsserter(mapTabulationAsserter(ordered))

   299                 .exercise();

   300         assertion.assertValue(m, () -> data.stream(), false);

   301     }

   302 

   303     private static void nestedMapEqualityAssertion(Object o1, Object o2) {

   304         if (o1 instanceof Map) {

   305             Map m1 = (Map) o1;

   306             Map m2 = (Map) o2;

   307             assertContentsUnordered(m1.keySet(), m2.keySet());

   308             for (Object k : m1.keySet())

   309                 nestedMapEqualityAssertion(m1.get(k), m2.get(k));

   310         }

   311         else if (o1 instanceof Collection) {

   312             assertContentsUnordered(((Collection) o1), ((Collection) o2));

   313         }

   314         else

   315             assertEquals(o1, o2);

   316     }

   317 

   318     private<T, R> void assertCollect(TestData.OfRef<T> data,

   319                                      Collector<T, ?, R> collector,

   320                                      Function<Stream<T>, R> streamReduction) {

   321         R check = streamReduction.apply(data.stream());

   322         withData(data).terminal(s -> s.collect(collector)).expectedResult(check).exercise();

   323     }

   324 

   325     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   326     public void testReducing(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   327         assertCollect(data, Collectors.reducing(0, Integer::sum),

   328                       s -> s.reduce(0, Integer::sum));

   329         assertCollect(data, Collectors.reducing(Integer.MAX_VALUE, Integer::min),

   330                       s -> s.min(Integer::compare).orElse(Integer.MAX_VALUE));

   331         assertCollect(data, Collectors.reducing(Integer.MIN_VALUE, Integer::max),

   332                       s -> s.max(Integer::compare).orElse(Integer.MIN_VALUE));

   333 

   334         assertCollect(data, Collectors.reducing(Integer::sum),

   335                       s -> s.reduce(Integer::sum));

   336         assertCollect(data, Collectors.minBy(Comparator.naturalOrder()),

   337                       s -> s.min(Integer::compare));

   338         assertCollect(data, Collectors.maxBy(Comparator.naturalOrder()),

   339                       s -> s.max(Integer::compare));

   340 

   341         assertCollect(data, Collectors.reducing(0, x -> x*2, Integer::sum),

   342                       s -> s.map(x -> x*2).reduce(0, Integer::sum));

   343 

   344         assertCollect(data, Collectors.summingLong(x -> x * 2L),

   345                       s -> s.map(x -> x*2L).reduce(0L, Long::sum));

   346         assertCollect(data, Collectors.summingInt(x -> x * 2),

   347                       s -> s.map(x -> x*2).reduce(0, Integer::sum));

   348         assertCollect(data, Collectors.summingDouble(x -> x * 2.0d),

   349                       s -> s.map(x -> x * 2.0d).reduce(0.0d, Double::sum));

   350 

   351         assertCollect(data, Collectors.averagingInt(x -> x * 2),

   352                       s -> s.mapToInt(x -> x * 2).average().orElse(0));

   353         assertCollect(data, Collectors.averagingLong(x -> x * 2),

   354                       s -> s.mapToLong(x -> x * 2).average().orElse(0));

   355         assertCollect(data, Collectors.averagingDouble(x -> x * 2),

   356                       s -> s.mapToDouble(x -> x * 2).average().orElse(0));

   357 

   358         // Test explicit Collector.of

   359         Collector<Integer, long[], Double> avg2xint = Collector.of(() -> new long[2],

   360                                                                    (a, b) -> {

   361                                                                        a[0] += b * 2;

   362                                                                        a[1]++;

   363                                                                    },

   364                                                                    (a, b) -> {

   365                                                                        a[0] += b[0];

   366                                                                        a[1] += b[1];

   367                                                                        return a;

   368                                                                    },

   369                                                                    a -> a[1] == 0 ? 0.0d : (double) a[0] / a[1]);

   370         assertCollect(data, avg2xint,

   371                       s -> s.mapToInt(x -> x * 2).average().orElse(0));

   372     }

   373 

   374     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   375     public void testJoining(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   376         withData(data)

   377                 .terminal(s -> s.map(Object::toString).collect(Collectors.joining()))

   378                 .expectedResult(join(data, ""))

   379                 .exercise();

   380 

   381         Collector<String, StringBuilder, String> likeJoining = Collector.of(StringBuilder::new, StringBuilder::append, (sb1, sb2) -> sb1.append(sb2.toString()), StringBuilder::toString);

   382         withData(data)

   383                 .terminal(s -> s.map(Object::toString).collect(likeJoining))

   384                 .expectedResult(join(data, ""))

   385                 .exercise();

   386 

   387         withData(data)

   388                 .terminal(s -> s.map(Object::toString).collect(Collectors.joining(",")))

   389                 .expectedResult(join(data, ","))

   390                 .exercise();

   391 

   392         withData(data)

   393                 .terminal(s -> s.map(Object::toString).collect(Collectors.joining(",", "[", "]")))

   394                 .expectedResult("[" + join(data, ",") + "]")

   395                 .exercise();

   396 

   397         withData(data)

   398                 .terminal(s -> s.map(Object::toString)

   399                                 .collect(StringBuilder::new, StringBuilder::append, StringBuilder::append)

   400                                 .toString())

   401                 .expectedResult(join(data, ""))

   402                 .exercise();

   403 

   404         withData(data)

   405                 .terminal(s -> s.map(Object::toString)

   406                                 .collect(() -> new StringJoiner(","),

   407                                          (sj, cs) -> sj.add(cs),

   408                                          (j1, j2) -> j1.merge(j2))

   409                                 .toString())

   410                 .expectedResult(join(data, ","))

   411                 .exercise();

   412 

   413         withData(data)

   414                 .terminal(s -> s.map(Object::toString)

   415                                 .collect(() -> new StringJoiner(",", "[", "]"),

   416                                          (sj, cs) -> sj.add(cs),

   417                                          (j1, j2) -> j1.merge(j2))

   418                                 .toString())

   419                 .expectedResult("[" + join(data, ",") + "]")

   420                 .exercise();

   421     }

   422 

   423     private<T> String join(TestData.OfRef<T> data, String delim) {

   424         StringBuilder sb = new StringBuilder();

   425         boolean first = true;

   426         for (T i : data) {

   427             if (!first)

   428                 sb.append(delim);

   429             sb.append(i.toString());

   430             first = false;

   431         }

   432         return sb.toString();

   433     }

   434 

   435     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   436     public void testSimpleToMap(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   437         Function<Integer, Integer> keyFn = i -> i * 2;

   438         Function<Integer, Integer> valueFn = i -> i * 4;

   439 

   440         List<Integer> dataAsList = Arrays.asList(data.stream().toArray(Integer[]::new));

   441         Set<Integer> dataAsSet = new HashSet<>(dataAsList);

   442 

   443         BinaryOperator<Integer> sum = Integer::sum;

   444         for (BinaryOperator<Integer> op : Arrays.asList((u, v) -> u,

   445                                                         (u, v) -> v,

   446                                                         sum)) {

   447             try {

   448                 exerciseMapCollection(data, toMap(keyFn, valueFn),

   449                                       new ToMapAssertion<>(keyFn, valueFn, op, HashMap.class));

   450                 if (dataAsList.size() != dataAsSet.size())

   451                     fail("Expected ISE on input with duplicates");

   452             }

   453             catch (IllegalStateException e) {

   454                 if (dataAsList.size() == dataAsSet.size())

   455                     fail("Expected no ISE on input without duplicates");

   456             }

   457 

   458             exerciseMapCollection(data, toMap(keyFn, valueFn, op),

   459                                   new ToMapAssertion<>(keyFn, valueFn, op, HashMap.class));

   460 

   461             exerciseMapCollection(data, toMap(keyFn, valueFn, op, TreeMap::new),

   462                                   new ToMapAssertion<>(keyFn, valueFn, op, TreeMap.class));

   463         }

   464 

   465         // For concurrent maps, only use commutative merge functions

   466         try {

   467             exerciseMapCollection(data, toConcurrentMap(keyFn, valueFn),

   468                                   new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentHashMap.class));

   469             if (dataAsList.size() != dataAsSet.size())

   470                 fail("Expected ISE on input with duplicates");

   471         }

   472         catch (IllegalStateException e) {

   473             if (dataAsList.size() == dataAsSet.size())

   474                 fail("Expected no ISE on input without duplicates");

   475         }

   476 

   477         exerciseMapCollection(data, toConcurrentMap(keyFn, valueFn, sum),

   478                               new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentHashMap.class));

   479 

   480         exerciseMapCollection(data, toConcurrentMap(keyFn, valueFn, sum, ConcurrentSkipListMap::new),

   481                               new ToMapAssertion<>(keyFn, valueFn, sum, ConcurrentSkipListMap.class));

   482     }

   483 

   484     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   485     public void testSimpleGroupingBy(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   486         Function<Integer, Integer> classifier = i -> i % 3;

   487 

   488         // Single-level groupBy

   489         exerciseMapCollection(data, groupingBy(classifier),

   490                               new GroupingByAssertion<>(classifier, HashMap.class,

   491                                                         new ToListAssertion<>()));

   492         exerciseMapCollection(data, groupingByConcurrent(classifier),

   493                               new GroupingByAssertion<>(classifier, ConcurrentHashMap.class,

   494                                                         new ToListAssertion<>()));

   495 

   496         // With explicit constructors

   497         exerciseMapCollection(data,

   498                               groupingBy(classifier, TreeMap::new, toCollection(HashSet::new)),

   499                               new GroupingByAssertion<>(classifier, TreeMap.class,

   500                                                         new ToCollectionAssertion<Integer>(HashSet.class, false)));

   501         exerciseMapCollection(data,

   502                               groupingByConcurrent(classifier, ConcurrentSkipListMap::new,

   503                                                    toCollection(HashSet::new)),

   504                               new GroupingByAssertion<>(classifier, ConcurrentSkipListMap.class,

   505                                                         new ToCollectionAssertion<Integer>(HashSet.class, false)));

   506     }

   507 

   508     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   509     public void testGroupingByWithMapping(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   510         Function<Integer, Integer> classifier = i -> i % 3;

   511         Function<Integer, Integer> mapper = i -> i * 2;

   512 

   513         exerciseMapCollection(data,

   514                               groupingBy(classifier, mapping(mapper, toList())),

   515                               new GroupingByAssertion<>(classifier, HashMap.class,

   516                                                         new MappingAssertion<>(mapper,

   517                                                                                new ToListAssertion<>())));

   518     }

   519 

   520     @Test

   521     public void testFlatMappingClose() {

   522         Function<Integer, Integer> classifier = i -> i;

   523         AtomicInteger ai = new AtomicInteger();

   524         Function<Integer, Stream<Integer>> flatMapper = i -> Stream.of(i, i).onClose(ai::getAndIncrement);

   525         Map<Integer, List<Integer>> m = Stream.of(1, 2).collect(groupingBy(classifier, flatMapping(flatMapper, toList())));

   526         assertEquals(m.size(), ai.get());

   527     }

   528 

   529     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   530     public void testGroupingByWithFlatMapping(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   531         Function<Integer, Integer> classifier = i -> i % 3;

   532         Function<Integer, Stream<Integer>> flatMapperByNull = i -> null;

   533         Function<Integer, Stream<Integer>> flatMapperBy0 = i -> Stream.empty();

   534         Function<Integer, Stream<Integer>> flatMapperBy2 = i -> Stream.of(i, i);

   535 

   536         exerciseMapCollection(data,

   537                               groupingBy(classifier, flatMapping(flatMapperByNull, toList())),

   538                               new GroupingByAssertion<>(classifier, HashMap.class,

   539                                                         new FlatMappingAssertion<>(flatMapperBy0,

   540                                                                                    new ToListAssertion<>())));

   541         exerciseMapCollection(data,

   542                               groupingBy(classifier, flatMapping(flatMapperBy0, toList())),

   543                               new GroupingByAssertion<>(classifier, HashMap.class,

   544                                                         new FlatMappingAssertion<>(flatMapperBy0,

   545                                                                                    new ToListAssertion<>())));

   546         exerciseMapCollection(data,

   547                               groupingBy(classifier, flatMapping(flatMapperBy2, toList())),

   548                               new GroupingByAssertion<>(classifier, HashMap.class,

   549                                                         new FlatMappingAssertion<>(flatMapperBy2,

   550                                                                                    new ToListAssertion<>())));

   551     }

   552 

   553     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   554     public void testTwoLevelGroupingBy(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   555         Function<Integer, Integer> classifier = i -> i % 6;

   556         Function<Integer, Integer> classifier2 = i -> i % 23;

   557 

   558         // Two-level groupBy

   559         exerciseMapCollection(data,

   560                               groupingBy(classifier, groupingBy(classifier2)),

   561                               new GroupingByAssertion<>(classifier, HashMap.class,

   562                                                         new GroupingByAssertion<>(classifier2, HashMap.class,

   563                                                                                   new ToListAssertion<>())));

   564         // with concurrent as upstream

   565         exerciseMapCollection(data,

   566                               groupingByConcurrent(classifier, groupingBy(classifier2)),

   567                               new GroupingByAssertion<>(classifier, ConcurrentHashMap.class,

   568                                                         new GroupingByAssertion<>(classifier2, HashMap.class,

   569                                                                                   new ToListAssertion<>())));

   570         // with concurrent as downstream

   571         exerciseMapCollection(data,

   572                               groupingBy(classifier, groupingByConcurrent(classifier2)),

   573                               new GroupingByAssertion<>(classifier, HashMap.class,

   574                                                         new GroupingByAssertion<>(classifier2, ConcurrentHashMap.class,

   575                                                                                   new ToListAssertion<>())));

   576         // with concurrent as upstream and downstream

   577         exerciseMapCollection(data,

   578                               groupingByConcurrent(classifier, groupingByConcurrent(classifier2)),

   579                               new GroupingByAssertion<>(classifier, ConcurrentHashMap.class,

   580                                                         new GroupingByAssertion<>(classifier2, ConcurrentHashMap.class,

   581                                                                                   new ToListAssertion<>())));

   582 

   583         // With explicit constructors

   584         exerciseMapCollection(data,

   585                               groupingBy(classifier, TreeMap::new, groupingBy(classifier2, TreeMap::new, toCollection(HashSet::new))),

   586                               new GroupingByAssertion<>(classifier, TreeMap.class,

   587                                                         new GroupingByAssertion<>(classifier2, TreeMap.class,

   588                                                                                   new ToCollectionAssertion<Integer>(HashSet.class, false))));

   589         // with concurrent as upstream

   590         exerciseMapCollection(data,

   591                               groupingByConcurrent(classifier, ConcurrentSkipListMap::new, groupingBy(classifier2, TreeMap::new, toList())),

   592                               new GroupingByAssertion<>(classifier, ConcurrentSkipListMap.class,

   593                                                         new GroupingByAssertion<>(classifier2, TreeMap.class,

   594                                                                                   new ToListAssertion<>())));

   595         // with concurrent as downstream

   596         exerciseMapCollection(data,

   597                               groupingBy(classifier, TreeMap::new, groupingByConcurrent(classifier2, ConcurrentSkipListMap::new, toList())),

   598                               new GroupingByAssertion<>(classifier, TreeMap.class,

   599                                                         new GroupingByAssertion<>(classifier2, ConcurrentSkipListMap.class,

   600                                                                                   new ToListAssertion<>())));

   601         // with concurrent as upstream and downstream

   602         exerciseMapCollection(data,

   603                               groupingByConcurrent(classifier, ConcurrentSkipListMap::new, groupingByConcurrent(classifier2, ConcurrentSkipListMap::new, toList())),

   604                               new GroupingByAssertion<>(classifier, ConcurrentSkipListMap.class,

   605                                                         new GroupingByAssertion<>(classifier2, ConcurrentSkipListMap.class,

   606                                                                                   new ToListAssertion<>())));

   607     }

   608 

   609     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   610     public void testGroupubgByWithReducing(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   611         Function<Integer, Integer> classifier = i -> i % 3;

   612 

   613         // Single-level simple reduce

   614         exerciseMapCollection(data,

   615                               groupingBy(classifier, reducing(0, Integer::sum)),

   616                               new GroupingByAssertion<>(classifier, HashMap.class,

   617                                                         new ReducingAssertion<>(0, LambdaTestHelpers.identity(), Integer::sum)));

   618         // with concurrent

   619         exerciseMapCollection(data,

   620                               groupingByConcurrent(classifier, reducing(0, Integer::sum)),

   621                               new GroupingByAssertion<>(classifier, ConcurrentHashMap.class,

   622                                                         new ReducingAssertion<>(0, LambdaTestHelpers.identity(), Integer::sum)));

   623 

   624         // With explicit constructors

   625         exerciseMapCollection(data,

   626                               groupingBy(classifier, TreeMap::new, reducing(0, Integer::sum)),

   627                               new GroupingByAssertion<>(classifier, TreeMap.class,

   628                                                         new ReducingAssertion<>(0, LambdaTestHelpers.identity(), Integer::sum)));

   629         // with concurrent

   630         exerciseMapCollection(data,

   631                               groupingByConcurrent(classifier, ConcurrentSkipListMap::new, reducing(0, Integer::sum)),

   632                               new GroupingByAssertion<>(classifier, ConcurrentSkipListMap.class,

   633                                                         new ReducingAssertion<>(0, LambdaTestHelpers.identity(), Integer::sum)));

   634 

   635         // Single-level map-reduce

   636         exerciseMapCollection(data,

   637                               groupingBy(classifier, reducing(0, mDoubler, Integer::sum)),

   638                               new GroupingByAssertion<>(classifier, HashMap.class,

   639                                                         new ReducingAssertion<>(0, mDoubler, Integer::sum)));

   640         // with concurrent

   641         exerciseMapCollection(data,

   642                               groupingByConcurrent(classifier, reducing(0, mDoubler, Integer::sum)),

   643                               new GroupingByAssertion<>(classifier, ConcurrentHashMap.class,

   644                                                         new ReducingAssertion<>(0, mDoubler, Integer::sum)));

   645 

   646         // With explicit constructors

   647         exerciseMapCollection(data,

   648                               groupingBy(classifier, TreeMap::new, reducing(0, mDoubler, Integer::sum)),

   649                               new GroupingByAssertion<>(classifier, TreeMap.class,

   650                                                         new ReducingAssertion<>(0, mDoubler, Integer::sum)));

   651         // with concurrent

   652         exerciseMapCollection(data,

   653                               groupingByConcurrent(classifier, ConcurrentSkipListMap::new, reducing(0, mDoubler, Integer::sum)),

   654                               new GroupingByAssertion<>(classifier, ConcurrentSkipListMap.class,

   655                                                         new ReducingAssertion<>(0, mDoubler, Integer::sum)));

   656     }

   657 

   658     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   659     public void testSimplePartitioningBy(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   660         Predicate<Integer> classifier = i -> i % 3 == 0;

   661 

   662         // Single-level partition to downstream List

   663         exerciseMapCollection(data,

   664                               partitioningBy(classifier),

   665                               new PartitioningByAssertion<>(classifier, new ToListAssertion<>()));

   666         exerciseMapCollection(data,

   667                               partitioningBy(classifier, toList()),

   668                               new PartitioningByAssertion<>(classifier, new ToListAssertion<>()));

   669     }

   670 

   671     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   672     public void testTwoLevelPartitioningBy(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   673         Predicate<Integer> classifier = i -> i % 3 == 0;

   674         Predicate<Integer> classifier2 = i -> i % 7 == 0;

   675 

   676         // Two level partition

   677         exerciseMapCollection(data,

   678                               partitioningBy(classifier, partitioningBy(classifier2)),

   679                               new PartitioningByAssertion<>(classifier,

   680                                                             new PartitioningByAssertion(classifier2, new ToListAssertion<>())));

   681 

   682         // Two level partition with reduce

   683         exerciseMapCollection(data,

   684                               partitioningBy(classifier, reducing(0, Integer::sum)),

   685                               new PartitioningByAssertion<>(classifier,

   686                                                             new ReducingAssertion<>(0, LambdaTestHelpers.identity(), Integer::sum)));

   687     }

   688 

   689     @Test(dataProvider = "StreamTestData<Integer>", dataProviderClass = StreamTestDataProvider.class)

   690     public void testComposeFinisher(String name, TestData.OfRef<Integer> data) throws ReflectiveOperationException {

   691         List<Integer> asList = exerciseTerminalOps(data, s -> s.collect(toList()));

   692         List<Integer> asImmutableList = exerciseTerminalOps(data, s -> s.collect(collectingAndThen(toList(), Collections::unmodifiableList)));

   693         assertEquals(asList, asImmutableList);

   694         try {

   695             asImmutableList.add(0);

   696             fail("Expecting immutable result");

   697         }

   698         catch (UnsupportedOperationException ignored) { }

   699     }

   700 

   701 }