7

我在使用改造的 RxJava 支持链接可观察对象时遇到问题。我可能误解了如何使用它，否则它可能是改造中的错误。希望这里有人可以帮助我了解发生了什么。编辑：我将 MockRestAdapter 用于这些响应 - 这可能是相关的，因为我看到 RxSupport 实现略有不同。

这是一个假的银行应用程序。它正在尝试进行转账，并且在转账完成后，它应该进行账户请求以更新账户值。这基本上只是我尝试flatMap的一个借口。不幸的是，以下代码不起作用，没有订阅者收到通知：

案例 1：链接两个改造产生的 observables

传输服务（注意：返回一个改造产生的 observable）：

@FormUrlEncoded @POST("/user/transactions/")
public Observable<TransferResponse> transfer(@Field("session_id") String sessionId,
                                             @Field("from_account_number") String fromAccountNumber,
                                             @Field("to_account_number") String toAccountNumber,
                                             @Field("amount") String amount);

帐户服务（注意：返回一个改造产生的可观察对象）：

@FormUrlEncoded @POST("/user/accounts")
public Observable<List<Account>> getAccounts(@Field("session_id") String sessionId);

将两个改造产生的 observables 链接在一起：

transfersService.transfer(session.getSessionId(), fromAccountNumber, toAccountNumber, amount)
            .flatMap(new Func1<TransferResponse, Observable<? extends List<Account>>>() {
                @Override public Observable<? extends List<Account>> call(TransferResponse transferResponse) {
                    return accountsService.getAccounts(session.getSessionId());
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());

案例 2：创建我自己的 observable 并与改造生产的一个链接

如果我忽略 Retrofit 中对“平面映射”调用的内置 Rx 支持，它会完美运行！所有订阅者都会收到通知。见下文：

新的帐户服务（注意：不产生可观察的）：

@FormUrlEncoded @POST("/user/accounts")
public List<Account> getAccountsBlocking(@Field("session_id") String sessionId);

创建我自己的 observable 并自己发出项目：

transfersService.transfer(session.getSessionId(), fromAccountNumber, toAccountNumber, amount)
            .flatMap(new Func1<TransferResponse, Observable<? extends List<Account>>>() {
                @Override public Observable<? extends List<Account>> call(TransferResponse transferResponse) {
                    return Observable.create(new Observable.OnSubscribe<List<Account>>() {
                        @Override public void call(Subscriber<? super List<Account>> subscriber) {
                            List<Account> accounts = accountsService.getAccountsBlocking(session.getSessionId());
                            subscriber.onNext(accounts);
                            subscriber.onCompleted();
                        }
                    });
                }
            })
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread());

任何帮助将不胜感激！

4

1 回答 1

5

答案是肯定的，你应该能够从 Retrofit 链接 observables。MockRestAdapter$MockRxSupport:createMockObservable 私有类中似乎存在错误。将订阅者订阅到 observable 的调度方式似乎是错误的。订阅 observable 是在 HttpExecutor 线程本身启动之后进行的。我相信在可以订阅 mockHandler.invokeSync 返回的 Observable 之前，来自您的 Schedulers.io() 线程的原始流程已完成并取消订阅。如果你看一下改造模拟模块中的代码，希望这个解释有某种意义。

作为目前使用改造模拟时的当前代码的解决方法，您可以用您自己的 ImmediateExecutor 实现替换内部默认 Executor。这至少允许在测试模拟时拥有一个由 Schedulers.io 提供的单线程流。

// ImmediateExecutor.java
public class ImmediateExecutor implements Executor {
    @Override
    public void execute(Runnable command) {
        command.run();
    }
}

// Create your RestAdapter with your ImmdiateExecutor
RestAdapter adapter = new RestAdapter.Builder()
            .setEndpoint(endpoint)
            .setExecutors(new ImmediateExecutor(), null)
            .build();

至于从源头解决问题，您还可以将改造模拟项目作为源代码包含在您的项目中，并使用以下代码修改 MockRestAdapter$MockRxSupport:createMockObservable 方法。我已经测试了您的用例，它确实解决了问题。

--- MockRestAdapter.java$MockRxSupport ----

Observable createMockObservable(final MockHandler mockHandler, final RestMethodInfo methodInfo,
        final RequestInterceptor interceptor, final Object[] args) {
      return Observable.create(new Observable.OnSubscribe<Object>() {
        @Override public void call(final Subscriber<? super Object> subscriber) {
          try {
            if (subscriber.isUnsubscribed()) return;
            Observable observable =
                (Observable) mockHandler.invokeSync(methodInfo, interceptor, args);

            observable.subscribeOn(Schedulers.from(httpExecutor));

            //noinspection unchecked
            observable.subscribe(subscriber);

          } catch (RetrofitError e) {
            subscriber.onError(errorHandler.handleError(e));
          } catch (Throwable e) {
            subscriber.onError(e);
          }
        }
      });
    }

在这里创建了改造项目的问题，我们会看看他们是否接受它。

于 2014-07-15T14:44:20.387 回答