added PreserveFullStackTraceOperator for better traceability

Author: Petikoch

build.gradle

@@ -28,6 +28,7 @@ def commonsLang3Version = '3.4'
 def groovyVersion = '2.4.5'
 def spockVersion = '1.0-groovy-2.4'
 def cglibVersion = '3.2.0'
+def awaitilityVersion = '1.7.0'
 def swingXVersion = '1.6.5-1'
 
 dependencies {
@@ -44,6 +45,7 @@ dependencies {
         exclude group: 'org.codehaus.groovy'
     }
     testCompile("cglib:cglib:${cglibVersion}")  // spock mocks
+    testCompile "com.jayway.awaitility:awaitility:${awaitilityVersion}"
 }
 
 sourceSets.main.java.srcDirs = ['src/main/java']

src/main/java/ch/petikoch/examples/mvvm_rxjava/example9/Example_9_View.java

@@ -24,6 +24,8 @@
 import java.awt.*;
 import java.lang.management.ManagementFactory;
 
+import static ch.petikoch.examples.mvvm_rxjava.utils.PreserveFullStackTraceOperator.preserveFullStackTrace;
+
 class Example_9_View extends StrictThreadingJFrame implements IView<Example_9_ViewModel> {
 
     private final Example_9_View_StatusPanel statusPanel = new Example_9_View_StatusPanel();
@@ -33,32 +35,35 @@ class Example_9_View extends StrictThreadingJFrame implements IView<Example_9_Vi
     @Override
     public void bind(final Example_9_ViewModel viewModel) {
         statusPanel.bind(viewModel.vm2v_status);
-        viewModel.vm2v_mainPanel.observeOn(SwingScheduler.getInstance()).subscribe(mainContentViewModel -> {
-            if (mainContentViewModel instanceof Example_9_ViewModel_Step1) {
-                Example_9_View_Step1Panel step1Panel = new Example_9_View_Step1Panel();
-                step1Panel.bind((Example_9_ViewModel_Step1) mainContentViewModel);
+        viewModel.vm2v_mainPanel
+                .observeOn(SwingScheduler.getInstance())
+                .lift(preserveFullStackTrace())
+                .subscribe(mainContentViewModel -> {
+                    if (mainContentViewModel instanceof Example_9_ViewModel_Step1) {
+                        Example_9_View_Step1Panel step1Panel = new Example_9_View_Step1Panel();
+                        step1Panel.bind((Example_9_ViewModel_Step1) mainContentViewModel);
 
-                mainPanel.removeAll();
-                mainPanel.add(step1Panel, BorderLayout.CENTER);
-                mainPanel.revalidate();
-            } else if (mainContentViewModel instanceof Example_9_ViewModel_Step2) {
-                Example_9_View_Step2Panel step2Panel = new Example_9_View_Step2Panel();
-                step2Panel.bind((Example_9_ViewModel_Step2) mainContentViewModel);
+                        mainPanel.removeAll();
+                        mainPanel.add(step1Panel, BorderLayout.CENTER);
+                        mainPanel.revalidate();
+                    } else if (mainContentViewModel instanceof Example_9_ViewModel_Step2) {
+                        Example_9_View_Step2Panel step2Panel = new Example_9_View_Step2Panel();
+                        step2Panel.bind((Example_9_ViewModel_Step2) mainContentViewModel);
 
-                mainPanel.removeAll();
-                mainPanel.add(step2Panel, BorderLayout.CENTER);
-                mainPanel.revalidate();
-            } else if (mainContentViewModel instanceof Example_9_ViewModel_Step3) {
-                Example__View_Step3Panel step3Panel = new Example__View_Step3Panel();
-                step3Panel.bind((Example_9_ViewModel_Step3) mainContentViewModel);
+                        mainPanel.removeAll();
+                        mainPanel.add(step2Panel, BorderLayout.CENTER);
+                        mainPanel.revalidate();
+                    } else if (mainContentViewModel instanceof Example_9_ViewModel_Step3) {
+                        Example__View_Step3Panel step3Panel = new Example__View_Step3Panel();
+                        step3Panel.bind((Example_9_ViewModel_Step3) mainContentViewModel);
 
-                mainPanel.removeAll();
-                mainPanel.add(step3Panel, BorderLayout.CENTER);
-                mainPanel.revalidate();
-            } else {
-                throw new IllegalStateException("Unhandled: " + mainContentViewModel);
-            }
-        });
+                        mainPanel.removeAll();
+                        mainPanel.add(step3Panel, BorderLayout.CENTER);
+                        mainPanel.revalidate();
+                    } else {
+                        throw new IllegalStateException("Unhandled: " + mainContentViewModel);
+                    }
+                });
     }
 
     public Example_9_View() {

src/main/java/ch/petikoch/examples/mvvm_rxjava/rxjava_mvvm/RxViewModel2SwingViewBinder.java

@@ -22,6 +22,8 @@
 import javax.swing.*;
 import javax.swing.text.JTextComponent;
 
+import static ch.petikoch.examples.mvvm_rxjava.utils.PreserveFullStackTraceOperator.preserveFullStackTrace;
+
 public class RxViewModel2SwingViewBinder {
 
     public static BooleanBindOfAble bindViewModelBoolean(Observable<Boolean> source) {
@@ -47,6 +49,7 @@ private BooleanBindOfAble(final Observable<Boolean> source) {
         public void toSwingViewEnabledPropertyOf(JComponent target) {
             source.onBackpressureLatest()
                     .observeOn(SwingScheduler.getInstance())
+                    .lift(preserveFullStackTrace())
                     .subscribe(target::setEnabled);
         }
     }
@@ -62,12 +65,14 @@ private StringBindOfAble(final Observable<String> source) {
         public void toSwingViewText(JTextComponent target) {
             source.onBackpressureLatest()
                     .observeOn(SwingScheduler.getInstance())
+                    .lift(preserveFullStackTrace())
                     .subscribe(target::setText);
         }
 
         public void toSwingViewLabel(JLabel target) {
             source.onBackpressureLatest()
                     .observeOn(SwingScheduler.getInstance())
+                    .lift(preserveFullStackTrace())
                     .subscribe(target::setText);
         }
     }
@@ -82,6 +87,7 @@ private BindOfAble(final Observable<T> source) {
 
         public void toAction(Action1<T> action) {
             source.observeOn(SwingScheduler.getInstance())
+                    .lift(preserveFullStackTrace())
                     .subscribe(action);
         }
     }

src/main/java/ch/petikoch/examples/mvvm_rxjava/utils/PreserveFullStackTraceOperator.java

@@ -0,0 +1,86 @@
+/**
+ * Copyright 2015 Peti Koch
+ *
+ * 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 ch.petikoch.examples.mvvm_rxjava.utils;
+
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import rx.Observable;
+import rx.Scheduler;
+import rx.Subscriber;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static java.util.Arrays.asList;
+
+/**
+ * A RxJava {@link rx.Observable.Operator} which takes care to preserve stacktraces across asynchronous
+ * {@link rx.Scheduler}-boundaries.
+ * <p>
+ * This {@link rx.Observable.Operator} is typically placed AFTER {@link Observable#observeOn(Scheduler)} or {@link
+ * Observable#subscribeOn(Scheduler)}.
+ * <p>
+ * See <a href="https://github.com/ReactiveX/RxJava/issues/3521#issuecomment-163571622">RxJava Issue 3521</a> for more
+ * background about the topic. This {@link rx.Observable.Operator} can be used as kind of a workaround for this unresolved
+ * RxJava issue (concerning RxJava versions <= 1.1.0 and probably also later versions).
+ * <p>
+ * Advantages: Analysing issues (stacktraces) gets much simpler for asynchronous use cases, since you have a "full" stacktrace<br>
+ * Disadvantage: Performance. Creating the internal RuntimeException-object to capture the stacktrace is "not cheap".
+ * But since we are operating at an asynchronous boundary, this doesn't add much additional overhead and can probably be neglected
+ */
+public class PreserveFullStackTraceOperator<T> implements Observable.Operator<T, T> {
+
+    private final RuntimeException asyncOriginStackTraceProvider = new RuntimeException("async origin");
+    private final long originThreadId = Thread.currentThread().getId(); // should be "enough" unique. See http://stackoverflow.com/a/591664/1662412
+
+    public static <T> PreserveFullStackTraceOperator<T> preserveFullStackTrace() {
+        return new PreserveFullStackTraceOperator<>();
+    }
+
+    @Override
+    public Subscriber<? super T> call(Subscriber<? super T> child) {
+        Subscriber<T> parent = new Subscriber<T>() {
+
+            @Override
+            public void onCompleted() {
+                child.onCompleted();
+            }
+
+            @Override
+            public void onError(Throwable throwable) {
+                if (Thread.currentThread().getId() != originThreadId) {
+                    List<StackTraceElement> originalStackTraceElements = Lists.newArrayList(throwable.getStackTrace());
+                    List<StackTraceElement> additionalAsyncOriginStackTraceElements = asList(asyncOriginStackTraceProvider.getStackTrace()).stream()
+                            .filter(stackTraceElement -> !PreserveFullStackTraceOperator.class.getName().equals(stackTraceElement.getClassName()))
+                            .collect(Collectors.toList());
+                    Iterable<StackTraceElement> modifiedStackTraceElements = Iterables.concat(originalStackTraceElements, additionalAsyncOriginStackTraceElements);
+                    throwable.setStackTrace(Iterables.toArray(modifiedStackTraceElements, StackTraceElement.class));
+                }
+                child.onError(throwable);
+            }
+
+            @Override
+            public void onNext(T t) {
+                child.onNext(t);
+            }
+        };
+
+        child.add(parent);
+
+        return parent;
+    }
+
+}

src/test/groovy/ch/petikoch/examples/mvvm_rxjava/utils/PreserveFullStackTraceOperatorTest.groovy

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2015 Peti Koch
+ *
+ * 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 ch.petikoch.examples.mvvm_rxjava.utils
+
+import com.jayway.awaitility.Awaitility
+import rx.Observable
+import rx.observers.TestSubscriber
+import rx.schedulers.Schedulers
+import spock.lang.Specification
+
+import java.util.concurrent.Callable
+
+import static ch.petikoch.examples.mvvm_rxjava.utils.PreserveFullStackTraceOperator.preserveFullStackTrace
+
+class PreserveFullStackTraceOperatorTest extends Specification {
+
+    def 'operator adds asynchronous origin stack trace elements'() {
+        given:
+        def asyncPipeline = Observable.fromCallable { throw new IllegalStateException("Boom!") }
+                .subscribeOn(Schedulers.io())
+        def originalThrowable = getThrowableFromFailingObservable(asyncPipeline)
+        assert originalThrowable instanceof IllegalStateException
+        assert originalThrowable.getMessage() == 'Boom!'
+        assert originalThrowable.getCause() == null
+
+        when:
+        def enhancedThrowable = getThrowableFromFailingObservable(
+                asyncPipeline.lift(preserveFullStackTrace())
+        )
+
+        then:
+        enhancedThrowable instanceof IllegalStateException
+        enhancedThrowable.getMessage() == 'Boom!'
+        enhancedThrowable.stackTrace.length > originalThrowable.stackTrace.length
+        enhancedThrowable.stackTrace.any { it.className == this.class.name } == true
+        enhancedThrowable.getCause() == null
+    }
+
+
+    def 'operator does not add additional stack trace elements if code is synchronous'() {
+        given:
+        def syncPipeline = Observable.fromCallable { throw new IllegalStateException("Boom!") }
+        def originalThrowable = getThrowableFromFailingObservable(syncPipeline)
+        assert originalThrowable instanceof IllegalStateException
+        assert originalThrowable.getMessage() == 'Boom!'
+        assert originalThrowable.getCause() == null
+
+        when:
+        def enhancedThrowable = getThrowableFromFailingObservable(
+                syncPipeline.lift(preserveFullStackTrace())
+        )
+
+        then:
+        enhancedThrowable instanceof IllegalStateException
+        enhancedThrowable.getMessage() == 'Boom!'
+        enhancedThrowable.stackTrace.length == originalThrowable.stackTrace.length
+        enhancedThrowable.getCause() == null
+    }
+
+    private Throwable getThrowableFromFailingObservable(Observable observable) {
+        TestSubscriber testSubscriber = new TestSubscriber()
+        observable.subscribe(testSubscriber)
+        Awaitility.await().until({ testSubscriber.getOnErrorEvents().size() == 1 } as Callable<Boolean>)
+        return testSubscriber.getOnErrorEvents().get(0)
+    }
+}