fix for #1042

aol · Mar 6, 2019 · 486523c · 486523c
1 parent 6a11519
commit 486523c
Show file tree

Hide file tree

Showing 8 changed files with 87 additions and 10 deletions.
diff --git a/...-futurestream/src/main/java/com/oath/cyclops/types/futurestream/BlockingStreamHelper.java b/...-futurestream/src/main/java/com/oath/cyclops/types/futurestream/BlockingStreamHelper.java
@@ -100,6 +100,33 @@ public static Object getSafe(final FastFuture next, final Optional<Consumer<Thro
 
         return MissingValue.MISSING_VALUE;
     }
+    @SuppressWarnings("rawtypes")
+    public static Object extractNonFiltered(final FastFuture next, final Optional<Consumer<Throwable>> errorHandler) {
+
+        try {
+            return next.join();
+        } catch (final SimpleReactCompletionException e) {
+            Throwable t = e;
+
+            while(t instanceof SimpleReactCompletionException){
+                t= t.getCause();
+            }
+            while(t instanceof SimpleReactFailedStageException){
+                t= t.getCause();
+            }
+            if(t instanceof FilteredExecutionPathException){
+                return MissingValue.MISSING_VALUE;
+            }else{
+                throw ExceptionSoftener.throwSoftenedException(t);
+            }
+        } catch (final RuntimeException e) {
+            capture(e, errorHandler);
+        } catch (final Exception e) {
+            capture(e, errorHandler);
+        }
+
+        return MissingValue.MISSING_VALUE;
+    }
 
     @SuppressWarnings("rawtypes")
     static Object getSafe(final CompletableFuture next, final Optional<Consumer<Throwable>> errorHandler) {

diff --git a/cyclops-futurestream/src/main/java/com/oath/cyclops/types/futurestream/LazyStream.java b/cyclops-futurestream/src/main/java/com/oath/cyclops/types/futurestream/LazyStream.java
@@ -137,7 +137,8 @@ default <A, R> R run(final Collector<U, A, R> collector) {
 
         return (R) batcher.getAllResults()
                           .stream()
-                          .map(cf -> BlockingStreamHelper.getSafe(cf, getErrorHandler()))
+            .map(cf -> BlockingStreamHelper.extractNonFiltered(cf, getErrorHandler()))
+                          //.map(cf -> BlockingStreamHelper.getSafe(cf, getErrorHandler()))
                           .filter(v -> v != MissingValue.MISSING_VALUE)
                           .collect((Collector) collector);
 

diff --git a/cyclops-futurestream/src/main/java/cyclops/futurestream/FutureStream.java b/cyclops-futurestream/src/main/java/cyclops/futurestream/FutureStream.java
@@ -2660,9 +2660,7 @@ default Set<U> toSet() {
         return collect(Collectors.toSet());
     }
 
-    /*
-     * @see cyclops2.stream.ReactiveSeq#toList()
-     */
+
     @Override
     default List<U> toList() {
         return collect(Collectors.toList());

diff --git a/cyclops-futurestream/src/test/java/cyclops/futurestream/FutureStreamReactiveSeqTest.java b/cyclops-futurestream/src/test/java/cyclops/futurestream/FutureStreamReactiveSeqTest.java
@@ -9,10 +9,12 @@
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasItems;
 import static org.junit.Assert.assertThat;
+import static org.junit.Assert.fail;
 
 public class FutureStreamReactiveSeqTest extends AbstractReactiveSeqTest {
     @Override
@@ -26,6 +28,7 @@ public ReactiveSeq<Integer> empty() {
         return FutureStream.builder()
                             .of();
     }
+
     @Test @Override
     public void recoverWithMiddleIterator(){
 
@@ -64,6 +67,49 @@ public void recoverWithMiddleList(){
         assertThat(result,hasItems(100,200,300));
     }
 
+    @Test
+    public void onErrorList(){
+        AtomicInteger count = new AtomicInteger(0);
+
+        try {
+            of(1, 2, 3)/**.filter(i->false)**/
+            .map(i -> {
+                throw new RuntimeException();
+            })
+                .onError(e -> count.incrementAndGet())
+                .toList();
+            fail("exception  expected");
+        }catch(Exception e){
+
+        }
+
+
+        assertThat(count.get(),equalTo(3));
+
+    }
+    @Test
+    public void onErrorIterator(){
+        AtomicInteger count = new AtomicInteger(0);
+
+        try {
+            Iterator<Integer> it = of(1, 2, 3).<Integer>map(i -> {
+                throw new RuntimeException();
+            })
+                .onError(e -> count.incrementAndGet())
+                .iterator();
+            while(it.hasNext()){
+                System.out.println(it.next());
+            }
+            fail("exception  expected");
+        }catch(Exception e){
+
+
+        }
+
+
+        assertThat(count.get(),equalTo(3));
+
+    }
 
 
 }
diff --git a/cyclops-futurestream/src/test/java/cyclops/futurestream/react/base/BaseSeqTest.java b/cyclops-futurestream/src/test/java/cyclops/futurestream/react/base/BaseSeqTest.java
@@ -457,7 +457,7 @@ public void testDuplicate(){
 
 
 	 Throwable ex;
-	    @Test
+	    @Test(expected = ClassCastException.class)
 	    public void testCastException() {
 	    	ex = null;
 	    	of(1, "a", 2, "b", 3, null).capture(e-> ex =e)

diff --git a/...ops-futurestream/src/test/java/cyclops/futurestream/react/base/BaseSequentialSeqTest.java b/...ops-futurestream/src/test/java/cyclops/futurestream/react/base/BaseSequentialSeqTest.java
@@ -447,7 +447,7 @@ public void testDuplicate(){
 
 
 	 Throwable ex;
-	    @Test
+	    @Test(expected =  ClassCastException.class)
 	    public void testCastException() {
 	    	ex = null;
 	    	of(1, "a", 2, "b", 3, null).capture(e-> ex =e)

diff --git a/...s-futurestream/src/test/java/cyclops/futurestream/react/lazy/sequence/LFSNoOrderTest.java b/...s-futurestream/src/test/java/cyclops/futurestream/react/lazy/sequence/LFSNoOrderTest.java
@@ -238,10 +238,15 @@ public void testQuadriplicateLimit(){
 	}
 	    @Test
 	    public void testCastException() {
-	    	of(1, "a", 2, "b", 3, null)
-	    			.peek(it ->System.out.println(it))
+	    	of(1, "a", 2, "b", 3)
+	    			.peek(it ->System.out.println(" it is " +it))
 	    			.cast(Integer.class)
-	    			.peek(i->System.out.println(i.getClass()))
+                    .recover(t->{
+                        return -10;
+                    })
+	    			.peek(i->{
+	    			    System.out.println(i.getClass());
+                    })
 	    				.peek(it ->System.out.println(it))
 	    				.toList()
 	    				.stream().map(i->i.getClass())

diff --git a/...s-futurestream/src/test/java/cyclops/futurestream/react/lazy/sequence/SequentialTest.java b/...s-futurestream/src/test/java/cyclops/futurestream/react/lazy/sequence/SequentialTest.java
@@ -311,7 +311,7 @@ public void testIntersperse() {
 
 
 	}
-	@Test
+	@Test(expected = ClassCastException.class)
 	public void cast(){
 		LazyReact.sequentialBuilder().of(1,2,3).cast(String.class).collect(Collectors.toList())
 		.stream().map(i->i.getClass())