2.x: Cleanup after the new groupBy additions

Author: akarnokd

src/main/java/io/reactivex/Flowable.java

@@ -9717,25 +9717,25 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      * {@link Subscriber} during its lifetime and if this {@code Subscriber} cancels before the
      * source terminates, the next emission by the source having the same key will trigger a new
      * {@code GroupedPublisher} emission. The {@code evictingMapFactory} is used to create a map that will
-     * be used to hold the {@link GroupedFlowable}s by key. The evicting map created by this factory must 
+     * be used to hold the {@link GroupedFlowable}s by key. The evicting map created by this factory must
      * notify the provided {@code Consumer<Object>} with the entry value (not the key!) when an entry in this
      * map has been evicted. The next source emission will bring about the  completion of the evicted
      * {@link GroupedFlowable}s and the arrival of an item with the same key as a completed {@link GroupedFlowable}
      * will prompt the creation and emission of a new {@link GroupedFlowable} with that key.
      * 
      * <p>A use case for specifying an {@code evictingMapFactory} is where the source is infinite and fast and
-     * over time the number of keys grows enough to be a concern in terms of the memory footprint of the 
+     * over time the number of keys grows enough to be a concern in terms of the memory footprint of the
      * internal hash map containing the {@link GroupedFlowable}s.
      * 
      * <p>The map created by an {@code evictingMapFactory} must be thread-safe.
      * 
      * <p>An example of an {@code evictingMapFactory} using <a href="https://google.github.io/guava/releases/24.0-jre/api/docs/com/google/common/cache/CacheBuilder.html">CacheBuilder</a> from the Guava library is below:
      * 
-     * <pre>
-     * Function&lt;Consumer&lt;Object&gt;, Map&lt;Integer, Object&gt;&gt; evictingMapFactory = 
+     * <pre><code>
+     * Function&lt;Consumer&lt;Object&gt;, Map&lt;Integer, Object&gt;&gt; evictingMapFactory =
      *   notify -&gt;
      *       CacheBuilder
-     *         .newBuilder() 
+     *         .newBuilder()
      *         .maximumSize(3)
      *         .removalListener(entry -&gt; {
      *              try {
@@ -9749,14 +9749,14 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      *         .asMap();
      *
      * // Emit 1000 items but ensure that the
-     * // internal map never has more than 3 items in it           
+     * // internal map never has more than 3 items in it
      * Flowable
      *   .range(1, 1000)
      *   // note that number of keys is 10
      *   .groupBy(x -&gt; x % 10, x -&gt; x, true, 16, evictingMapFactory)
      *   .flatMap(g -&gt; g)
      *   .forEach(System.out::println);
-     * </pre>
+     * </code></pre>
      * 
      * <p>
      * <img width="640" height="360" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/groupBy.png" alt="">
@@ -9786,7 +9786,7 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
      * @param bufferSize
      *            the hint for how many {@link GroupedFlowable}s and element in each {@link GroupedFlowable} should be buffered
      * @param evictingMapFactory
-     *            The factory used to create a map that will be used by the implementation to hold the 
+     *            The factory used to create a map that will be used by the implementation to hold the
      *            {@link GroupedFlowable}s. The evicting map created by this factory must
      *            notify the provided {@code Consumer<Object>} with the entry value (not the key!) when
      *            an entry in this map has been evicted. The next source emission will bring about the
@@ -9808,7 +9808,7 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
     @Beta
     public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T, ? extends K> keySelector,
             Function<? super T, ? extends V> valueSelector,
-            boolean delayError, int bufferSize, 
+            boolean delayError, int bufferSize,
             Function<? super Consumer<Object>, ? extends Map<K, Object>> evictingMapFactory) {
         ObjectHelper.requireNonNull(keySelector, "keySelector is null");
         ObjectHelper.requireNonNull(valueSelector, "valueSelector is null");
@@ -9817,7 +9817,7 @@ public final <K, V> Flowable<GroupedFlowable<K, V>> groupBy(Function<? super T,
 
         return RxJavaPlugins.onAssembly(new FlowableGroupBy<T, K, V>(this, keySelector, valueSelector, bufferSize, delayError, evictingMapFactory));
     }
-    
+
     /**
      * Returns a Flowable that correlates two Publishers when they overlap in time and groups the results.
      * <p>

src/main/java/io/reactivex/internal/operators/flowable/FlowableGroupBy.java

@@ -41,7 +41,7 @@ public final class FlowableGroupBy<T, K, V> extends AbstractFlowableWithUpstream
     final boolean delayError;
     final Function<? super Consumer<Object>, ? extends Map<K, Object>> mapFactory;
 
-    public FlowableGroupBy(Flowable<T> source, Function<? super T, ? extends K> keySelector, Function<? super T, ? extends V> valueSelector, 
+    public FlowableGroupBy(Flowable<T> source, Function<? super T, ? extends K> keySelector, Function<? super T, ? extends V> valueSelector,
             int bufferSize, boolean delayError, Function<? super Consumer<Object>, ? extends Map<K, Object>> mapFactory) {
         super(source);
         this.keySelector = keySelector;
@@ -52,6 +52,7 @@ public FlowableGroupBy(Flowable<T> source, Function<? super T, ? extends K> keyS
     }
 
     @Override
+    @SuppressWarnings({ "unchecked", "rawtypes" })
     protected void subscribeActual(Subscriber<? super GroupedFlowable<K, V>> s) {
 
         final Map<Object, GroupedUnicast<K, V>> groups;
@@ -63,8 +64,8 @@ protected void subscribeActual(Subscriber<? super GroupedFlowable<K, V>> s) {
                 groups = new ConcurrentHashMap<Object, GroupedUnicast<K, V>>();
             } else {
                 evictedGroups = new ConcurrentLinkedQueue<GroupedUnicast<K, V>>();
-                Consumer<Object> evictionAction = (Consumer<Object>)(Consumer<?>) new EvictionAction<K, V>(evictedGroups);
-                groups = (Map<Object, GroupedUnicast<K,V>>)(Map<Object, ?>) mapFactory.apply(evictionAction);
+                Consumer<Object> evictionAction = (Consumer) new EvictionAction<K, V>(evictedGroups);
+                groups = (Map) mapFactory.apply(evictionAction);
             }
         } catch (Exception e) {
             Exceptions.throwIfFatal(e);
@@ -176,7 +177,7 @@ public void onNext(T t) {
             }
 
             group.onNext(v);
-            
+
             if (evictedGroups != null) {
                 GroupedUnicast<K, V> evictedGroup;
                 while ((evictedGroup = evictedGroups.poll()) != null) {

src/test/java/io/reactivex/internal/operators/flowable/FlowableGroupByTest.java

@@ -1848,7 +1848,7 @@ public Publisher<Integer> apply(GroupedFlowable<Integer, Integer> g) throws Exce
         .test()
         .assertResult(1);
     }
-    
+
     @Test
     public void mapFactoryThrows() {
         final IOException ex = new IOException("boo");
@@ -1860,13 +1860,13 @@ public Map<Integer, Object> apply(final Consumer<Object> notify) throws Exceptio
                         throw ex;
                     }
                 };
-        Flowable.just(1) 
-          .groupBy(Functions.<Integer>identity(), Functions.identity(), true, 16, evictingMapFactory) 
-          .test() 
+        Flowable.just(1)
+          .groupBy(Functions.<Integer>identity(), Functions.identity(), true, 16, evictingMapFactory)
+          .test()
           .assertNoValues()
           .assertError(ex);
     }
-    
+
     @Test
     public void mapFactoryExpiryCompletesGroupedFlowable() {
         final List<Integer> completed = new CopyOnWriteArrayList<Integer>();
@@ -1887,15 +1887,15 @@ public void mapFactoryExpiryCompletesGroupedFlowable() {
         ts.assertComplete();
         ts.assertValueCount(3);
     }
-    
+
     private static final Function<Integer, Integer> mod5 = new Function<Integer, Integer>() {
 
         @Override
         public Integer apply(Integer n) throws Exception {
             return n % 5;
         }
     };
-    
+
     @Test
     public void mapFactoryWithExpiringGuavaCacheDemonstrationCodeForUseInJavadoc() {
         //javadoc will be a version of this using lambdas and without assertions
@@ -1911,9 +1911,9 @@ public void mapFactoryWithExpiringGuavaCacheDemonstrationCodeForUseInJavadoc() {
                 .assertComplete();
         ts.assertValueCount(numValues);
         //the exact eviction behaviour of the guava cache is not specified so we make some approximate tests
-        assertTrue(completed.size() > numValues *0.9);
+        assertTrue(completed.size() > numValues * 0.9);
     }
-    
+
     @Test
     public void mapFactoryEvictionQueueClearedOnErrorCoverageOnly() {
         Function<Consumer<Object>, Map<Integer, Object>> evictingMapFactory = createEvictingMapFactorySynchronousOnly(1);
@@ -2045,7 +2045,7 @@ public Set<Entry<K, V>> entrySet() {
 
     private static Function<Consumer<Object>, Map<Integer, Object>> createEvictingMapFactoryGuava(final int maxSize) {
         Function<Consumer<Object>, Map<Integer, Object>> evictingMapFactory =  //
-                new Function<Consumer<Object>, Map<Integer, Object>>(){
+                new Function<Consumer<Object>, Map<Integer, Object>>() {
 
             @Override
             public Map<Integer, Object> apply(final Consumer<Object> notify) throws Exception {
@@ -2068,7 +2068,7 @@ public void onRemoval(RemovalNotification<Integer,Object> notification) {
 
     private static Function<Consumer<Object>, Map<Integer, Object>> createEvictingMapFactorySynchronousOnly(final int maxSize) {
         Function<Consumer<Object>, Map<Integer, Object>> evictingMapFactory =  //
-                new Function<Consumer<Object>, Map<Integer, Object>>(){
+                new Function<Consumer<Object>, Map<Integer, Object>>() {
 
                     @Override
                     public Map<Integer, Object> apply(final Consumer<Object> notify) throws Exception {