add test and guava as test dep, check for null evictingMapFactory

Author: davidmoten

build.gradle

@@ -13,6 +13,7 @@ apply plugin: 'nebula.rxjava-project'
 dependencies {
     testCompile 'junit:junit:4.12'
     testCompile 'org.mockito:mockito-core:1.10.19'
+    testCompile 'com.google.guava:guava:19.0'
 
     perfCompile 'org.openjdk.jmh:jmh-core:1.11.3'
     perfCompile 'org.openjdk.jmh:jmh-generator-annprocess:1.11.3'

src/main/java/rx/Observable.java

@@ -6385,8 +6385,7 @@ public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? su
      *            items to the appropriate {@code GroupedObservable}s. The {@code Map} instance must be thread-safe 
      *            and any eviction must trigger a call to the supplied action (synchronously or asynchronously). 
      *            This can be used to limit the size of the map by evicting keys by maximum size or access time for 
-     *            instance. If {@code evictingMapFactory} is null then no eviction strategy will be applied (and a suitable default thread-safe 
-     *            implementation of {@code Map} will be supplied). Here's an example using Guava's {@code CacheBuilder} from v19.0:
+     *            instance. Here's an example using Guava's {@code CacheBuilder} from v19.0:
      *            <pre>
      *            {@code
      *            Func1<Action1<K>, Map<K, Object>> mapFactory 
@@ -6405,10 +6404,16 @@ public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? su
      * @return an {@code Observable} that emits {@link GroupedObservable}s, each of which corresponds to a
      *         unique key value and each of which emits those items from the source Observable that share that
      *         key value
+     * @throws NullPointerException
+     *             if {@code evictingMapFactory} is null
      * @see <a href="http://reactivex.io/documentation/operators/groupby.html">ReactiveX operators documentation: GroupBy</a>
      */
     @Experimental
-    public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? super T, ? extends K> keySelector, final Func1<? super T, ? extends R> elementSelector, final Func1<Action1<K>, Map<K, Object>> evictingMapFactory) {
+    public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? super T, ? extends K> keySelector, 
+            final Func1<? super T, ? extends R> elementSelector, final Func1<Action1<K>, Map<K, Object>> evictingMapFactory) {
+        if (evictingMapFactory == null) {
+            throw new NullPointerException("evictingMapFactory cannot be null");
+        }
         return lift(new OperatorGroupBy<T, K, R>(keySelector, elementSelector, evictingMapFactory));
     }
 

src/test/java/rx/internal/operators/OperatorGroupByTest.java

@@ -30,6 +30,7 @@
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
+import rx.exceptions.Exceptions;
 import rx.exceptions.TestException;
 import rx.functions.*;
 import rx.internal.util.*;
@@ -1865,4 +1866,46 @@ public String call(Integer x) {
         }).toList().toBlocking().single();
         assertEquals(expected, ts.getOnNextEvents());
     }
+    
+    private static final Func1<Integer, Integer> EVICTING_MAP_KEY_SELECTOR = new Func1<Integer, Integer> (){
+        @Override
+        public Integer call(Integer t) {
+            return t /10;
+        }};
+    
+    @Test
+    public void testEvictingMapFactoryIfMapPutThrowsThen() {
+        Func1<Integer, Integer> elementSelector = UtilityFunctions.identity();
+        Exceptions.throwIfFatal(null);
+        final RuntimeException exception = new RuntimeException("boo");
+        //normally we would use Guava CacheBuilder for instance but to save a test dependency 
+        //we make something custom
+        Func1<Action1<Integer>, Map<Integer, Object>> mapFactory = new Func1<Action1<Integer>, Map<Integer, Object>>() {
+            @SuppressWarnings("serial")
+            @Override
+            public Map<Integer, Object> call(final Action1<Integer> evicted) {
+                // is a bit risky to override the put method because 
+                // of possible side-effects (e.g. remove could call put and we did not know it)
+                // to fix just need to use composition but needs a verbose implementation of Map
+                // interface
+                return new ConcurrentHashMap<Integer,Object>() {
+                    
+                    @Override
+                    public Object put(Integer key, Object value) {
+                        throw exception;
+                    }};
+            }};
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        Observable
+            .range(1, 100)
+            .groupBy(EVICTING_MAP_KEY_SELECTOR, elementSelector, mapFactory)
+            .flatMap(UtilityFunctions.<Observable<Integer>>identity())
+            .subscribe(ts);
+        ts.assertError(exception);
+    }
+    
+    @Test
+    public void mapFactoryEvictionWorksWithGuavaCache() {
+        
+    }
 }