标签: rx-android

所以我有一个A依赖于第二个的异步操作B.怎样使B A A的延续,你不应该用那么功能的Javascript承诺,或与ContinueWith在.NET TPL？

让我们从Android获取一个scenarion,我有两个服务端点:

• 一个用于获取图像的URL
• 一个用于获取图像本身

码:

private void loadImage(final ImageView imageView) {
    Observable<String> stringObs = getImageUrlAsync();
    stringObs.subscribe(new Action1<String> () {
        @Override
        public void call(String imageUrl) {
            // First async operation done
            Observable<Bitmap> bitmapObs = getBitmapAsync(imageUrl);
            bitmapObs.subscribe(new Action1<Bitmap>() {
                @Override
                public void call(Bitmap image) {
                    // second async operation done
                    imageView.setImageBitmap(image);
                }
            });
        }
    });
}

我不希望这两个操作嵌套在代码方面,但是它的格式更像是它的行为 - 顺序.

我一直无法找到关于RxJava/RxAndroid中的Observables的延续.有人可以帮忙吗？

我刚刚开始钻研RxJava/RxAndroid和Retrofit.我喜欢RxAndroid提供的强大功能,但我被困在一个特定的地方,似乎无法找到解决方案.我基本上有一个API,它接受用户并从API获取有关该用户的信息,而这非常有用,我想做的不是返回每个用户的信息,而是单独返回userinfo的arraylist.

 Observable.from(interestingUsers)  //3 users declared statically for test
              .flatMap(user -> gitHubService.getUserData(user))
              .observeOn(AndroidSchedulers.mainThread())
              .subscribe(gitHubUser -> Log.e("Log", "User data returned : " + gitHubUser.getName()));

上面的代码只是将各个用户记录三次,我想要的是作为单个结果返回的3个用户的列表.

我知道这可能是微不足道的,我无法弄明白.任何帮助,将不胜感激.

我对 rxJava 相当陌生。在我的 api 响应中，我获得了有关总页数和当前页码的信息，例如：

 "pages": 22,
 "page": 1,

我正在使用 Retrofit 在数据层进行 api 调用，我的 api 服务如下：

@GET("/story")
    Observable <StoryCollectionEntity> storyCollection(
              @Query("feed_ids") String feed_items, 
              @Query("page") int  page);

然后：

   public Observable<StoryCollectionEntity> storyCollection() {
        return mUserApi.storyCollection(items,page);
}

我在域层进行了这样的订阅：

public void execute(Subscriber UseCaseSubscriber) {
        this.subscription = this.buildUseCaseObservable()
                 .subscribeOn(Schedulers.from(threadExecutor))
                .observeOn(postExecutionThread.getScheduler())
                .subscribe(UseCaseSubscriber);

    }
 @Override public Observable buildUseCaseObservable() {
    return this.userRepository.stories();
  }

我正在弄清楚如何通过发出下一页的结果让这个观察者对 recyclerView 滚动事件做出反应。即第 2 页到第二个滚动事件和第 3 页在第 3 次滚动...等

我想根据链请求更新 UI 2 次。

请求getUserByID()完成后首先更新 UI 。第二次getParentName()完成但是当我尝试在第一个请求时更新 UI 并获得线程问题的错误原因时。

我不知道如何解决这个问题。

示例代码

API.getUserByID("USER ID")
        .flatMap(user-> {
            // How to update UI here?
            return API.getParentName(user.getId()))
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(parent-> {
            // It's ok to update UI cause of on main thread.
            parentTextView.setText(parent.getName())
        });

谢谢你。

我覆盖我的调度程序以进行测试:

RxJavaPlugins.getInstance().registerSchedulersHook(object : RxJavaSchedulersHook() {
    override fun getIOScheduler() = Schedulers.immediate()
    override fun getComputationScheduler() = Schedulers.immediate()
    override fun getNewThreadScheduler() = Schedulers.immediate()

  })

但文档说:

不推荐使用RxJavaHooks的静态方法.

如何使用覆盖这些调度程序RxJavaHooks？

我正在尝试为Android学习RXJAVA.部件有意义,我仍然对很多其他部分感到困惑,但是,考虑到一些时间,我希望它会更有意义.

目前我在使用'地图'功能时遇到了麻烦.我收到错误但无法弄清楚如何解决它.

在我分享我的代码之前,我将解释我的理解..在一个简单的层面上.. Observable - 发出数据的代码.Observer - 处理发出数据的代码.Map - 接收类型A的数据并将其返回处理或作为类型B的代码.

所以,考虑到这一点:

在gradle我有:

compile 'io.reactivex.rxjava2:rxjava:2.0.1' 
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

如果我有:

    //declaration at top of file
    private Observable<Integer> myIntObservable;
    private Observer<Integer> myIntObserver;
    private Observer<String> myStringObserver;
    private Observable<String> myStringObservable;

    //usage in a function
    myIntObserver = new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {

        }

        @Override
        public void onNext(Integer value) {
            Toast.makeText(getApplicationContext(), "" + value, Toast.LENGTH_SHORT).show();
        }

        @Override
        public void onError(Throwable e) {

        }

        @Override
        public void onComplete() {
            Toast.makeText(getApplicationContext(), "Int Observer Async Complete", Toast.LENGTH_SHORT).show(); …

我正在尝试根据Rxjava中的某些条件创建组列表。

以下是我的回复：

{  
   "dates":[  
      {  
         "date":18,
         "value":"1"
      },
      {  
         "date":18,
         "value":"2"
      },
      {  
         "date":18,
         "value":"3"
      },
      {  
         "date":19,
         "value":"1"
      },
      {  
         "date":19,
         "value":"2"
      },
      {  
         "date":19,
         "value":"3"
      },
      {  
         "date":19,
         "value":"4"
      }
   ]
}

如何使用Rxjava按值18 [值1，值2，值3，最高值= 3，最低值= 1]分组19 [值1，值2，值3，值4，最高值= 4，最低值= 1]

注意：我可以创建for循环，但是将从服务器获取响应，并且由于返回了使用rx java功能的可观察到的想法。

任何帮助将不胜感激。

谢谢Shanthi

我使用该Observable.fromCallable方法创建了一个Observable，并订阅了它，如下面的代码片段所示。

Observable<String> stringObservable = Observable.fromCallable(new Callable<String>() {
        @Override
        public String call() throws Exception {
            Thread.sleep(1000);
            return Thread.currentThread().getName();
        }
});

stringObservable.subscribeOn(Schedulers.io());
stringObservable.observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String aDouble) {
                Toast.makeText(SimpleActivity.this, "onNext: " + aDouble, 
                Toast.LENGTH_LONG).show();
            }

            @Override
            public void onError(Throwable e) {
                new AlertDialog.Builder(SimpleActivity.this)
                        .setTitle("Error")
                        .setMessage(e.toString())
                        .show();
            }

            @Override
            public void onComplete() {

            }
        });

上面的片段产生了一个祝酒词，表明Callable在主线程而不是Schedulers.io线程上运行。发生了什么？

我正在学习本教程,然后添加以下行:

.retryWhen(errors -> errors.flatMap(error -> Observable.timer(30, TimeUnit.SECONDS)))

对我来说Transformer,但它会触发编译错误:

错误:不兼容的类型:无法推断类型变量R(参数不匹配; lambda表达式中的错误返回类型Observable无法转换为Publisher)其中R,T是类型变量:R extends方法flatMap中声明的Object(函数> )T扩展在类Flowable中声明的Object

而下面的波浪线error -> Observable.timer(30, TimeUnit.SECONDS)说:

没有类型变量R的实例存在,以便Observable符合Publisher

我究竟做错了什么？这个例子中缺少什么？

我不是RxJava的菜鸟，但是有一天...; ）

是）我有的：

• 使用事件计划器模型接收json
• 在此模型中有任务
• 我想将此任务保存到本地数据库中，但也要在有关系的情况下将其添加到它们中

有经验的人可以看看下面的代码：

    private fun saveEventPlanners(eventPlanners: ArrayList<EventPlanner>) {
    LogMgr.d(TAG, "saveEventPlanners() events:$eventPlanners")

    compositeDisposable.add(wfmStorageDomain.saveEventPlanners(eventPlanners)
            .subscribeOn(rxSchedulers.computation())
            .observeOn(rxSchedulers.computation())
            .subscribe({ saved ->
                LogMgr.v(TAG, "event planners saved successfully, value: $saved")

                val taskList = ArrayList<Task>()

                eventPlanners.forEach { eventPlanner ->

                    if (eventPlanner.status == EventPlanner.EventPlannerStatus.NEW) {
                        FS.get().wfmComponent.getEventPlannerStatusChanger().updateEventPlannersStatus(eventPlanner.event_id!!, EventPlanner.EventPlannerStatus.RECEIVED)
                    }

                    (eventPlanner.tasks as ArrayList<Task>).forEach {
                        val task = Task()
                        task.id = it.id
                        task.status_id = Task.STATUS.NEW
                        task.name = it.name
                        task.end_scenario_id = it.end_scenario_id
                        task.start_scenario_id = it.start_scenario_id
                        task.isBind = it.isBind
                        task.parent_event_planner_id = eventPlanner.event_planner_id

                        taskList.add(task)
                    }
                    compositeDisposable.add(wfmStorageDomain.saveTasks(taskList)
                            .subscribeOn(rxSchedulers.computation())
                            .observeOn(rxSchedulers.computation())
                            .subscribe({
                                LogMgr.d(TAG, …