Fix issue #1

Don't use observeOn(Schedulers.io()) on "long living" Observables, since Schedulers.io() assigns one dedicated thread for the lifetime of the Observable subscription.
Even worse: If nobody evers unsubscribe, you have thread-leaks

Author: Petikoch

README.adoc

@@ -132,7 +132,7 @@ link:./src/main/java/ch/petikoch/examples/mvvm_rxjava/example2[]
 
 * A simple form submit of two textfields
 * The ViewModel combines the two textfield values into one DTO and calls the Model API on a IO-Thread
-* The `RxViewModel2ModelBinder` code does the switch to the `Schedulers.io()` scheduler
+* The `RxModelInvoker` code does the switch to the `Schedulers.io()` scheduler
 
 image::example2.png[]
 

build.gradle

@@ -7,7 +7,7 @@ apply plugin: 'java'
 apply plugin: 'eclipse'
 apply plugin: 'groovy'
 
-version = '1.1.1-SNAPSHOT'
+version = '1.2.0'
 
 def fileEncoding = 'UTF-8'
 

src/main/java/ch/petikoch/examples/mvvm_rxjava/example2/Example_2_ViewModel.java

@@ -46,7 +46,7 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_2_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(model::createAccount);
+        onEventFrom(vm2m_nameFirstname).executeAsync(model::createAccount);
     }
 
 }

src/main/java/ch/petikoch/examples/mvvm_rxjava/example3/Example_3_ViewModel.java

@@ -54,6 +54,6 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_3_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(model::createAccount);
+        onEventFrom(vm2m_nameFirstname).executeAsync(model::createAccount);
     }
 }

src/main/java/ch/petikoch/examples/mvvm_rxjava/example4/Example_4_ViewModel.java

@@ -63,7 +63,7 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_4_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(nameFirstname -> {
+        onEventFrom(vm2m_nameFirstname).executeAsync(nameFirstname -> {
             model.createAccount(nameFirstname);
 
             vm2v_nameEnabled.onNext(true);

src/main/java/ch/petikoch/examples/mvvm_rxjava/example5/Example_5_ViewModel.java

@@ -70,7 +70,7 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_5_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(nameFirstname -> {
+        onEventFrom(vm2m_nameFirstname).executeAsync(nameFirstname -> {
 
             Single<Boolean> modelCallFinished = Single.<Boolean>create(singleSubscriber -> {
                 try {

src/main/java/ch/petikoch/examples/mvvm_rxjava/example5a/Example_5a_ViewModel.java

@@ -68,7 +68,7 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_5a_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(nameFirstname -> {
+        onEventFrom(vm2m_nameFirstname).executeAsync(nameFirstname -> {
 
             Single.merge(model.createAccount(nameFirstname), v2vm_cancelButtonEvents.first().toSingle())
                     .toBlocking()

src/main/java/ch/petikoch/examples/mvvm_rxjava/example6/Example_6_ViewModel.java

@@ -77,7 +77,7 @@ private void wireInternally() {
 
     @Override
     public void connectTo(final Example_6_Model model) {
-        onEventFrom(vm2m_nameFirstname).execute(nameFirstname -> {
+        onEventFrom(vm2m_nameFirstname).executeAsync(nameFirstname -> {
 
             vm2v_resultat1.onNext("Creating acount...");
             vm2v_resultat2.onNext("Sending email...");

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

@@ -15,9 +15,9 @@
  */
 package ch.petikoch.examples.mvvm_rxjava.rxjava_mvvm;
 
+import ch.petikoch.examples.mvvm_rxjava.utils.AsyncUtils;
 import rx.Observable;
 import rx.functions.Action1;
-import rx.schedulers.Schedulers;
 
 public class RxModelInvoker {
 
@@ -33,10 +33,10 @@ private ModelInvoker(final Observable<T> source) {
             this.source = source;
         }
 
-        public void execute(final Action1<? super T> onNext) {
-            source.observeOn(Schedulers.io()).subscribe(onNext);
+        public void executeAsync(final Action1<? super T> onNext) {
+            source.subscribe(t -> {
+                AsyncUtils.executeAsync(() -> onNext.call(t));
+            });
         }
-
     }
-
 }