package com.github.ajalt.reprint.reactive;


import android.support.annotation.NonNull;
import android.support.annotation.Nullable;

import com.github.ajalt.reprint.core.AuthenticationFailureReason;
import com.github.ajalt.reprint.core.AuthenticationListener;
import com.github.ajalt.reprint.core.Reprint;
import com.github.ajalt.reprint.core.ReprintModule;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;
import rx.functions.Func2;

/**
 * ReactiveX interface to reprint authentication.
 */
public class RxReprint {
    /**
     * Return an {@link Observable} whose {@link Subscriber#onNext(Object)} will be called at most
     * once.
     * <p>
     * The argument is the {@link ReprintModule#tag()} of the module that was used for
     * authentication. Any failures will cause {@link Subscriber#onNext(Object)} to be called with
     * an {@link AuthenticationFailure} containing the reason for the failure.
     * <p>
     * When either onNext or onFailure is called, the sensor will be off, so you will usually want
     * to resubscribe if the failure is non-fatal.
     */
    public static Observable<Integer> authenticate() {
        return Observable.create(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(final Subscriber<? super Integer> subscriber) {
                Reprint.authenticateWithoutRestart(new AuthenticationListener() {
                    private boolean listening = true;

                    @Override
                    public void onSuccess(int moduleTag) {
                        if (!listening) return;
                        listening = false;
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onNext(moduleTag);
                            subscriber.onCompleted();
                        }
                    }

                    @Override
                    public void onFailure(@NonNull AuthenticationFailureReason failureReason, boolean fatal, @Nullable CharSequence errorMessage, int moduleTag, int errorCode) {
                        if (!listening) return;
                        final AuthenticationFailure result = new AuthenticationFailure(failureReason, fatal, errorMessage, moduleTag, errorCode);
                        if (!subscriber.isUnsubscribed()) {
                            subscriber.onError(result);
                        }
                        if (fatal) {
                            listening = false;
                        }
                    }
                });
            }
        }).onBackpressureLatest().doOnUnsubscribe(new Action0() {
            @Override
            public void call() {
                Reprint.cancelAuthentication();
            }
        });
    }

    /**
     * Returns a predicate suitable for passing to {@link Observable#retry()} that will retry on a
     * non-fatal authentication error.
     * <p>
     * Since the observable returned by {@link #authenticate()} will terminate when a recoverable
     * error happens, it's a common pattern to report help text in {@link
     * Observable#doOnError(Action1)} and call {@link Observable#retry()} to restart the
     * subscription when the error is non-fatal.
     * <p>
     * This predicate will restart on non-fatal errors up to a specified number of failures.
     *
     * @param retryCount The maximum number of times to retry.
     */
    public static Func2<Integer, Throwable, Boolean> retryNonFatal(final int retryCount) {
        return new Func2<Integer, Throwable, Boolean>() {
            @Override
            public Boolean call(Integer count, Throwable throwable) {
                if (!(throwable instanceof AuthenticationFailure)) return false;
                AuthenticationFailure e = (AuthenticationFailure) throwable;
                return !e.fatal || e.failureReason == AuthenticationFailureReason.TIMEOUT && count < retryCount;
            }
        };
    }
}


@@ -348,24 +348,48 @@
 intModule;%0A%0A
+import rx.AsyncEmitter;%0A
 import rx.Ob
@@ -1239,41 +1239,42 @@
 ble.
-create(new Observable.OnSubscribe
+fromAsync(new Action1%3CAsyncEmitter
 %3CInt
@@ -1278,16 +1278,17 @@
 Integer%3E
+%3E
 () %7B%0A   
@@ -1345,27 +1345,21 @@
 nal 
-Subscriber%3C? super 
+AsyncEmitter%3C
 Inte
@@ -1363,24 +1363,21 @@
 nteger%3E 
-subscrib
+emitt
 er) %7B%0A  
@@ -1719,80 +1719,13 @@
     
-if (!subscriber.isUnsubscribed()) %7B%0A                            subscrib
+emitt
 er.o
@@ -1762,36 +1762,29 @@
             
-    subscrib
+emitt
 er.onComplet
@@ -1793,34 +1793,8 @@
 ();%0A
-                        %7D%0A
     
@@ -1935,16 +1935,58 @@
 eReason,
+%0A                                         
  boolean
@@ -2029,16 +2029,58 @@
 Message,
+%0A                                         
  int mod
@@ -2180,271 +2180,142 @@
     
-final AuthenticationFailure result = new AuthenticationFailure(failureReason, fatal, errorMessage, moduleTag, errorCode);%0A                        if (!subscriber.isUnsubscribed()) %7B%0A                            subscriber.onError(result);%0A                        %7D
+emitter.onError(new AuthenticationFailure(%0A                                failureReason, fatal, errorMessage, moduleTag, errorCode));
 %0A   
@@ -2490,12 +2490,23 @@
    %7D
-).on
+, AsyncEmitter.
 Back
@@ -2517,15 +2517,19 @@
 sure
-Latest(
+Mode.LATEST
 ).do

Use new Observable.fromAsync function.