我对RXJava一般来说相对较新(实际上只开始使用它与RXJava2),我发现的大多数文档往往是RXJava1; 我现在通常可以在两者之间进行转换,但整个Reactive的东西都很大,它是一个压倒性的API,有很好的文档(当你能找到它时).我正在尝试简化我的代码,我想用婴儿步骤来做.我想要解决的第一个问题是我在当前项目中做了很多常见的模式:
您有一个请求,如果成功,您将用于发出第二个请求.
如果其中一个失败,您需要能够识别哪个失败.(主要是显示自定义UI警报).
这就是我现在通常这样做的方式:
(.subscribeOn/observeOn为简单起见省略)
Single<FirstResponse> first = retrofitService.getSomething();
first
.subscribeWith(
new DisposableSingleObserver<FirstResponse>() {
@Override
public void onSuccess(final FirstResponse firstResponse) {
// If FirstResponse is OK…
Single<SecondResponse> second =
retrofitService
.getSecondResponse(firstResponse.id) //value from 1st
.subscribeWith(
new DisposableSingleObserver<SecondResponse>() {
@Override
public void onSuccess(final SecondResponse secondResponse) {
// we're done with both!
}
@Override
public void onError(final Throwable error) {
//2nd request Failed,
}
});
}
@Override
public void onError(final Throwable error) {
//firstRequest Failed,
}
});
在RXJava2中有更好的方法来处理这个问题吗？
我尝试过 …
Maybe<>如果可能不为空,我有一个源和一些我想用这个值执行的动作:
// Maybe<T> maybe();
// Completable action(T value);
return maybe().flatMapCompletable(val -> action(val));
但是当可能是空的时候我想要'完成'可以完成:
return Completable.complete();
如何进行此切换:如果可能不是空的,可以获得一个完成,否则另一个？
我有一个问题,理解为什么以下代码不起作用.我做错了什么或者是RxJava2实现中的某种错误？
private Disposable savedDisposable;
@Test
public void test() {
final TestObserver<Integer> observer = new TestObserver<>();
Observable<Integer> t = Observable.just(10)
.delay(100, TimeUnit.MILLISECONDS)
.doOnSubscribe(disposable -> savedDisposable = disposable);
t.subscribe(observer);
savedDisposable.dispose(); //this doesn't work
//observer.dispose(); //this works
assertTrue(observer.isDisposed());
}
我很高兴能与新RxJava Sources如:Single,Maybe,Completable,,使您接口的类的"源"的创建过程更清洁,防止了很多错误(如忘记调用onComplete())
但它需要大量的样板才能将它们组合成一个复杂的流.
例如,我们有加载和缓存数据的常见Android情况.假设我们有2个来源api,并cache和我们想将其结合:
public interface Api {
Single<Integer> loadFromNetwork();
}
public interface Cache {
Maybe<Integer> loadFromCache(); //maybe because cache might not have item.
}
让我们试着把它结合起来:
final Single<Integer> result = cache.loadFromCache()
.switchIfEmpty(api.loadFromNetwork());
它不会编译,因为Maybe没有重载Maybe.switchIfEmpty(Single):Single
所以我们必须转换一切:
final Single<Integer> result = cache.loadFromCache()
.switchIfEmpty(api.loadFromNetwork().toMaybe())
.toSingle();
组合它的另一种可能方法还需要сonversion:
final Single<Integer> result = Observable.concat(
cache.loadFromCache().toObservable(),
api.loadFromNetwork().toObservable()
).firstOrError();
因此,我没有看到任何使用新源的方法,没有很多转换会增加代码噪声并创建大量额外的对象.
由于这样的问题,我不能使用Single,Maybe,Completable和继续使用Observable无处不在.
所以我的问题是:
什么是相结合的最佳实践Single, …
我希望能够对Observable具有延迟发射的单元进行单元测试,但实际上并没有等待延迟时间.有没有办法做到这一点？
我目前正在使用CountDownHatch来延迟断言,并且工作正常,但会增加测试运行时间.
例:
val myObservable: PublishSubject<Boolean> = PublishSubject.create<Boolean>()
fun myObservable(): Observable<Boolean> = myObservable.delay(3, TimeUnit.SECONDS)
@Test
fun testMyObservable() {
val testObservable = myObservable().test()
myObservable.onNext(true)
// val lock = CountDownLatch(1)
// lock.await(3100, TimeUnit.MILLISECONDS)
testObservable.assertValue(true)
}
我想构建一个Repository返回a 的类Single<Something>.
该类应首先查看Cache返回的内容Maybe<Something>,如果Maybe完成后Service返回到我返回的内容Single<Something>
interface Cache {
fun getSomething(): Maybe<Something>
}
interface Service {
fun getSomething(): Single<Something>
}
class Repository (
private val cache: Cache,
private val service: Service
) {
fun getSomething(): Single<Something> {
return cache.getSomething()
.????(feed.getSomething()) //onCompleteTransformToSingle() or similar
}
}
我已经搜索了JavaDoc,但似乎并不存在这种情况的变换器.
有一个很好的处理方式吗？
我试图根据它的属性过滤列表.例如,Sensors类有一个属性isActive,我想用isActiveas 获取所有对象,true但我无法做到.我尝试了不同的方法,但我没有找到解决方案.有人可以帮我做吗？
这是我的代码:
mCompositeDisposable.add(
fcService.getStationList()
.subscribeOn(Schedulers.io())
.flatMap( stations -> {
return fcService.getSensorList(stations.get(0).getName().getOriginal());
})
.subscribe(this::handleSensors, this::handleError)
);
当我subscribe({})在Singleton类中使用Observable时,我是否需要.dispose()在某个时候调用方法？如果是,何时何地？因为在应用程序运行之前将保留单例.像这样的东西(Kotlin):
@Singleton
class MySingletonClass @Inject constructor(
private val api: MyAPIManager
) {
fun fetchData() {
//subscribed inside the Singleton
api.service.getSomeDataFromAPI()
.toRxObservable()
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe({
//do something internally with response
},
{
//handle error internally
})
}
该subscribe()方法返回一个Disposable.
我的主要问题是:我需要打电话dispose()吗？因为我认为我只能在应用程序完成或被杀时调用它,这是没有必要的.
医生说
从概念上讲,它是Single和Completable的联合,提供捕获排放模式的方法,其中可能存在0或1项或由某些反应源发出的错误信号.
但我不确定它的真正含义.它似乎是java8的Optional.
以下两个代码具有相同的结果,但我不知道Maybe可以做什么和Optional不能(或繁琐)做什么.
@Test
public void testMaybe1() {
Observable.just(3, 2, 1, 0, -1)
.map(i -> {
try {
int result = 6 / i;
return Maybe.just(result);
} catch (Exception e) {
return Maybe.empty();
}
})
.blockingForEach(maybe -> {
logger.info("result = {}", maybe.blockingGet());
}
);
}
@Test
public void testMaybe2() {
Observable.just(3, 2, 1, 0, -1)
.map(i -> {
try {
int result = 6 / i;
return Optional.of(result);
} catch (Exception e) { …Maybe switchIfEmpty如果源Maybe为空,我使用方法提供替代结果.但是,我希望仅在源为空时执行备用源,而在源不为空时不执行备用源.
在下面的例子中,我想避免执行costlyFallback如果源返回非空Maybe.当前实现总是调用它,因为它需要传递给switchIfEmpty方法.Maybe.fromCallable看起来很有希望,但它只适用于不包括返回a的callables Maybe.empty.任何提示都表示赞赏.如果switchIfEmpty愿意接受一些懒惰的评估Maybe提供者会很好.
public class StartRxMaybe {
public static void main(String... args) {
System.out.println(new StartRxMaybe().start().blockingGet());
}
private Maybe<Integer> start() {
return func()
.switchIfEmpty(costlyFallback());
}
private Maybe<Integer> func() {
System.out.println("Non-empty maybe returned");
return Maybe.just(1);
}
private Maybe<Integer> costlyFallback() {
System.out.println("Fallback executed anyway");
return LocalDate.now().getMonth() == Month.JULY
? Maybe.just(2)
: Maybe.empty();
}
}
rx-java2 ×10
rx-java ×8
java ×6
android ×5
kotlin ×2
dagger-2 ×1
optional ×1
retrofit2 ×1
rx-android ×1
unit-testing ×1