RxJava Android wait for callback to finish before returning data

Question

I'm new to RxJava. I have an Android app that is using AWS Cognito SDK for authentication. I have an AwsAuthClient class that handles calling the SDK and returning the results. I have a fragment that calls the SignUp method in AwsAuthClient. I need to return the results of signup to the fragment so that it can react appropriately.

RegisterFragment class:

public class RegisterFragment{
    AwsAuthClient authClient;

    public void onCreateAccountClick() {
        Subscription createSubscription = authClient.SignUp(params)
            .compose(Transformers.applyIoToMainSchedulers())
            .subscribe((CognitoUser currentUser) -> {
                transitionToVerificationScreen();
             }, (Throwable throwable) -> {
                 // Report the error.
             });
    }
}

Here is the AwsAuthClient:

public class AwsAuthClient {

    public void SignUp(CreateParams createParams){

        // Create a CognitoUserAttributes object and add user attributes
        CognitoUserAttributes userAttributes = new CognitoUserAttributes();

        // Add the user attributes. Attributes are added as key-value pairs
        // Adding user's given name.
        // Note that the key is "given_name" which is the OIDC claim for given name
        userAttributes.addAttribute("given_name", createParams.getFirstname() + " " + createParams.getLastname());

        // Adding user's phone number
        userAttributes.addAttribute("phone_number", createParams.getPhone());

        // Adding user's email address
        userAttributes.addAttribute("email", createParams.getPhone());

        SignUpHandler signupCallback = new SignUpHandler() {

            @Override
            public void onSuccess(CognitoUser cognitoUser, boolean userConfirmed, CognitoUserCodeDeliveryDetails cognitoUserCodeDeliveryDetails) {
                // Sign-up was successful

                currentUser = cognitoUser;

                // Check if this user (cognitoUser) needs to be confirmed
                if(!userConfirmed) {
                    // This user must be confirmed and a confirmation code was sent to the user
                    // cognitoUserCodeDeliveryDetails will indicate where the confirmation code was sent
                    // Get the confirmation code from user
                    Timber.d("Sent confirmation code");
                }
                else {
                    // The user has already been confirmed
                    Timber.d("User has already been confirmed.");
                }
            }

            @Override
            public void onFailure(Exception exception) {
                // Sign-up failed, check exception for the cause
            }
        };

        userPool.signUpInBackground(userId, password, userAttributes, null, signupCallback);
    }
}

How can I return the results of onSuccess or OnFailure up to the RegisterFragment class?

Answer 1

It looks like the Cognito SDK already provides an async way to get information. In order for you to wrap this into an rx stream, you should consider using a Subject.

Subject are both Observables capable of emitting data, and Observers capable of receiving data. A Subject can wait to receive the callback data, take the data, and then emit it onto a stream.

public Observable<CognitoUser> SignUp(CreateParams createParams){
    BehaviorSubject<CognitoUser> subject = BehaviorSubject.create();

    // ...

    SignUpHandler signupCallback = new SignUpHandler() {

        @Override
        public void onSuccess(CognitoUser cognitoUser, boolean userConfirmed, CognitoUserCodeDeliveryDetails cognitoUserCodeDeliveryDetails) {
            // Sign-up was successful

            // Check if this user (cognitoUser) needs to be confirmed
            if(!userConfirmed) {
                // This user must be confirmed and a confirmation code was sent to the user
                // cognitoUserCodeDeliveryDetails will indicate where the confirmation code was sent
                // Get the confirmation code from user
                Timber.d("Sent confirmation code");
            }
            else {
                // The user has already been confirmed
                Timber.d("User has already been confirmed.");
            }

            subject.onNext(cognitoUser);
            subject.onComplete();
        }

        @Override
        public void onFailure(Exception exception) {
            subject.onError(exception);
        }
    };

    userPool.signUpInBackground(userId, password, userAttributes, null, signupCallback);
    return subject;
}

Answer 2

If you are using RxJava2. You can use the create() operator to create your own async call:

public class AwsAuthClient {

    public Observable<CognitoUser> SignUp(CreateParams createParams){
        return Observable.create(emitter -> {
            SignUpHandler signupCallback = new SignUpHandler() {

                @Override
                public void onSuccess(CognitoUser cognitoUser, boolean userConfirmed, CognitoUserCodeDeliveryDetails cognitoUserCodeDeliveryDetails) {
                // Sign-up was successful
                emitter.onNext(cognitoUser);
                // Check if this user (cognitoUser) needs to be confirmed
                if(!userConfirmed) {
                // This user must be confirmed and a confirmation code was sent to the user
                // cognitoUserCodeDeliveryDetails will indicate where the confirmation code was sent
                // Get the confirmation code from user
                Timber.d("Sent confirmation code");
                }
                else {
                    // The user has already been confirmed
                    Timber.d("User has already been confirmed.");
                }
                emitter.onComplete();
            }

            @Override
            public void onFailure(Exception exception) {
                // Sign-up failed, check exception for the cause
                emitter.onError(exception);
            }
        };
        //cancel the call
        Observable.setCancellable(//your cancel code)
    })
}

Edit: If you are using RxJava1(latest version 1.3.2) you can just use Observable.create(Action1>,BackPressureMode) instead of create and it's safe

        Observable.create(new Action1<Emitter<CognitoUser extends Object>>() {
        @Override
        public void call(Emitter<CognitoUser> emitter) {
            SignUpHandler signupCallback = new SignUpHandler() {
                @Override
                public void onSuccess(CognitoUser cognitoUser, boolean userConfirmed, CognitoUserCodeDeliveryDetails cognitoUserCodeDeliveryDetails) {
                    if (!userConfirmed) {
                        Timber.d("Sent confirmation code");
                    } else {
                        Timber.d("User has already been confirmed.");
                    }
                    emitter.onNext(cognitoUser);

                    emitter.onComplete();
                }

                @Override
                public void onFailure(Exception exception) {
                    emitter.onError(exception);
                }
            };
            emitter.setCancellation(new Cancellable() {
                @Override
                public void cancel() throws Exception {
                    //Your Cancellation
                }
            });
            signUpInBackground(userId, password, userAttributes, null, signupCallback);
        }

        //Because RxJava 1 doesn't have Flowable so you need add backpressure by default.
    }, Emitter.BackpressureMode.NONE );