feat: port example Simple and Advanced exponential backoff (using delay and retryWhen)

Author: Kaushik Gopal

app/src/main/java/com/morihacky/android/rxjava/fragments/ExponentialBackoffFragment.java

@@ -7,42 +7,34 @@
 import android.view.View;
 import android.view.ViewGroup;
 import android.widget.ListView;
-
+import butterknife.Bind;
+import butterknife.ButterKnife;
+import butterknife.OnClick;
 import com.morihacky.android.rxjava.R;
 import com.morihacky.android.rxjava.wiring.LogAdapter;
-
+import hu.akarnokd.rxjava.interop.RxJavaInterop;
+import io.reactivex.Flowable;
+import io.reactivex.disposables.CompositeDisposable;
+import io.reactivex.functions.Function;
+import io.reactivex.subscribers.DisposableSubscriber;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-
-import butterknife.Bind;
-import butterknife.ButterKnife;
-import butterknife.OnClick;
-import rx.Observable;
-import rx.Observer;
-import rx.Subscriber;
-import rx.functions.Func1;
+import org.reactivestreams.Publisher;
 import rx.observables.MathObservable;
-import rx.subscriptions.CompositeSubscription;
 import timber.log.Timber;
 
+
 import static android.os.Looper.getMainLooper;
 
 public class ExponentialBackoffFragment
       extends BaseFragment {
 
     @Bind(R.id.list_threading_log) ListView _logList;
     private LogAdapter _adapter;
+    private CompositeDisposable _disposables = new CompositeDisposable();
     private List<String> _logs;
 
-    private CompositeSubscription _subscriptions = new CompositeSubscription();
-
-    @Override
-    public void onActivityCreated(@Nullable Bundle savedInstanceState) {
-        super.onActivityCreated(savedInstanceState);
-        _setupLogger();
-    }
-
     @Override
     public View onCreateView(LayoutInflater inflater,
                              @Nullable ViewGroup container,
@@ -52,11 +44,17 @@ public View onCreateView(LayoutInflater inflater,
         return layout;
     }
 
+    @Override
+    public void onActivityCreated(@Nullable Bundle savedInstanceState) {
+        super.onActivityCreated(savedInstanceState);
+        _setupLogger();
+    }
+
     @Override
     public void onPause() {
         super.onPause();
 
-        _subscriptions.clear();
+        _disposables.clear();
     }
 
     @Override
@@ -72,27 +70,30 @@ public void startRetryingWithExponentialBackoffStrategy() {
         _logs = new ArrayList<>();
         _adapter.clear();
 
-        _subscriptions.add(//
-              Observable//
-                    .error(new RuntimeException("testing")) // always fails
-                    .retryWhen(new RetryWithDelay(5, 1000)) // notice this is called only onError (onNext values sent are ignored)
-                    .doOnSubscribe(() -> _log("Attempting the impossible 5 times in intervals of 1s"))//
-                    .subscribe(new Observer<Object>() {
-                        @Override
-                        public void onCompleted() {
-                            Timber.d("on Completed");
-                        }
-
-                        @Override
-                        public void onError(Throwable e) {
-                            _log("Error: I give up!");
-                        }
-
-                        @Override
-                        public void onNext(Object aVoid) {
-                            Timber.d("on Next");
-                        }
-                    }));
+        DisposableSubscriber<Object> disposableSubscriber = new DisposableSubscriber<Object>() {
+            @Override
+            public void onNext(Object aVoid) {
+                Timber.d("on Next");
+            }
+
+            @Override
+            public void onComplete() {
+                Timber.d("on Completed");
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                _log("Error: I give up!");
+            }
+        };
+
+        Flowable.error(new RuntimeException("testing")) // always fails
+                .retryWhen(new RetryWithDelay(5, 1000)) // notice this is called only onError (onNext
+                // values sent are ignored)
+                .doOnSubscribe(subscription -> _log("Attempting the impossible 5 times in intervals of 1s"))
+                .subscribe(disposableSubscriber);
+
+        _disposables.add(disposableSubscriber);
     }
 
     @OnClick(R.id.btn_eb_delay)
@@ -101,41 +102,41 @@ public void startExecutingWithExponentialBackoffDelay() {
         _logs = new ArrayList<>();
         _adapter.clear();
 
-        _subscriptions.add(//
-
-              Observable.range(1, 4)//
-                    .delay(integer -> {
-                        // Rx-y way of doing the Fibonnaci :P
-                        return MathObservable//
-                              .sumInteger(Observable.range(1, integer))
-                              .flatMap(targetSecondDelay -> Observable.just(integer)
-                                    .delay(targetSecondDelay, TimeUnit.SECONDS));
-                    })//
-                    .doOnSubscribe(() ->
-                          _log(String.format("Execute 4 tasks with delay - time now: [xx:%02d]",
-                                _getSecondHand())))//
-                    .subscribe(new Subscriber<Integer>() {
-                        @Override
-                        public void onCompleted() {
-                            Timber.d("onCompleted");
-                            _log("Completed");
-                        }
-
-                        @Override
-                        public void onError(Throwable e) {
-                            Timber.d(e, "arrrr. Error");
-                            _log("Error");
-                        }
-
-                        @Override
-                        public void onNext(Integer integer) {
-                            Timber.d("executing Task %d [xx:%02d]", integer, _getSecondHand());
-                            _log(String.format("executing Task %d  [xx:%02d]",
-                                  integer,
-                                  _getSecondHand()));
-
-                        }
-                    }));
+        DisposableSubscriber<Integer> disposableSubscriber = new DisposableSubscriber<Integer>() {
+            @Override
+            public void onNext(Integer integer) {
+                Timber.d("executing Task %d [xx:%02d]", integer, _getSecondHand());
+                _log(String.format("executing Task %d  [xx:%02d]", integer, _getSecondHand()));
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                Timber.d(e, "arrrr. Error");
+                _log("Error");
+            }
+
+            @Override
+            public void onComplete() {
+                Timber.d("onCompleted");
+                _log("Completed");
+            }
+        };
+
+        Flowable
+              .range(1, 4)
+              .delay(integer -> {
+                  // Rx-y way of doing the Fibonnaci :P
+                  return RxJavaInterop
+                        .toV2Flowable(MathObservable.sumInteger(rx.Observable.range(1, integer)))
+                        .flatMap(targetSecondDelay -> Flowable
+                              .just(integer)
+                              .delay(targetSecondDelay, TimeUnit.SECONDS));
+              })
+              .doOnSubscribe(s -> _log(String.format("Execute 4 tasks with delay - time now: [xx:%02d]",
+                                                     _getSecondHand())))
+              .subscribe(disposableSubscriber);
+
+        _disposables.add(disposableSubscriber);
     }
 
     // -----------------------------------------------------------------------------------
@@ -176,7 +177,7 @@ private void _log(String logMsg) {
 
     //public static class RetryWithDelay
     public class RetryWithDelay
-          implements Func1<Observable<? extends Throwable>, Observable<?>> {
+          implements Function<Flowable<? extends Throwable>, Publisher<?>> {
 
         private final int _maxRetries;
         private final int _retryDelayMillis;
@@ -193,14 +194,14 @@ public RetryWithDelay(final int maxRetries, final int retryDelayMillis) {
         // only onNext triggers a re-subscription (onError + onComplete kills it)
 
         @Override
-        public Observable<?> call(Observable<? extends Throwable> inputObservable) {
+        public Publisher<?> apply(Flowable<? extends Throwable> inputObservable) {
 
             // it is critical to use inputObservable in the chain for the result
             // ignoring it and doing your own thing will break the sequence
 
-            return inputObservable.flatMap(new Func1<Throwable, Observable<?>>() {
+            return inputObservable.flatMap(new Function<Throwable, Publisher<?>>() {
                 @Override
-                public Observable<?> call(Throwable throwable) {
+                public Publisher<?> apply(Throwable throwable) {
                     if (++_retryCount < _maxRetries) {
 
                         // When this Observable calls onNext, the original
@@ -209,15 +210,14 @@ public Observable<?> call(Throwable throwable) {
                         Timber.d("Retrying in %d ms", _retryCount * _retryDelayMillis);
                         _log(String.format("Retrying in %d ms", _retryCount * _retryDelayMillis));
 
-                        return Observable.timer(_retryCount * _retryDelayMillis,
-                              TimeUnit.MILLISECONDS);
+                        return Flowable.timer(_retryCount * _retryDelayMillis, TimeUnit.MILLISECONDS);
                     }
 
                     Timber.d("Argh! i give up");
 
                     // Max retries hit. Pass an error so the chain is forcibly completed
                     // only onNext triggers a re-subscription (onError + onComplete kills it)
-                    return Observable.error(throwable);
+                    return Flowable.error(throwable);
                 }
             });
         }