RxJava Android 在返回数据之前等待回调完成

Question

我是 RxJava 新手。我有一个使用 AWS Cognito SDK 进行身份验证的 Android 应用程序。我有一个AwsAuthClient 类来处理调用 SDK 并返回结果。我有一个片段调用AwsAuthClient 中的SignUp 方法。我需要将注册结果返回给片段，以便它能够做出适当的反应。

RegisterFragment 类：

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.
             });
    }
}

这是 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);
    }
}

如何将 onSuccess 或 OnFailure 的结果返回到 RegisterFragment 类？

Answer 1

看起来 Cognito SDK 已经提供了一种异步方式来获取信息。为了让您将其包装到 rx 流中，您应该考虑使用 Subject。

Subject 既是Observables 能够发出数据，又是Observers 能够接收数据。 Subject 可以等待接收回调数据，获取数据，然后将其发送到流中。

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

如果您使用的是 RxJava2。您可以使用 create() 运算符创建自己的异步调用：

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)
    })
}

编辑：如果您使用的是 RxJava1（最新版本 1.3.2），您可以只使用 Observable.create(Action1>,BackPressureMode) 而不是 create，它是 安全

        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 );