标签: rx-java2

我是RXJava/RXAndroid的新手.我想实现这个案例:根据RXJava中的某些条件选择不同的方式.例如,首先,我从网络获取用户信息,如果这是VIP用户,我将继续从网络获取更多信息或只在主线程中显示一些信息(打破链.)这里的流程图:https: //i.stack.imgur.com/0hztR.png

我做了一些搜索,只发现"switchIfEmpty"可能有所帮助.我写下面的代码:

getUserFromNetwork("userId")
                .flatMap(new Function<User, ObservableSource<User>>() {
                    @Override
                    public ObservableSource<User> apply(User user) throws Exception {
                        if(!user.isVip){
                            //show user info on MainThread!
                            return Observable.empty();
                        }else{
                            return getVipUserFromNetwork("userId");
                        }
                    }
                }).switchIfEmpty(new ObservableSource<User>() {
                    @Override
                    public void subscribe(Observer<? super User> observer) {
                        //show user info in main thread
                        //just break the chain for normal user
                        observer.onComplete();
                    }
                }).doOnNext(new Consumer<User>() {
                    @Override
                    public void accept(User user) throws Exception {
                        //show vip user info in main thread
                    }
                }).subscribe();

有更简单的方法来实现这一目标吗？

谢谢!

我当前的Android应用程序允许用户远程搜索内容.

例如,向用户显示一个EditText接受其搜索字符串并触发远程API调用的用户,该调用返回与输入的文本匹配的结果.

更糟糕的情况是,我只需添加一个TextWatcher并在每次调用时触发API调用onTextChanged.这可以通过在进行第一次API调用之前强制用户输入至少N个字符来进行搜索来改进.

"完美"解决方案具有以下特点: -

一旦用户开始输入搜索字符串

定期(每M毫秒)消耗输入的整个字符串.每当周期到期并且当前用户输入与先前的用户输入不同时触发API调用.

[是否有可能与输入的文本长度相关的动态超时？例如,当文本"短"时,API响应大小将很大并且需要更长时间才能返回和解析; 随着搜索文本变得越来越长,API响应大小将随着"飞行"和解析时间而减少

当用户重新键入EditText字段时,重新启动Periodic消耗文本.

每当用户按下ENTER键触发"最终"API调用,并停止监视用户输入到EditText字段.

设置用户在触发API调用之前必须输入的文本的最小长度,但将此最小长度与覆盖的超时值组合,以便在用户希望搜索"短"文本字符串时可以.

我确信RxJava和/或RxBindings可以支持上述要求,但到目前为止我还没有实现一个可行的解决方案.

我的尝试包括

private PublishSubject<String> publishSubject;

  publishSubject = PublishSubject.create();
        publishSubject.filter(text -> text.length() > 2)
                .debounce(300, TimeUnit.MILLISECONDS)
                .toFlowable(BackpressureStrategy.LATEST)
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(final String s) throws Exception {
                        Log.d(TAG, "accept() called with: s = [" + s + "]");
                    }
                });


       mEditText.addTextChangedListener(new TextWatcher() {
            @Override
            public void beforeTextChanged(final CharSequence s, final int start, final int count, final int after) {

            }

            @Override …

我刚看了杰克沃顿 与RxJava管理国的会议.

他建议以这种方式将事件从视图转换为动作:

Observable<Event> events = RxView.clicks(view).map(__ -> new Event());
ObservableTransformer<Event, Action> action = events -> events.flatMap(/* ... */);
events.compose(action).subscribe();

我想知道这个实现的不同之处:

Observable<Event> events = RxView.clicks(view).map(__ -> new Event());    
Observable<Action> action = events.flatMap(/* ... */);
action.subscribe();

使用compose()带有 ObservableTransformer的 a和flatMap()带有两个Observable 的简单有什么区别？

我想创建一个Observable,它从Observable中发出一些包含Interval Observable对象列表的项目,这样第一个observable中的项目就会有一些时间延迟.这是我的实现:

 val just1 = ArrayList<SomeClass1>()

填充一些项目

fun populateJust1() {
just1.add(SomeClass1("23", 23))
just1.add(SomeClass1("24", 24))
just1.add(SomeClass1("25", 25))
}

populateJust1()

和区间Observable压缩

Observable.fromIterable(just1)
            .zipWith(Observable.interval(2, TimeUnit.SECONDS)) { item: SomeClass1, interval: Long -> item }
            .subscribe(Consumer<SomeClass1> { someClass1 -> Log.v("someClass1", someClass1.toString()) })

但是,IDE,Android Studio 3.0以红色方式强调了zipWith运算符,并说:

使用提供的参数不能调用以下任何函数.zipWith(((observer:Observer)→Unit)!,((t1:SomeClass1,t2:Long)→R)!)其中R不能推断; U =乐趣的长拉链(其他:((观察者:观察者)→单位)!,拉链:((t1:SomeClass1,t2:U)→R)!):可观察!在io.reactivex.Observable zipWith(ObservableSource!,BiFunction!)中定义,其中无法推断出R; U =长!为了好玩的zipWith(其他:ObservableSource!,拉链:BiFunction!):可观察!在io.reactivex.Observable zipWith中定义((Mutable)Iterable!,BiFunction!)其中U,R无法推断出有趣的zipWith(其他:( Mutable)Iterable!,zipper:BiFunction!):Observable!在io.reactivex.Observable zipWith中定义((Mutable)Iterable!,((t1:SomeClass1,t2:Long)→R)!)其中无法推断出R; U =乐趣的长拉链(其他:(可变)Iterable!,拉链:((t1:SomeClass1,t2:U)→R)!):可观察!在io.reactivex.Observable中定义

怎么了？我刚从Java类转换了这个代码,其中一切都是正确的

假设您要在Observable链中插入Completable,例如对于每个发出的元素,有一个可运行的块并在其完成之前阻塞,您会选择哪个选项？(这里Completable.complete()只是举个例子)

1. .flatMap { Completable.complete().andThen(Observable.just(it)) }

2. .doOnNext { Completable.complete().blockingAwait() }

3. 别的什么？

我很难找到一个合适的解决方案,使view/viewmodel知道网络状态,特别是使用RxJava.

我试图关注谷歌的NetworkBoundResource和Iammert的networkBoundResouce(也试过GithubBrowserSample, 但我真的不明白代码或它应该如何封装.

我正在寻找如何使用RxJava2实现NetworkBoundResource,使数据自动刷新数据库.

我将展示我目前拥有的东西:

与arch.lifecycle.ViewModel的片段

public class LoginViewModel extends ViewModel {

    static final int SCREEN_JUST_BUTTONS = 0;
    static final int SCREEN_LOGIN_INPUT = 1;
    static final int SCREEN_PASS_INPUT = 2;

    // using a PublishSubject because we are not interested in the last object that was emitted
    // before subscribing. Like this we avoid displaying the snackbar multiple times
    @NonNull
    private final PublishSubject<Integer> snackbarText;

    private int SCREEN = 0;

    private LoginUiModel loginUiModel;
    @NonNull
    private BaseSchedulerProvider schedulerProvider; …

我正在试试Room和RxJava.我对两者都很新.我的代码工作正常,但我有兴趣知道最佳实践.

想象一下有10个字段的对象说CompleteSong.在我的应用程序中,Flowable<List<CompleteSong>>每当我在数据库中进行更新时,我都会回复房间Flowable,所有值都会更新.我正在显示该列表RecyclerView.我有两种可能性继续下去,即

1.我可以存储正在更改的对象并等待Flowable更新.当可流动更新我更新1项时RecyclerView

2.我可以更新RecyclerView时,Flowable更新解析列表并寻找的改变实例CompleteSong

我曾经使用前一种方法,似乎反应性较低但优化,而后者似乎更具反应性,但同时不太优化.

有人可以告诉我这个,应该做些什么？我应该采取什么方法,或者还有其他一些我无法发现的方法.

注意:我是新手,所以如果我搞砸了,请考虑更新我的术语.

我有以下rxJava链:

 override fun combineLocationToPlace(req: Flowable<Place>): Flowable<Place> {
        var combinedFlowable = Flowable
                .combineLatest(
                        req,
                        getLastLocation().lastOrError().toFlowable(),
                        BiFunction<Place, Location, Place> { t1, location ->
                            Timber.w("FIRSTINIT - Retrieved location $location")
                            var placeLocation = Location(t1.placeName)
                            placeLocation.latitude = t1.latitude
                            placeLocation.longitude = t1.longitude
                            t1.distance = location.distanceTo(placeLocation)
                            t1
                        })


        return combinedFlowable
                .onErrorResumeNext { t: Throwable ->
                    Timber.w(t, "FIRSTINIT - Could not retrieve location for place (${t.message}) returning original request")
                    req
                }
                .doOnError {
                    Timber.w("FIRSTINIT - did detect the error here...")
                }

        return combinedFlowable
    }

简而言之,我正在从本地数据库(一个地方)检索一些数据,我想将它与GPS中的最新位置结合起来:

 override fun getLastLocation(requestIfEmpty: …

我很高兴能与新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, …

我在我的应用程序中使用来自Android Architecture Components的ViewModel.在ViewModel中,我正在使用RxJava2订阅,并Disposable在订阅后保留对象.之前,我这样做的时候Activity,我是用来处置Disposable的onDestroy()-以避免内存泄漏:

@Override
protected void onDestroy() {
    disposable.dispose();
    super.onDestroy();
}

何时以及如何处理它ViewModel？我真的需要这样做吗？