标签: rx-kotlin

我为什么要使用Kotlin协同程序？

似乎RxKotlin库更加多样化.Kotlin协同程序看起来效率显着降低,相比之下使用起来更加麻烦.

我根据安德烈·布雷斯拉夫(JetBrains)的设计讲话对协同程序提出了自己的看法:https://www.youtube.com/watch？v = 4W3ruTWUhpw

可以在此处访问来自谈话的幻灯片:https://www.slideshare.net/abreslav/jvmls-2016-coroutines-in-kotlin

编辑(感谢@hotkey):

关于当前协程状态的更好来源:https://github.com/Kotlin/KEEP/blob/master/proposals/coroutines.md

我将开始在我的Android项目中使用反应式编程.我使用Kotlin作为主要语言,现在我想申请Rx.我的第一选择是RxAndroid,但后来我注意到有RxKotlin.

据我所知,两者都是从RxJava分叉的,所以RxAndroid可能为android常见任务提供了一些API.另一方面,RxKotlin支持lambas开箱即用,让我避免将kotlin与java混合.

在这种情况下哪一个是首选的库？

我正在尝试测试以下RxKotlin/RxJava 2代码:

validate(data)
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .flatMap { ... }

我正在尝试覆盖调度程序,如下所示:

// Runs before each test suite
RxJavaPlugins.setInitIoSchedulerHandler { Schedulers.trampoline() }
RxAndroidPlugins.setInitMainThreadSchedulerHandler { Schedulers.trampoline() }

但是,运行测试时出现以下错误:

java.lang.ExceptionInInitializerError
...
Caused by: java.lang.NullPointerException: Scheduler Callable result can't be null
    at io.reactivex.internal.functions.ObjectHelper.requireNonNull(ObjectHelper.java:39)
    at io.reactivex.plugins.RxJavaPlugins.applyRequireNonNull(RxJavaPlugins.java:1317)
    at io.reactivex.plugins.RxJavaPlugins.initIoScheduler(RxJavaPlugins.java:306)
    at io.reactivex.schedulers.Schedulers.<clinit>(Schedulers.java:84)

有没有人遇到过这个问题？

使用RxKotlin/RxJava 1和以下调度程序覆盖时,测试工作正常:

RxAndroidPlugins.getInstance().registerSchedulersHook(object : RxAndroidSchedulersHook() {
    override fun getMainThreadScheduler() = Schedulers.immediate()
})

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

谢谢!

我像这样初始化我的变量: -

 val user: BehaviorSubject<User?> user = BehaviorSubject.create()

但我不能这样做.IDE抛出错误: -

user.onNext(null)

这样做,IDE说你永远不会为空: -

user.filter( u -> u!=null)

我对Kotlin lambda语法感到困惑.

起初,我有

.subscribe(
          { println(it) }
          , { println(it.message) }
          , { println("completed") }
      )

哪个工作正常.

然后我将onNext移动到另一个名为GroupRecyclerViewAdapter的类中,该类实现了Action1<ArrayList<Group>>.

.subscribe(
          view.adapter as GroupRecyclerViewAdapter
          , { println(it.message) }
          , { println("completed") }
      )

但是,我得到了错误:

Error:(42, 17) Type mismatch: inferred type is () -> ??? but rx.functions.Action1<kotlin.Throwable!>! was expected
Error:(42, 27) Unresolved reference: it
Error:(43, 17) Type mismatch: inferred type is () -> kotlin.Unit but rx.functions.Action0! was expected

我可以通过更改为:修复错误:

.subscribe(
          view.adapter as GroupRecyclerViewAdapter
          , Action1<kotlin.Throwable> { println(it.message) }
          , Action0 …

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

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

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

3. 别的什么？

大家!我有一些问题.我是RxJava/RxKotlin/RxAndroid的初学者,并且不了解一些功能.例如:

import rus.pifpaf.client.data.catalog.models.Category
import rus.pifpaf.client.data.main.MainRepository
import rus.pifpaf.client.data.main.models.FrontDataModel
import rus.pifpaf.client.data.product.models.Product
import rx.Observable
import rx.Single
import rx.lang.kotlin.observable
import java.util.*


class MainInteractor {

    private var repository: MainRepository = MainRepository()

    fun getFrontData() {

        val cats = getCategories()
        val day = getDayProduct()
        val top = getTopProducts()

        return Observable.zip(cats, day, top, MainInteractor::convert)
    }

    private fun getTopProducts(): Observable<List<Product>> {
        return repository.getTop()
                .toObservable()
                .onErrorReturn{throwable -> ArrayList() }

    }

    private fun getDayProduct(): Observable<Product> {
        return repository.getSingleProduct()
                .toObservable()
                .onErrorReturn{throwable -> Product()}

    }

    private fun getCategories(): Observable<List<Category>> {
        return repository.getCategories()
                .toObservable() …

我有,Observable<Rates>而Rate只是一个简单的对象:

Rate(val value:String){}
Rates(val rates: List<Rate>)

我想改变Observable<Rates>成Observable<HashMap<String,Long>.

所以例如对于Rates(arrayOf(Rate("1"),Rate("2"), Rate("3"),Rate("3"), Rate("2"),Rate("2")))我期望结果的费率:

(1 -> 1)
(2 -> 3)
(3 -> 2)
(4 -> 0)
(5 -> 0)

我开始创建类似的东西:

service.getRates()
        .flatMap {it-> Observable.from(it.rates) }
        .filter { !it.value.isNullOrEmpty() }
        .groupBy {it -> it.value}
        .collect({ HashMap<String,Long>()}, { b, t -> b.put(t.key, t.count???)}

但我被困在这里,我不知道所有的价值？如果没有5中的4,我不知道如何添加空值(0).有没有办法用rx做到这一点？

我对 rxJava 中 doOnSuccess 的用例感到困惑。
让我们看看代码：

情况1：

networkApi.callSomething()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())               
    .doOnSuccess(__ -> showLog(SUCCESS))
    .doOnError(__ -> showLog(ERROR))
    .subscribeBy(
             onSuccess = {//Do something}, 
             onError = {//Show log here}
          )

案例2：

networkApi.callSomething()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())               
    .subscribeBy(
             onSuccess = {
               //Do something
               showLog(SUCCESS)
             }, 
             onError = {showLog(ERROR)}
          )

像往常一样，我认为案例 2 很好。
我也参考了 github 中的一些源代码，我看到有些人确实喜欢案例 1。
我试着问自己doOnSuccess这里的用例是什么？

是否有我们需要应用doOnSuccess()运算符的用例？

我正在尝试使用 Kotlin Flows 创建一个移动数据窗口。它可以在 RxKotlin 中使用缓冲区来实现，但缓冲区与使用 Flows 不同。

RxKotlin 有一个buffer运算符，定期将 Observable 发出的项目收集到包中并发出这些包，而不是一次发出一个项目 - buffer(count,skip)

Kotlin Flow 有一个buffer但只是在单独的协程中运行收集器 -缓冲区

Flows 中是否有现有的运算符可以实现此目的？