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

Question

我是 RxJava 的新手。我有一个 Android 应用程序正在使用 AWS Cognito SDK 进行身份验证。我有一个 AwsAuthClient class 来处理调用 SDK 和 returning 结果。我有一个在 AwsAuthClient 中调用 SignUp 方法的片段。我需要 return 片段的注册结果，以便它可以做出适当的反应。

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

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

如何 return onSuccess 或 OnFailure 的结果到 RegisterFragment class？

Answer 1

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

Subject都是能够发送数据的Observable，和能够接收数据的Observer。 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) 而不是创建，它是 安全的

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

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

RxJava Android wait for callback to finish before returning data

android

rx-java

rx-android