onErrorFlatMap + OnErrorThrowable

Author: benjchristensen

rxjava-core/src/main/java/rx/Observable.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.TimeUnit;
 
 import rx.exceptions.Exceptions;
+import rx.exceptions.OnErrorThrowable;
 import rx.exceptions.OnErrorNotImplementedException;
 import rx.functions.Action0;
 import rx.functions.Action1;
@@ -79,16 +80,13 @@
 import rx.operators.OperationMergeMaxConcurrent;
 import rx.operators.OperationMinMax;
 import rx.operators.OperationMulticast;
-import rx.operators.OperatorOnErrorResumeNextViaFunction;
 import rx.operators.OperationOnErrorResumeNextViaObservable;
 import rx.operators.OperationOnErrorReturn;
 import rx.operators.OperationOnExceptionResumeNextViaObservable;
 import rx.operators.OperationParallelMerge;
 import rx.operators.OperationReplay;
 import rx.operators.OperationRetry;
 import rx.operators.OperationSample;
-import rx.operators.OperatorObserveOnBounded;
-import rx.operators.OperatorScan;
 import rx.operators.OperationSequenceEqual;
 import rx.operators.OperationSingle;
 import rx.operators.OperationSkip;
@@ -117,8 +115,11 @@
 import rx.operators.OperatorMap;
 import rx.operators.OperatorMerge;
 import rx.operators.OperatorObserveOn;
+import rx.operators.OperatorOnErrorResumeNextViaFunction;
+import rx.operators.OperatorOnErrorFlatMap;
 import rx.operators.OperatorParallel;
 import rx.operators.OperatorRepeat;
+import rx.operators.OperatorScan;
 import rx.operators.OperatorSubscribeOn;
 import rx.operators.OperatorTake;
 import rx.operators.OperatorTimeout;
@@ -5209,7 +5210,7 @@ public final Boolean call(T t) {
      * @see <a href="https://github.com/Netflix/RxJava/wiki/Error-Handling-Operators#wiki-onerrorresumenext">RxJava Wiki: onErrorResumeNext()</a>
      */
     public final Observable<T> onErrorResumeNext(final Func1<Throwable, ? extends Observable<? extends T>> resumeFunction) {
-        return create(OperatorOnErrorResumeNextViaFunction.onErrorResumeNextViaFunction(this, resumeFunction));
+        return lift(new OperatorOnErrorResumeNextViaFunction<T>(resumeFunction));
     }
 
     /**
@@ -5267,6 +5268,15 @@ public final Observable<T> onErrorReturn(Func1<Throwable, ? extends T> resumeFun
         return create(OperationOnErrorReturn.onErrorReturn(this, resumeFunction));
     }
 
+    /**
+     * Allows inserting onNext events into a stream when onError events are received
+     * and continuing the original sequence instead of terminating. Thus it allows a sequence
+     * with multiple onError events.
+     */
+    public final Observable<T> onErrorFlatMap(final Func1<OnErrorThrowable, ? extends Observable<? extends T>> resumeFunction) {
+        return lift(new OperatorOnErrorFlatMap<T>(resumeFunction));
+    }
+
     /**
      * Instruct an Observable to pass control to another Observable rather than invoking
      * {@link Observer#onError onError} if it encounters an {@link java.lang.Exception}.

rxjava-core/src/main/java/rx/exceptions/OnErrorThrowable.java

@@ -0,0 +1,52 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.exceptions;
+
+public class OnErrorThrowable extends RuntimeException {
+
+    private static final long serialVersionUID = -569558213262703934L;
+
+    private final boolean hasValue;
+    private final Object value;
+
+    public OnErrorThrowable(Throwable exception) {
+        super(exception);
+        hasValue = false;
+        this.value = null;
+    }
+
+    public OnErrorThrowable(Throwable exception, Object value) {
+        super(exception);
+        hasValue = true;
+        this.value = value;
+    }
+
+    public Object getValue() {
+        return value;
+    }
+
+    public boolean isValueNull() {
+        return hasValue;
+    }
+
+    public static OnErrorThrowable from(Throwable t) {
+        if (t instanceof OnErrorThrowable) {
+            return (OnErrorThrowable) t;
+        } else {
+            return new OnErrorThrowable(t);
+        }
+    }
+}

rxjava-core/src/main/java/rx/operators/OperatorCast.java

@@ -17,7 +17,7 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
-
+import rx.exceptions.OnErrorThrowable;
 
 /**
  * Converts the elements of an observable sequence to the specified type.
@@ -46,7 +46,11 @@ public void onError(Throwable e) {
 
             @Override
             public void onNext(T t) {
-                o.onNext(castClass.cast(t));
+                try {
+                    o.onNext(castClass.cast(t));
+                } catch (Throwable e) {
+                    onError(new OnErrorThrowable(e, t));
+                }
             }
         };
     }

rxjava-core/src/main/java/rx/operators/OperatorDoOnEach.java

@@ -18,6 +18,7 @@
 import rx.Observable.Operator;
 import rx.Observer;
 import rx.Subscriber;
+import rx.exceptions.OnErrorThrowable;
 
 /**
  * Converts the elements of an observable sequence to the specified type.
@@ -59,7 +60,7 @@ public void onNext(T value) {
                 try {
                     doOnEachObserver.onNext(value);
                 } catch (Throwable e) {
-                    onError(e);
+                    onError(new OnErrorThrowable(e, value));
                     return;
                 }
                 observer.onNext(value);

rxjava-core/src/main/java/rx/operators/OperatorFilter.java

@@ -17,8 +17,8 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
-import rx.observables.GroupedObservable;
 
 /**
  * Filters an Observable by discarding any items it emits that do not meet some test.
@@ -48,13 +48,13 @@ public void onError(Throwable e) {
             }
 
             @Override
-            public void onNext(T value) {
+            public void onNext(T t) {
                 try {
-                    if (predicate.call(value)) {
-                        child.onNext(value);
+                    if (predicate.call(t)) {
+                        child.onNext(t);
                     }
-                } catch (Throwable ex) {
-                    child.onError(ex);
+                } catch (Throwable e) {
+                    child.onError(new OnErrorThrowable(e, t));
                 }
             }
 

rxjava-core/src/main/java/rx/operators/OperatorGroupBy.java

@@ -23,6 +23,7 @@
 import rx.Observable.OnSubscribe;
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Action0;
 import rx.functions.Func1;
 import rx.observables.GroupedObservable;
@@ -130,7 +131,7 @@ public void onNext(T t) {
                     // we have the correct group so send value to it
                     gps.onNext(t);
                 } catch (Throwable e) {
-                    onError(e);
+                    onError(new OnErrorThrowable(e, t));
                 }
             }
 

rxjava-core/src/main/java/rx/operators/OperatorMap.java

@@ -17,6 +17,7 @@
 
 import rx.Observable.Operator;
 import rx.Subscriber;
+import rx.exceptions.OnErrorThrowable;
 import rx.functions.Func1;
 
 /**
@@ -52,7 +53,7 @@ public void onNext(T t) {
                 try {
                     o.onNext(transformer.call(t));
                 } catch (Throwable e) {
-                    onError(e);
+                    onError(new OnErrorThrowable(e, t));
                 }
             }
 

rxjava-core/src/main/java/rx/operators/OperatorOnErrorFlatMap.java

@@ -0,0 +1,82 @@
+/**
+ * Copyright 2014 Netflix, Inc.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package rx.operators;
+
+import rx.Observable;
+import rx.Observable.Operator;
+import rx.Subscriber;
+import rx.exceptions.OnErrorThrowable;
+import rx.functions.Func1;
+
+/**
+ * Allows inserting onNext events into a stream when onError events are received
+ * and continuing the original sequence instead of terminating. Thus it allows a sequence
+ * with multiple onError events.
+ */
+public final class OperatorOnErrorFlatMap<T> implements Operator<T, T> {
+
+    private final Func1<OnErrorThrowable, ? extends Observable<? extends T>> resumeFunction;
+
+    public OperatorOnErrorFlatMap(Func1<OnErrorThrowable, ? extends Observable<? extends T>> f) {
+        this.resumeFunction = f;
+    }
+
+    @Override
+    public Subscriber<? super T> call(final Subscriber<? super T> child) {
+        return new Subscriber<T>(child) {
+
+            @Override
+            public void onCompleted() {
+                child.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                try {
+                    Observable<? extends T> resume = resumeFunction.call(OnErrorThrowable.from(e));
+                    resume.subscribe(new Subscriber<T>() {
+
+                        @Override
+                        public void onCompleted() {
+                            // ignore as we will continue the parent Observable
+                        }
+
+                        @Override
+                        public void onError(Throwable e) {
+                            // if the splice also fails we shut it all down
+                            child.onError(e);
+                        }
+
+                        @Override
+                        public void onNext(T t) {
+                            child.onNext(t);
+                        }
+
+                    });
+                } catch (Throwable e2) {
+                    child.onError(e2);
+                }
+            }
+
+            @Override
+            public void onNext(T t) {
+                child.onNext(t);
+            }
+
+        };
+    }
+
+}