标签: rx-java

我的问题应该听起来像傻瓜,但我只是从Asynktask跳到RxJava.所以:

可以使用RxJava Observable和Volley Requests吗？这意味着,使用未来的请求.

我问这个问题,因为像改造这样的另一个httpClient 使用RxJava非常好,但是个人喜欢Volley,所以它可能吗？

编辑

基于第一个答案,我知道这是可能的.

你能分享一些展示如何做到这一点的样本吗？

问题

我有两个Apis.Api 1给了我一个项目列表,Api 2给了我更多关于我从Api获得的每个项目的详细信息.到目前为止,我解决它的方式导致性能不佳.

问题

在Retrofit和RxJava的帮助下,快速,快速地解决了这个问题.

我的方法

在片刻我的解决方案看起来像这样:

第1步:Single<ArrayList<Information>>从Api 1 执行Retrofit .

第2步:我遍历这些项目并向Api 2发出请求.

第3步:改造退货按顺序执行Single<ExtendedInformation>每个项目

步骤4:在完成Api 2的所有调用完成后,我为组合信息和扩展信息的所有项创建一个新对象.

我的守则

 public void addExtendedInformations(final Information[] informations) {
        final ArrayList<InformationDetail> informationDetailArrayList = new ArrayList<>();
        final JSONRequestRatingHelper.RatingRequestListener ratingRequestListener = new JSONRequestRatingHelper.RatingRequestListener() {
            @Override
            public void onDownloadFinished(Information baseInformation, ExtendedInformation extendedInformation) {
                informationDetailArrayList.add(new InformationDetail(baseInformation, extendedInformation));
                if (informationDetailArrayList.size() >= informations.length){
                    listener.onAllExtendedInformationLoadedAndCombined(informationDetailArrayList);
                }
            }
        };

        for (Information information : informations) {
            getExtendedInformation(ratingRequestListener, information);
        }
    }

    public void getRatingsByTitle(final JSONRequestRatingHelper.RatingRequestListener ratingRequestListener, final Information information) {
        Single<ExtendedInformation> …

我正试图绕过黄金法则(如果有的话):

何时使用BehaviorSubject？

和

何时使用PublishSubject？

他们之间的区别非常明显

有很多种科目.对于这个特定的要求,PublishSubject运行良好,因为我们希望从它停止的位置继续序列.所以假设事件1,2,3在(B)中发出,在(A)连接之后我们只想看到4,5,6.如果我们使用ReplaySubject,我们会看到[1,2,3],4, 5,6; 或者如果我们使用了BehaviorSubject,我们会看到3,4,5,6等等(来源:如何考虑RxJava中的主题(第1部分))

我已经看到它Subject用于两个上下文(至少),UI上下文和监听器上下文.

• UI上下文(以MVVM为例)

例如这里一个BehaviorSubject被使用,并且他们为什么使用它显然Subject并没有Observable,但我已经改变了BehaviorSubject到PublishSubject,但应用程序的行为仍然是相同的.

• 监听器上下文

他们为什么要创建项目领域BehaviorSubject而不是PublishSubject？

在使用 Android Room 时我应该考虑哪些例外情况。从我的研究中，我发现只有一种例外情况可能发生。

房间例外

这也是当你有Single<T>一个返回类型并且你有一个空的回报时。除此之外，我找不到任何其他可能引发异常的情况。

当然，如果你有一些逻辑上不正确的实现，可能会有一些例外，比如

• 编辑方案，但未实施 Migration
• OnConflictStrategy插入时不执行
• 在主线程上运行房间，但不允许它 allowMainThreadQueries()

我做了一些研究并尝试了几乎所有可能的情况，主要是 RxJava 返回类型，我看到了上面提到的一个异常，就是这样。

这是我运行的测试

我想确保我对每个可能的场景都有实现，并且没有一些异常和意外崩溃。我在想SQLite可能会发生异常的情况，但我相信它已经包裹在 Room 中并且会处理。（没有把握）

你能给出可能发生的任何其他可能的例外吗？

使用Obtrables with Retrofit时如何处理网络故障？

鉴于此代码:

Observable<GetJobResponse> observable = api.getApiService().getMyData();
observable
    .doOnNext(new Action1<GetJobResponse>() {
        @Override
        public void call(GetJobResponse getJobResponse) {
            //do stuff with my data
        }
    })
    .doOnError(new Action1<Throwable>() {
        @Override
        public void call(Throwable throwable) {
            //do stuff with error message
        }
    });

如果没有网络,请求就会失败,并且不会调用onError.它不会崩溃,但会无声地失败.日志显示Retrofit获取错误:

 java.net.UnknownHostException: Unable to resolve host "api-staging.sittercity.com": No address associated with hostname
     at java.net.InetAddress.lookupHostByName(InetAddress.java:424)
     at java.net.InetAddress.getAllByNameImpl(InetAddress.java:236)
     at java.net.InetAddress.getAllByName(InetAddress.java:214)
     at com.squareup.okhttp.internal.Dns$1.getAllByName(Dns.java:29)

使用回调,这只是传递给onFailure(RetrofitError错误).如何使用RxJava浮出水面？

我正在尝试转换我目前使用Retrofit的应用程序,以使用RX Java.为了处理分页,我传统上是从响应头中获取nextPage URL.

@Override
    public void success(Assignment assignment, Response response) {
        response.getHeaders(); // Do stuff with header info
}

但是,由于切换到RX Java,我不知道如何从我的改装调用中获取响应信息.

   @GET("/{item_id}/users")
    Observable<List<Objects>> getObjects(@Path("object_id") long object_id);

    @GET("/{next}")
    Observable<List<Objects>> getNextPageObjects(@Path("next") String nextURL);

有没有办法让我的改装调用返回我的标题信息和我的类型对象？

在这个博客中,他给出了这个(复制/粘贴以下代码)回调地狱的例子.但是,没有提到如何使用Reactive Extensions消除该问题.

所以这里F3取决于F1完成,F4和F5取决于F2完成.

1. 想知道Rx中的功能等价物是什么.
2. 如何在Rx中表示F1,F2,F3,F4和F5都应该异步拉出？

注意:我目前正试图绕过Rx,所以在问这个问题之前我没有尝试解决这个例子.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

public class CallbackB {

    /**
     * Demonstration of nested callbacks which then need to composes their responses together.
     * <p>
     * Various different approaches for composition can be done but eventually they end up relying upon
     * synchronization techniques such as the CountDownLatch used here or converge on callback design
     * changes similar to <a href="https://github.com/Netflix/RxJava">Rx</a>.
     */ …

我做了这些单元测试,结果不是我预期的结果:

// This one outputs "subscribe.onError" 
@Test
public void observable_doOnError_subscribingToError() throws InterruptedException {
    Observable<String> obs = getErrorProducingObservable();
    obs.doOnError(throwable -> System.out.println("doOnError"));
    obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(
        s -> {},
        error -> System.out.println("subscribe.onError")
    );
    Thread.sleep(300);
}

// This one outputs "subscribe.onError" 
@Test
public void observable_onErrorReturn() throws InterruptedException {
    Observable<String> obs = getErrorProducingObservable();
    obs.onErrorReturn(throwable -> "Yeah I got this");
    obs.subscribeOn(Schedulers.immediate()).observeOn(Schedulers.immediate()).subscribe(
        s -> System.out.println("got: " + s),
        error -> System.out.println("subscribe.onError")
    );
    Thread.sleep(300);
}

private Observable<String> getErrorProducingObservable()  {
    return Observable.create(subscriber -> {
        subscriber.onError(new RuntimeException("Somebody set up us the bomb")); …

我需要提出两个服务请求并将其结合起来:

ServiceA()=> [{"id":1,"name":"title"},{"id":1,"name":"title"}]

ServiceB(id)=> {"field":"value","field1":"value"}

目前,我已设法将结果组合在一起,但我需要将id参数作为参数传递给ServiceB并获取对第一个结果的访问权限.

到目前为止我尝试了什么:

   Retrofit repo = new Retrofit.Builder()
                .baseUrl("https://api.themoviedb.org/3/genre/")
                .addConverterFactory(GsonConverterFactory.create())
                .addCallAdapterFactory(RxJavaCallAdapterFactory.create())
                .build();

        Observable<GenerosResponse> Genres  = repo
                .create(services.class)
                .getAllGeneros("movie","list","da0d692f7f62a1dc687580f79dc1e6a0")
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread());

        Observable<ResponseMovies> Movies = repo
                .create(services.class)
                .getAllMovies("28","movies","da0d692f7f62a1dc687580f79dc1e6a0",12)
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread());

        Observable<CollectionsMovies> combined = Observable.zip(Genres, Movies, new Func2<GenerosResponse, ResponseMovies, CollectionsMovies>() {
            @Override
            public CollectionsMovies call(GenerosResponse generosResponse, ResponseMovies responseMovies) {
                return new CollectionsMovies(generosResponse, responseMovies);
            }
        });

        combined.
                subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(...);

编辑

解决方案根据@Maxim Ostrovidov的回答:

 private Observable<GenerosResponse> makeRequestToServiceA() {
        return  service.getAllGeneros("movie","list","da0d692f7f62a1dc687580f79dc1e6a0"); //some network call
    } …

我从Android的角度来问这个问题,但这应该适用于RxJava.

作为最佳实践,应我的观点始终处分甚至短暂的Completable,Single,Maybe和终止Observable的Rx类型应该在短期内结束,但是当用户关闭该视图可以仍在执行？我知道当Rx链终止时,它会被丢弃,但这可能会在视图关闭后的某个时候发生.

例如,Single正在执行HTTP GET.调用将完成,但可能是在视图被破坏后,暂时阻止垃圾回收.

如果a CompositeDisposable用于Disposable长期观察收集这些s,我会认为应该注意clear()或Disposable定期删除这些s以防止无限大小的增长CompositeDisposable？