标签: rx-android

我有以下方法使用otto和发布对UI的响应AsyncTask.

private static void onGetLatestStoryCollectionSuccess(final StoryCollection storyCollection, final Bus bus) {
    new AsyncTask<Void, Void, Void>() {
        @Override
        protected Void doInBackground(Void... params) {
            bus.post(new LatestStoryCollectionResponse(storyCollection));
            return null;
        }
    }.execute();
}

我需要帮助将其转换AsyncTask为RxJava使用RxAndroid库.

我在RxAndroid中使用RxJava和Android应用程序.我正在使用mergeDelayError将两个逆向拟合网络调用组合成一个observable,如果发出一个,它将处理发出的项目,如果有的话,则处理错误.这不起作用,它只会在遇到错误时触发onError操作.现在为了测试这个我转移到一个非常简单的例子,当我有一个onError调用时,仍然不会调用successAction.见下面的例子.

Observable.mergeDelayError(
                Observable.error(new RuntimeException()),
                Observable.just("Hello")
            )
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(Schedulers.io())
            .finallyDo(completeAction)
            .subscribe(successAction, errorAction);

只有在使用两个成功的可观察对象时才会调用成功操作.我错过了mergeDelayError应该如何工作的东西？

编辑:

我发现,如果我删除了observeOn,subscribeOn一切都按预期工作.我需要指定线程和思想,这是使用Rx的重点.知道为什么指定那些Schedulers会破坏行为吗？

我想做这个:

Observable.just(bitmap)
            .map(new Func1<Bitmap, File>() {
                @Override
                public File call(Bitmap photoBitmap) {

                    //File creation throws IOException, 
                    //I just want it to hit the onError() inside subscribe()

                    File photoFile = new File(App.getAppContext().getCacheDir(), "userprofilepic_temp.jpg");
                    if(photoFile.isFile()) {//delete the file first if it exists otherwise the new file won't be created
                        photoFile.delete();
                    }
                    photoFile.createNewFile(); //saves the file in the cache dir

                    FileOutputStream fos = new FileOutputStream(photoFile);
                    photoBitmap.compress(Bitmap.CompressFormat.JPEG, 90, fos);//jpeg format
                    fos.close();

                    return photoFile;

                }
            })
            .subscribe(//continue implementation...);

基本上在call()方法中,它可以抛出异常.如何让Observer处理它onError().或者这不是思考这个问题的正确方法吗？

我是RxJava的新手所以请原谅我,如果这听起来太新手了:-).

截至目前,我有一个抽象的CallbackClass,它实现了Retofit Callback.在那里我捕获了Callback的"onResponse"和"onError"方法,并在最终转发到自定义实现的方法之前处理各种错误类型.我还使用此集中式类来进行请求/响应应用程序日志记录和其他内容.

例如:对于来自我的服务器的特定错误代码,我在响应正文中收到一个新的Auth令牌,刷新令牌,然后clone.enqueue调用.当然,我的服务器的响应还有其他一些全局行为.

当前解决方案(无Rx):

    public abstract void onResponse(Call<T> call, Response<T> response, boolean isSuccess);

    public abstract void onFailure(Call<T> call, Response<T> response, Throwable t, boolean isTimeout);

    @Override
    public void onResponse(Call<T> call, Response<T> response) {
        if (_isCanceled) return;

        if (response != null && !response.isSuccessful()) {
            if (response.code() == "SomeCode" && retryCount < RETRY_LIMIT) {
                TokenResponseModel newToken = null;
                try {
                    newToken = new Gson().fromJson(new String(response.errorBody().bytes(), "UTF-8"), TokenResponseModel.class);
                } catch (Exception e) {
                    e.printStackTrace();
                }

                    SomeClass.token = newToken.token;
                    retryCount++;
                    call.clone().enqueue(this);
                    return;
                }
            } …

据我所知,AndroidObservable有助于确保:

1. 订阅者始终在主线程上观察
2. 当分离/活动分离/停止时,观察立即停止,并且不更新与框架相关的组件(如ui textviews等).

但是,为了确保上下文被释放(防止泄漏),我看到的大多数示例通常都说你必须做.unsubscribe onDestroyView/onDestroy,这实质上会停止订阅,并阻止订阅者接收这些更新.

所以我的问题是:

如果我通过.observeOn(AndroidSchedulers.mainThread()手动指示订阅应该在主线程上发生,那么使用AndroidObservables还有其他优势吗？

以下两种方法有什么不同吗？

_subscription1 = AndroidObservable.bindFragment(MyFragment.this, myCustomAwesomeObservable()) //
                           .subscribeOn(Schedulers.io()) //
                           .subscribe(...);


_subscription2 =  myCustomAwesomeObservable()
                           .subscribeOn(Schedulers.io()) //
                           .observeOn(AndroidSchedulers.mainThread()) //
                           .subscribe(...);


@Override
public void onDestroyView() {
    _subscription1.unsubscribe();
    _subscription2.unsubscribe();
    super.onDestroyView();
}

我对RxJava很新,所以这可能是一个愚蠢的问题.我将描述我的情景.

我在UI线程上运行了一些代码,它们将更新一些图像,但这些图像不是很重要,它们在生成它们时消耗了一些资源,所以我想在单个线程上生成它们(当然不是UI线程)和逐个生成它们.我猜蹦床调度程序是我想要的,但我的问题是,如果我使用它然后它在UI线程上工作,我希望它在另一个线程上做.

显然我可以编写自己的线程,我可以在其中对项目进行排队,然后逐个处理这些,但我想RxJava可能会有一个简单的解决方案吗？

我当前的代码如下所示:

Observable<Bitmap> getImage = Observable.create(new Observable.OnSubscribe<Bitmap>() {
    @Override public void call(Subscriber<? super Bitmap> subscriber) {
        Log.w(TAG,"On ui thread? "+ UIUtils.isRunningOnUIThread());
        subscriber.onNext(doComplexTaskToGetImage());
        subscriber.onCompleted();
    }
});

getImage.subscribeOn(Schedulers.trampoline()).subscribe(new Action1<Bitmap>() {
    @Override public void call(Bitmap bitmap) {
        codeToSetTheBitmap(bitmap);
    }
});

我的日志上写着"On ui thread？" 永远都是如此.那么我如何使代码和所有后续尝试做同样的事情按顺序在单个线程(而不是ui线程)上运行而不编写一堆代码来排队工作？

编辑:

我相信现在可以使用Schedulers.single()或者如果你想要你自己可以使用它new SingleScheduler().我还在测试,但是当我发布这个帖子时,我认为它确实符合我的要求.

我刚刚开始使用RxJava/RxAndroid.我想避免上下文泄漏,所以我创建了一个像这样的BaseFragment:

public abstract class BaseFragment extends Fragment {

    protected CompositeSubscription compositeSubscription = new CompositeSubscription();

    @Override
    public void onDestroy() {
        super.onDestroy();

        compositeSubscription.unsubscribe();
    } 
} 

在我的片段中扩展了BaseFragment,我这样做:

protected void fetchNewerObjects(){
        if(!areNewerObjectsFetching()){ //if it is not already fetching newer objects

            Runtime.getRuntime().gc();//clean out memory if possible

            fetchNewObjectsSubscription = Observable
                .just(new Object1())
                .map(new Func1<Object1, Object2>() {
                    @Override
                    public Object2 call(Object1 obj1) {
                        //do bg stuff
                        return obj2;
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<Object2>() {
                    @Override
                    public void onCompleted() {
                        compositeSubscription.remove(fetchNewObjectsSubscription);
                        fetchNewObjectsSubscription = null;
                    }

                    @Override
                    public …

我正在开发具有后台数据同步功能的Android应用程序.我目前正在使用RxJava定期在服务器上发布一些数据.除此之外,我想为用户提供一个"强制同步"按钮,它会立即触发同步.我知道如何使用Observable.interval()以定期时间间隔推送数据,我知道如何使用它Observalbe.just()来推送那个被强制的数据,但是我想将它们排队,如果发生这种情况,那么在前一个仍然运行的情况下触发它.

所以让我们举个例子,当1min是自动同步的间隔时,让我们说同步持续40秒(我夸张这里只是为了更容易点).现在,如果有任何机会,用户在自动仍在运行时按下"强制"按钮(反之亦然 - 强制一个仍在运行时自动触发),我想将第二个同步请求排队第一个完成.

我画了这张图片,可能会有更多的视角:

如您所见,自动触发(由某些人Observable.interval()),在同步过程中,用户按下"强制"按钮.现在我们要等待第一个请求完成,然后再次启动强制请求.有一次,当强制请求正在运行时,再次触发了新的自动请求,只是将其添加到队列中.从队列中完成最后一个之后,一切都停止了,然后稍后再次安排自动计划.

希望有人能指出我纠正操作员如何做到这一点.我已尝试过Observable.combineLatest(),但是队列列表在开始时被调度,当我向队列添加新的同步时,它在前一个操作完成时没有继续.

Darko,非常感谢任何帮助

背景

public Observable<List<Foo>> search(SearchView searchView) {

    return RxSearchView.queryTextChanges(searchView)
            .filter(charSequence -> !TextUtils.isEmpty(charSequence))
            .throttleLast(100, TimeUnit.MILLISECONDS)
            .debounce(200, TimeUnit.MILLISECONDS)
            .observeOn(AndroidSchedulers.mainThread())
            .subscribeOn(AndroidSchedulers.mainThread())
            .flatMap(this::performSearch) //Search the DB
            .onErrorResumeNext(this::doSomething);
}

我试图用AndroidJUnit4跑步者测试上面的方法Mocktio.

@Test
public void testSearchCallsDataManager_WhenCalled() {

    String input = "abc";

    when(mockSearchView.getQuery()).thenReturn(input);

    searchRequestManager.search(mockSearchView).subscribe(testSubscriber); //Using standard TestSubscriber

    testSubscriber.assertNoErrors();
    testSubscriber.assertNotCompleted();
    verify(mockDataManager).getFoos(input);
}

问题

我尝试过使用a mockSearchView和real SearchView.

mockSearchView = mock(SearchView.class);
searchView = new SearchView(InstrumentationRegistry.getContext(), null);
searchView = new SearchView(InstrumentationRegistry.getTargetContext(), null);

在实例化期间,实际对象在测试运行时会导致不同的异常.模拟对象在执行期间似乎没有任何效果.

更新

为清楚起见:理想情况下,如果我可以模拟SearchView,那将是很好的,因为我想测试发生什么后发生的事情以及使用正确的输入调用performSearch方法.

我有一个简单的用例,其中:

• Activity1创建一个fragment1

• 创建后的fragment1通知活动它是否已创建并更新其activity1视图.

• 获取通知更新fragment1视图后的activity1.

我正在使用rxandroid,sublibrary rxlifecycle组件和android,但我还在学习阶段,rx-lifecyclestackoverflow上甚至没有标记,所以我仍然在努力理解这个库的流程..

编辑

我不喜欢使用EventBus,就像每个人都大喊大叫做某事一样,所以Rxjava Observable方法会很有用