add tests

Author: davidmoten

src/main/java/rx/Observable.java

@@ -6391,8 +6391,8 @@ public final <K, R> Observable<GroupedObservable<K, R>> groupBy(final Func1<? su
      *            Func1<Action1<K>, Map<K, Object>> mapFactory 
      *              = action -> CacheBuilder.newBuilder()
      *                  .maximumSize(1000)
-     *                  .expireAfterAccess(12, TimeUnit.HOUR)
-     *                  .removalListener(key -> action.call(key))
+     *                  .expireAfterAccess(12, TimeUnit.HOURS)
+     *                  .removalListener(notification -> action.call(notification.getKey()))
      *                  .<K, Object> build().asMap();
      *            }
      *            </pre>

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

@@ -26,11 +26,16 @@
 import org.junit.*;
 import org.mockito.*;
 
+import com.google.common.cache.CacheBuilder;
+import com.google.common.cache.RemovalListener;
+import com.google.common.cache.RemovalNotification;
+
 import rx.*;
 import rx.Observable;
 import rx.Observable.OnSubscribe;
 import rx.Observer;
 import rx.exceptions.Exceptions;
+import rx.exceptions.OnErrorNotImplementedException;
 import rx.exceptions.TestException;
 import rx.functions.*;
 import rx.internal.util.*;
@@ -1858,29 +1863,62 @@ public String call(Integer x) {
             })
             .subscribe(ts);
         assertEquals(Arrays.asList(4, 5, 6, 7, 8, 9), evictedKeys);
-        List<String> expected = Observable.range(1, 100).map(new Func1<Integer, String>() {
-            @Override
-            public String call(Integer x) {
-                return (x /10) + ":" + x;
-            }
-        }).toList().toBlocking().single();
+        List<String> expected = Observable
+                .range(1, 100)
+                .map(new Func1<Integer, String>() {
+                        @Override
+                        public String call(Integer x) {
+                            return (x /10) + ":" + x;
+                        }
+                    })
+                .toList().toBlocking().single();
         assertEquals(expected, ts.getOnNextEvents());
     }
 
+    
+    Func1<Integer, Integer> elementSelector = UtilityFunctions.identity();
     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() {
+    public void testEvictingMapFactoryIfMapPutThrowsRuntimeExceptionThenErrorEmittedByStream() {
         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>>() {
+        Func1<Action1<Integer>, Map<Integer, Object>> mapFactory = createMapFactoryThatThrowsOnPut(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(expected = OnErrorNotImplementedException.class)
+    public void testEvictingMapFactoryIfMapPutThrowsOnErrorNotImplementedExceptionThenErrorEmittedByStream() {
+        
+        Exceptions.throwIfFatal(null);
+        final RuntimeException exception = new OnErrorNotImplementedException("boo", new RuntimeException());
+        //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 = createMapFactoryThatThrowsOnPut(exception);
+        TestSubscriber<Object> ts = TestSubscriber.create();
+        Observable
+            .range(1, 100)
+            .groupBy(EVICTING_MAP_KEY_SELECTOR, elementSelector, mapFactory)
+            .flatMap(UtilityFunctions.<Observable<Integer>>identity())
+            .subscribe(ts);
+    }
+
+    private Func1<Action1<Integer>, Map<Integer, Object>> createMapFactoryThatThrowsOnPut(
+            final RuntimeException exception) {
+        return new Func1<Action1<Integer>, Map<Integer, Object>>() {
             @SuppressWarnings("serial")
             @Override
             public Map<Integer, Object> call(final Action1<Integer> evicted) {
@@ -1895,17 +1933,61 @@ public Object put(Integer key, Object value) {
                         throw exception;
                     }};
             }};
-        TestSubscriber<Object> ts = TestSubscriber.create();
+    }
+    
+    @Test
+    public void mapFactoryEvictionWorksWithGuavaCache() {
+        final List<Integer> evictedKeys = new ArrayList<Integer>();
+        Func1<Action1<Integer>, Map<Integer, Object>> mapFactory =
+            new Func1<Action1<Integer>, Map<Integer, Object>>() {
+                @Override
+                public Map<Integer, Object> call(final Action1<Integer> action) {
+                    return CacheBuilder.newBuilder()
+                        .maximumSize(5)
+                        .removalListener(new RemovalListener<Integer, Object>() {
+                            @Override
+                            public void onRemoval(RemovalNotification<Integer, Object> notification) {
+                                action.call(notification.getKey());
+                                evictedKeys.add(notification.getKey());                            
+                            }
+                        })
+                        .build().asMap();
+                }
+            };
+        Func1<Integer, Integer> elementSelector = UtilityFunctions.identity();
+        TestSubscriber<String> ts = TestSubscriber.create();
         Observable
             .range(1, 100)
             .groupBy(EVICTING_MAP_KEY_SELECTOR, elementSelector, mapFactory)
-            .flatMap(UtilityFunctions.<Observable<Integer>>identity())
+            .flatMap(new Func1<GroupedObservable<Integer, Integer>, Observable<String>>() {
+                @Override
+                public Observable<String> call(final GroupedObservable<Integer, Integer> g) {
+                    return g.map(new Func1<Integer, String>() {
+                        @Override
+                        public String call(Integer x) {
+                            return g.getKey() + ":" + x;
+                        }
+                    });
+                }
+            })
             .subscribe(ts);
-        ts.assertError(exception);
+        assertEquals(Arrays.asList(0, 1, 2, 3, 4), evictedKeys.subList(0, 5));
+        List<String> expected = Observable
+                .range(1, 100)
+                .map(new Func1<Integer, String>() {
+                        @Override
+                        public String call(Integer x) {
+                            return (x /10) + ":" + x;
+                        }
+                    })
+                .toList().toBlocking().single();
+        assertEquals(expected, ts.getOnNextEvents());
     }
 
-    @Test
-    public void mapFactoryEvictionWorksWithGuavaCache() {
-        
+    @Test(expected = NullPointerException.class)
+    public void testGroupByThrowsNpeIfEvictingMapFactoryNull() {
+        Observable
+        .range(1, 100)
+        .groupBy(EVICTING_MAP_KEY_SELECTOR, elementSelector, null);
     }
 }