时间戳

首页 > 解决方案 > RxJava：暂停代码执行直到项目被发出

java - RxJava：暂停代码执行直到项目被发出

问题描述

我刚开始学习 RxJava，最近一直在努力。我有一段代码需要使用 RxJava 完成，我不希望在调用我的观察者onSuccess()或观察者之前运行其余代码onError()。我目前的实现如下：

    @Override
    public Bundle getAuthToken(AccountAuthenticatorResponse accountAuthenticatorResponse, Account account, String authTokenType, Bundle options) {
        final AccountManager manager = AccountManager.get(mContext);
        final String username = account.name;
        String token = manager.peekAuthToken(account, authTokenType);

        if (TextUtils.isEmpty(token)){
            final String password = manager.getPassword(account);
            if (password != null){
                LoginClient client = ClientGenerator.createClient(LoginClient.class);
                String encodedString = encodeClientIDAndSecret();

                // Current implementation of observer
                Single<TokenResponse> single = client.getRxAccessToken(encodedString, LoginClient.GRANT_TYPE, account.name, password, LoginClient.SCOPE);
                single.map(new Function<TokenResponse, String>() {
                    @Override
                    public String apply(TokenResponse tokenResponse) throws Exception {
                        return tokenResponse.getAccessToken();
                    }
                })
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribe(new DisposableSingleObserver<String>() {
                            @Override
                            public void onSuccess(String accessToken) {
                                Log.d(TAG, "Here is the access token: " + accessToken);
                            }

                            @Override
                            public void onError(Throwable e) {
                                Log.d(TAG, "Unsuccessful response...");
                            }
                        });

            }
        }

        token = manager.peekAuthToken(account, authTokenType);
        if (!TextUtils.isEmpty(token)){
            final Bundle result = new Bundle();
            result.putString(AccountManager.KEY_ACCOUNT_NAME, account.name);
            result.putString(AccountManager.KEY_ACCOUNT_TYPE, account.type);
            result.putString(AccountManager.KEY_AUTHTOKEN, token);
            return result;
        } else {
            final Intent intent = new Intent(mContext, LoginActivity.class);
            intent.putExtra(AccountManager.KEY_ACCOUNT_AUTHENTICATOR_RESPONSE, accountAuthenticatorResponse);
            intent.putExtra(LoginActivity.EXTRA_ACCOUNT_TYPE, account.type);
            intent.putExtra(LoginActivity.EXTRA_AUTH_TOKEN_TYPE, authTokenType);

            final Bundle bundle = new Bundle();
            bundle.putParcelable(AccountManager.KEY_INTENT, intent);
            return bundle;
        }
    }

有什么方法可以阻止我的其余代码运行，直到我的观察者完成订阅？

为了提供一些更好的上下文，我试图实现的是 Android 的自定义帐户身份验证器。当用户第一次登录时，他们没有身份验证令牌，所以我使用 RxJava 获取了一个新令牌（如上所示）。然后，我需要返回Bundle到 Android AccountManager，这取决于我是否能够成功获取令牌。

我能够使用 RxJava 检索访问令牌。但是，如果我尝试在观察者之外读取令牌，则令牌为空，因为请求尚未完成......这就是为什么我希望以某种方式暂停其余代码的执行，直到订阅已经完成。

与往常一样，我们将非常感谢您在此特定问题上提供的任何帮助 :)

标签： javaandroidrx-javarx-java2

解决方案

如果这是你的单身

Single<TokenResponse> singleResponse = client.getRxAccessToken(
    encodedString,
    LoginClient.GRANT_TYPE,
    account.name,
    password,
    LoginClient.SCOPE
);

然后尝试以下操作（getAuthToken() 的返回类型需要更改）：

@Override
public Single<Bundle> getAuthToken(AccountAuthenticatorResponse accountAuthenticatorResponse, Account account, String authTokenType, Bundle options) {
    Return singleResponse
        .flatMap(tokenResponse -> createBundle(tokenResponse))
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
}

现在创建一个createBundle(TokenResponse tokenResponse)这样的方法

private Single<Bundle> createBundle(TokenResponse tokenResponse) {
    return Single.create(
        e -> {
            String token = tokenResponse.getAccessToken();
            manager.peekAuthToken(account, authTokenType);
            if (!TextUtils.isEmpty(token)){
                final Bundle result = new Bundle();
                result.putString(AccountManager.KEY_ACCOUNT_NAME, account.name);
                result.putString(AccountManager.KEY_ACCOUNT_TYPE, account.type);
                result.putString(AccountManager.KEY_AUTHTOKEN, token);

                e.onSuccess(result);
            } else {
                final Intent intent = new Intent(mContext, LoginActivity.class);
                intent.putExtra(
                    AccountManager.KEY_ACCOUNT_AUTHENTICATOR_RESPONSE, accountAuthenticatorResponse);
                intent.putExtra(LoginActivity.EXTRA_ACCOUNT_TYPE, account.type);
                intent.putExtra(LoginActivity.EXTRA_AUTH_TOKEN_TYPE, authTokenType);
                final Bundle bundle = new Bundle();
                bundle.putParcelable(AccountManager.KEY_INTENT, intent);

                e.onSuccess(bundle);
            }
        }
    );
}

现在订阅的人getAuthToken()将获得Bundle

getAuthToken()
    .subscribe(
        bundle -> //Use bundle as needed,
        error -> //Handle error scenario
    )

希望这个答案有帮助。

推荐阅读