View Javadoc
1   /*
2    * Licensed to the Apache Software Foundation (ASF) under one or more
3    * contributor license agreements.  See the NOTICE file distributed with
4    * this work for additional information regarding copyright ownership.
5    * The ASF licenses this file to You under the Apache License, Version 2.0
6    * (the "License"); you may not use this file except in compliance with
7    * the License.  You may obtain a copy of the License at
8    *
9    *      https://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS,
13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14   * See the License for the specific language governing permissions and
15   * limitations under the License.
16   */
17  package org.apache.commons.lang3;
18  
19  import java.util.ArrayList;
20  import java.util.Collection;
21  import java.util.Collections;
22  import java.util.List;
23  import java.util.Set;
24  import java.util.function.BiConsumer;
25  import java.util.function.BinaryOperator;
26  import java.util.function.Consumer;
27  import java.util.function.Function;
28  import java.util.function.Predicate;
29  import java.util.function.Supplier;
30  import java.util.stream.Collector;
31  import java.util.stream.Collectors;
32  import java.util.stream.Stream;
33  
34  import org.apache.commons.lang3.Functions.FailableConsumer;
35  import org.apache.commons.lang3.Functions.FailableFunction;
36  import org.apache.commons.lang3.Functions.FailablePredicate;
37  
38  /**
39   * Provides utility functions, and classes for working with the
40   * {@code java.util.stream} package, or more generally, with Java 8 lambdas. More
41   * specifically, it attempts to address the fact that lambdas are supposed
42   * not to throw Exceptions, at least not checked Exceptions, AKA instances
43   * of {@link Exception}. This enforces the use of constructs like:
44   *
45   * <pre>{@code
46   *     Consumer<java.lang.reflect.Method> consumer = m -> {
47   *         try {
48   *             m.invoke(o, args);
49   *         } catch (Throwable t) {
50   *             throw Functions.rethrow(t);
51   *         }
52   *    };
53   *    stream.forEach(consumer);
54   * }</pre>
55   * <p>
56   * Using a {@link FailableStream}, this can be rewritten as follows:
57   * </p>
58   * <pre>{@code
59   *     Streams.failable(stream).forEach(m -> m.invoke(o, args));
60   * }</pre>
61   * <p>
62   * Obviously, the second version is much more concise and the spirit of
63   * Lambda expressions is met better than in the first version.
64   * </p>
65   *
66   * @see Stream
67   * @see Functions
68   * @since 3.10
69   * @deprecated Use {@link org.apache.commons.lang3.stream.Streams}.
70   */
71  @Deprecated
72  public class Streams {
73  
74      /**
75       * A Collector type for arrays.
76       *
77       * @param <O> The array type.
78       * @deprecated Use {@link org.apache.commons.lang3.stream.Streams.ArrayCollector}.
79       */
80      @Deprecated
81      public static class ArrayCollector<O> implements Collector<O, List<O>, O[]> {
82          private static final Set<Characteristics> characteristics = Collections.emptySet();
83          private final Class<O> elementType;
84  
85          /**
86           * Constructs a new instance for the given element type.
87           *
88           * @param elementType The element type.
89           */
90          public ArrayCollector(final Class<O> elementType) {
91              this.elementType = elementType;
92          }
93  
94          @Override
95          public BiConsumer<List<O>, O> accumulator() {
96              return List::add;
97          }
98  
99          @Override
100         public Set<Characteristics> characteristics() {
101             return characteristics;
102         }
103 
104         @Override
105         public BinaryOperator<List<O>> combiner() {
106             return (left, right) -> {
107                 left.addAll(right);
108                 return left;
109             };
110         }
111 
112         @Override
113         public Function<List<O>, O[]> finisher() {
114             return list -> list.toArray(ArrayUtils.newInstance(elementType, list.size()));
115         }
116 
117         @Override
118         public Supplier<List<O>> supplier() {
119             return ArrayList::new;
120         }
121     }
122 
123     /**
124      * A reduced, and simplified version of a {@link Stream} with
125      * failable method signatures.
126      *
127      * @param <O> The streams element type.
128      * @deprecated Use {@link org.apache.commons.lang3.stream.Streams.FailableStream}.
129      */
130     @Deprecated
131     public static class FailableStream<O> {
132 
133         private Stream<O> stream;
134         private boolean terminated;
135 
136         /**
137          * Constructs a new instance with the given {@code stream}.
138          *
139          * @param stream The stream.
140          */
141         public FailableStream(final Stream<O> stream) {
142             this.stream = stream;
143         }
144 
145         /**
146          * Returns whether all elements of this stream match the provided predicate.
147          * May not evaluate the predicate on all elements if not necessary for
148          * determining the result.  If the stream is empty then {@code true} is
149          * returned and the predicate is not evaluated.
150          *
151          * <p>
152          * This is a short-circuiting terminal operation.
153          * </p>
154          *
155          * <p>
156          * Note
157          * This method evaluates the <em>universal quantification</em> of the
158          * predicate over the elements of the stream (for all x P(x)).  If the
159          * stream is empty, the quantification is said to be <em>vacuously
160          * satisfied</em> and is always {@code true} (regardless of P(x)).
161          * </p>
162          *
163          * @param predicate A non-interfering, stateless predicate to apply to
164          * elements of this stream.
165          * @return {@code true} If either all elements of the stream match the
166          * provided predicate or the stream is empty, otherwise {@code false}.
167          */
168         public boolean allMatch(final FailablePredicate<O, ?> predicate) {
169             assertNotTerminated();
170             return stream().allMatch(Functions.asPredicate(predicate));
171         }
172 
173         /**
174          * Returns whether any elements of this stream match the provided
175          * predicate.  May not evaluate the predicate on all elements if not
176          * necessary for determining the result.  If the stream is empty then
177          * {@code false} is returned and the predicate is not evaluated.
178          *
179          * <p>
180          * This is a short-circuiting terminal operation.
181          * </p>
182          *
183          * Note
184          * This method evaluates the <em>existential quantification</em> of the
185          * predicate over the elements of the stream (for some x P(x)).
186          *
187          * @param predicate A non-interfering, stateless predicate to apply to
188          * elements of this stream.
189          * @return {@code true} if any elements of the stream match the provided
190          * predicate, otherwise {@code false}.
191          */
192         public boolean anyMatch(final FailablePredicate<O, ?> predicate) {
193             assertNotTerminated();
194             return stream().anyMatch(Functions.asPredicate(predicate));
195         }
196 
197         /**
198          * Throws IllegalStateException if this stream is already terminated.
199          *
200          * @throws IllegalStateException if this stream is already terminated.
201          */
202         protected void assertNotTerminated() {
203             if (terminated) {
204                 throw new IllegalStateException("This stream is already terminated.");
205             }
206         }
207 
208         /**
209          * Performs a mutable reduction operation on the elements of this stream using a
210          * {@link Collector}.  A {@link Collector}
211          * encapsulates the functions used as arguments to
212          * {@link #collect(Supplier, BiConsumer, BiConsumer)}, allowing for reuse of
213          * collection strategies and composition of collect operations such as
214          * multiple-level grouping or partitioning.
215          *
216          * <p>
217          * If the underlying stream is parallel, and the {@link Collector}
218          * is concurrent, and either the stream is unordered or the collector is
219          * unordered, then a concurrent reduction will be performed
220          * (see {@link Collector} for details on concurrent reduction.)
221          * </p>
222          *
223          * <p>
224          * This is an intermediate operation.
225          * </p>
226          *
227          * <p>
228          * When executed in parallel, multiple intermediate results may be
229          * instantiated, populated, and merged so as to maintain isolation of
230          * mutable data structures.  Therefore, even when executed in parallel
231          * with non-thread-safe data structures (such as {@link ArrayList}), no
232          * additional synchronization is needed for a parallel reduction.
233          * </p>
234          * <p>
235          * Note
236          * The following will accumulate strings into an ArrayList:
237          * </p>
238          * <pre>{@code
239          *     List<String> asList = stringStream.collect(Collectors.toList());
240          * }</pre>
241          *
242          * <p>
243          * The following will classify {@code Person} objects by city:
244          * </p>
245          * <pre>{@code
246          *     Map<String, List<Person>> peopleByCity
247          *         = personStream.collect(Collectors.groupingBy(Person::getCity));
248          * }</pre>
249          *
250          * <p>
251          * The following will classify {@code Person} objects by state and city,
252          * cascading two {@link Collector}s together:
253          * </p>
254          * <pre>{@code
255          *     Map<String, Map<String, List<Person>>> peopleByStateAndCity
256          *         = personStream.collect(Collectors.groupingBy(Person::getState,
257          *                                                      Collectors.groupingBy(Person::getCity)));
258          * }</pre>
259          *
260          * @param <R> the type of the result.
261          * @param <A> the intermediate accumulation type of the {@link Collector}.
262          * @param collector the {@link Collector} describing the reduction.
263          * @return the result of the reduction.
264          * @see #collect(Supplier, BiConsumer, BiConsumer)
265          * @see Collectors
266          */
267         public <A, R> R collect(final Collector<? super O, A, R> collector) {
268             makeTerminated();
269             return stream().collect(collector);
270         }
271 
272         /**
273          * Performs a mutable reduction operation on the elements of this FailableStream.
274          * A mutable reduction is one in which the reduced value is a mutable result
275          * container, such as an {@link ArrayList}, and elements are incorporated by updating
276          * the state of the result rather than by replacing the result. This produces a result equivalent to:
277          * <pre>{@code
278          *     R result = supplier.get();
279          *     for (T element : this stream)
280          *         accumulator.accept(result, element);
281          *     return result;
282          * }</pre>
283          *
284          * <p>
285          * Like {@link #reduce(Object, BinaryOperator)}, {@code collect} operations
286          * can be parallelized without requiring additional synchronization.
287          * </p>
288          *
289          * <p>
290          * This is an intermediate operation.
291          * </p>
292          *
293          * <p>
294          * Note There are many existing classes in the JDK whose signatures are
295          * well-suited for use with method references as arguments to {@code collect()}.
296          * For example, the following will accumulate strings into an {@link ArrayList}:
297          * </p>
298          * <pre>{@code
299          *     List<String> asList = stringStream.collect(ArrayList::new, ArrayList::add,
300          *                                                ArrayList::addAll);
301          * }</pre>
302          *
303          * <p>
304          * The following will take a stream of strings and concatenates them into a
305          * single string:
306          * </p>
307          * <pre>{@code
308          *     String concat = stringStream.collect(StringBuilder::new, StringBuilder::append,
309          *                                          StringBuilder::append)
310          *                                 .toString();
311          * }</pre>
312          *
313          * @param <R> type of the result.
314          * @param <A> Type of the accumulator.
315          * @param supplier a function that creates a new result container. For a
316          *                 parallel execution, this function may be called
317          *                 multiple times and must return a fresh value each time.
318          * @param accumulator An associative, non-interfering, stateless function for
319          *   incorporating an additional element into a result.
320          * @param combiner An associative, non-interfering, stateless
321          *   function for combining two values, which must be compatible with the
322          *   accumulator function.
323          * @return The result of the reduction.
324          */
325         public <A, R> R collect(final Supplier<R> supplier, final BiConsumer<R, ? super O> accumulator, final BiConsumer<R, R> combiner) {
326             makeTerminated();
327             return stream().collect(supplier, accumulator, combiner);
328         }
329 
330         /**
331          * Returns a FailableStream consisting of the elements of this stream that match
332          * the given FailablePredicate.
333          *
334          * <p>
335          * This is an intermediate operation.
336          * </p>
337          *
338          * @param predicate a non-interfering, stateless predicate to apply to each
339          * element to determine if it should be included.
340          * @return the new stream.
341          */
342         public FailableStream<O> filter(final FailablePredicate<O, ?> predicate) {
343             assertNotTerminated();
344             stream = stream.filter(Functions.asPredicate(predicate));
345             return this;
346         }
347 
348         /**
349          * Performs an action for each element of this stream.
350          *
351          * <p>
352          * This is an intermediate operation.
353          * </p>
354          *
355          * <p>
356          * The behavior of this operation is explicitly nondeterministic.
357          * For parallel stream pipelines, this operation does <em>not</em>
358          * guarantee to respect the encounter order of the stream, as doing so
359          * would sacrifice the benefit of parallelism.  For any given element, the
360          * action may be performed at whatever time and in whatever thread the
361          * library chooses.  If the action accesses shared state, it is
362          * responsible for providing the required synchronization.
363          * </p>
364          *
365          * @param action a non-interfering action to perform on the elements.
366          */
367         public void forEach(final FailableConsumer<O, ?> action) {
368             makeTerminated();
369             stream().forEach(Functions.asConsumer(action));
370         }
371 
372         /**
373          * Marks this stream as terminated.
374          *
375          * @throws IllegalStateException if this stream is already terminated.
376          */
377         protected void makeTerminated() {
378             assertNotTerminated();
379             terminated = true;
380         }
381 
382         /**
383          * Returns a stream consisting of the results of applying the given
384          * function to the elements of this stream.
385          *
386          * <p>
387          * This is an intermediate operation.
388          * </p>
389          *
390          * @param <R> The element type of the new stream.
391          * @param mapper A non-interfering, stateless function to apply to each element.
392          * @return the new stream.
393          */
394         public <R> FailableStream<R> map(final FailableFunction<O, R, ?> mapper) {
395             assertNotTerminated();
396             return new FailableStream<>(stream.map(Functions.asFunction(mapper)));
397         }
398 
399         /**
400          * Performs a reduction on the elements of this stream, using the provided
401          * identity value and an associative accumulation function, and returns
402          * the reduced value.  This is equivalent to:
403          * <pre>{@code
404          *     T result = identity;
405          *     for (T element : this stream)
406          *         result = accumulator.apply(result, element)
407          *     return result;
408          * }</pre>
409          *
410          * but is not constrained to execute sequentially.
411          *
412          * <p>
413          * The {@code identity} value must be an identity for the accumulator
414          * function. This means that for all {@code t},
415          * {@code accumulator.apply(identity, t)} is equal to {@code t}.
416          * The {@code accumulator} function must be an associative function.
417          * </p>
418          *
419          * <p>
420          * This is an intermediate operation.
421          * </p>
422          *
423          * Note Sum, min, max, average, and string concatenation are all special
424          * cases of reduction. Summing a stream of numbers can be expressed as:
425          *
426          * <pre>{@code
427          *     Integer sum = integers.reduce(0, (a, b) -> a+b);
428          * }</pre>
429          *
430          * or:
431          *
432          * <pre>{@code
433          *     Integer sum = integers.reduce(0, Integer::sum);
434          * }</pre>
435          *
436          * <p>
437          * While this may seem a more roundabout way to perform an aggregation
438          * compared to simply mutating a running total in a loop, reduction
439          * operations parallelize more gracefully, without needing additional
440          * synchronization and with greatly reduced risk of data races.
441          * </p>
442          *
443          * @param identity the identity value for the accumulating function.
444          * @param accumulator an associative, non-interfering, stateless
445          *                    function for combining two values.
446          * @return the result of the reduction.
447          */
448         public O reduce(final O identity, final BinaryOperator<O> accumulator) {
449             makeTerminated();
450             return stream().reduce(identity, accumulator);
451         }
452 
453         /**
454          * Converts the FailableStream into an equivalent stream.
455          *
456          * @return A stream, which will return the same elements, which this FailableStream would return.
457          */
458         public Stream<O> stream() {
459             return stream;
460         }
461     }
462 
463     /**
464      * Converts the given {@link Collection} into a {@link FailableStream}.
465      * This is basically a simplified, reduced version of the {@link Stream}
466      * class, with the same underlying element stream, except that failable
467      * objects, like {@link FailablePredicate}, {@link FailableFunction}, or
468      * {@link FailableConsumer} may be applied, instead of
469      * {@link Predicate}, {@link Function}, or {@link Consumer}. The idea is
470      * to rewrite a code snippet like this:
471      * <pre>{@code
472      *     final List<O> list;
473      *     final Method m;
474      *     final Function<O,String> mapper = (o) -> {
475      *         try {
476      *             return (String) m.invoke(o);
477      *         } catch (Throwable t) {
478      *             throw Functions.rethrow(t);
479      *         }
480      *     };
481      *     final List<String> strList = list.stream()
482      *         .map(mapper).collect(Collectors.toList());
483      *  }</pre>
484      * <p>
485      * as follows:
486      * </p>
487      *  <pre>{@code
488      *     final List<O> list;
489      *     final Method m;
490      *     final List<String> strList = Functions.stream(list.stream())
491      *         .map((o) -> (String) m.invoke(o)).collect(Collectors.toList());
492      *  }</pre>
493      * <p>
494      * While the second version may not be <em>quite</em> as
495      * efficient (because it depends on the creation of additional,
496      * intermediate objects, of type FailableStream), it is much more
497      * concise, and readable, and meets the spirit of Lambdas better
498      * than the first version.
499      * </p>
500      *
501      * @param <O> The streams element type.
502      * @param stream The stream, which is being converted.
503      * @return The {@link FailableStream}, which has been created by
504      *   converting the stream.
505      */
506     public static <O> FailableStream<O> stream(final Collection<O> stream) {
507         return stream(stream.stream());
508     }
509 
510     /**
511      * Converts the given {@link Stream stream} into a {@link FailableStream}.
512      * This is basically a simplified, reduced version of the {@link Stream}
513      * class, with the same underlying element stream, except that failable
514      * objects, like {@link FailablePredicate}, {@link FailableFunction}, or
515      * {@link FailableConsumer} may be applied, instead of
516      * {@link Predicate}, {@link Function}, or {@link Consumer}. The idea is
517      * to rewrite a code snippet like this:
518      * <pre>{@code
519      *     final List<O> list;
520      *     final Method m;
521      *     final Function<O,String> mapper = (o) -> {
522      *         try {
523      *             return (String) m.invoke(o);
524      *         } catch (Throwable t) {
525      *             throw Functions.rethrow(t);
526      *         }
527      *     };
528      *     final List<String> strList = list.stream()
529      *         .map(mapper).collect(Collectors.toList());
530      *  }</pre>
531      * <p>
532      * as follows:
533      * </p>
534      *  <pre>{@code
535      *     final List<O> list;
536      *     final Method m;
537      *     final List<String> strList = Functions.stream(list.stream())
538      *         .map((o) -> (String) m.invoke(o)).collect(Collectors.toList());
539      *  }</pre>
540      * <p>
541      * While the second version may not be <em>quite</em> as
542      * efficient (because it depends on the creation of additional,
543      * intermediate objects, of type FailableStream), it is much more
544      * concise, and readable, and meets the spirit of Lambdas better
545      * than the first version.
546      * </p>
547      *
548      * @param <O> The streams element type.
549      * @param stream The stream, which is being converted.
550      * @return The {@link FailableStream}, which has been created by
551      *   converting the stream.
552      */
553     public static <O> FailableStream<O> stream(final Stream<O> stream) {
554         return new FailableStream<>(stream);
555     }
556 
557     /**
558      * Returns a {@link Collector} that accumulates the input elements into a
559      * new array.
560      *
561      * @param elementType Type of an element in the array.
562      * @param <O> the type of the input elements.
563      * @return a {@link Collector} which collects all the input elements into an
564      * array, in encounter order.
565      */
566     public static <O> Collector<O, ?, O[]> toArray(final Class<O> elementType) {
567         return new ArrayCollector<>(elementType);
568     }
569 
570     /**
571      * Constructs a new instance.
572      */
573     public Streams() {
574         // empty
575     }
576 }