rx-java

0热度

2回答

有没有办法让一个Observable只抛出一个例外发生在像flatMap()功能？在Java中，在一个正常的功能，你可以指定它可能会抛出异常： public void dangerousFunction() throws IOException {} 在像flatMap()一个Observable功能，我似乎无法找到类似的语法。我可以用一个try/catch这样的： myObservable

0热度

1回答

使用RxJava的同步序列

我在如何使用RxJava表达以下方面挣扎。让我们想象一下我的包列表中的流，我想在网络上传输： PublishSubject<List<Packet>> packetsSubject = ...; 而且我有以下传输功能： public Observable<Status> transmit(Packet p) {...} 我想传递一个列表，每包只要包的返回状态是Status.OK。换句话说，

0热度

2回答

结合多个可观察结果

我试图实现这样的事情： - 进行两个不同的API调用并将结果组合到一个列表中。每次调用的结果都是一个HashMap，我有一个函数可以将它转换为一个列表。我现在想要做的是将列表合并为一个，但我遇到了困难。这是我的改装服务看起来像： @GET("data/price?fsym=ETH") fun getETHRates(@Query("tsyms") tsyms : String) : Singl

2热度

3回答

从正确的线程调用RxJava2可以取消/抛弃

我正在实现一个可观察的来自Resource的线。问题是，这个资源真的不喜欢从它创建的不同线程中关闭（它会杀死一只小狗并在发生这种情况时抛出一个异常）。当我处理认购，资源Cancellable/Disposable从main线程中调用，而观察到的是订阅的Schedulers.io()。这里是科特林代码： fun lines(): Observable<String> = Obser

0热度

1回答

试图注入到工作

类我有我的应用程序的网络组件，让我注入改装到我的活动&片段，我想把它注入到我的工作类，这里就是我所做的 NetComponent接口： @Singleton @Component(modules={AppModule.class, NetModule.class}) public interface NetComponent { void inject(MainActivity a

0热度

1回答

RxJava超时没有退订源

我订阅热可观察，然后应用timeout运营商，但我不想取消订阅TimeoutException时，只产生特殊项目（我知道源将最终发出新的项目）。我怎样才能做到这一点？我试图结合超时与onErrorReturn但这又导致onComplete呼叫用户。

5热度

1回答

异步，在Java矢量/流数据可组合的返回值9

说我有一个API，基于一些查询条件，会发现或构建一个小窗口： Widget getMatchingWidget(WidgetCriteria c) throws Throwable 的（同步）的客户端代码的外观像： try { Widget w = getMatchingWidget(criteria); processWidget(w); } catch (Throwa

2热度

1回答

字符串值如何传递给updateText（）方法？

与RxJava修修补补的第一次，我已经创建了一个的Rx链听在String对象的状态的变化： observable=Observable .defer(new Callable<ObservableSource<String>>() { @Override public ObservableSource<String> call() throws Exception {

0热度

2回答

使用observables改进并行响应处理

我正在开发一个应用程序，用户可以在注册后编辑他/她的配置文件。问题是编辑配置文件时会对后端使用多个API调用。一个用于编辑电子邮件，一个用于编辑密码等等。我将retrofit2与RxJava结合使用，因此呼叫是Observables/Singles。当用户按下编辑按钮并在不同线程上执行调用时，会显示一个微调器。如果它们并行执行，我该如何处理成功/失败？用户需要知道编辑是否成功，并在呼叫完成时隐藏微

0热度

2回答

单独播放合并流

我有一个ObservableTransformer由几个子流组成，如下例所示。 ObservableTransformer<Action, Result> actionsToResults = actions -> actions.publish(shared -> Observable.merge( Observable.merge(