kotlin 让订阅者使用 RxJava2 观察可观察对象

Question

Android Studio 3.0 Beta2

我创建了两种方法，一种用于创建可观察对象，另一种用于创建订阅者。

但是，我在尝试让订阅者订阅可观察对象时遇到问题。在 Java 中这可以工作，我正在尝试让它在 Kotlin 中工作。

在我的 onCreate(..) 方法中，我试图设置它。这是正确的方法吗？

class MainActivity : AppCompatActivity() {

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        /* CANNOT SET SUBSCRIBER TO SUBCRIBE TO THE OBSERVABLE */
        createStringObservable().subscribe(createStringSubscriber())
    }


    fun createStringObservable(): Observable<String> {
        val myObservable: Observable<String> = Observable.create {
            subscriber ->
            subscriber.onNext("Hello, World!")
            subscriber.onComplete()
        }

        return myObservable
    }

    fun createStringSubscriber(): Subscriber<String> {
        val mySubscriber = object: Subscriber<String> {
            override fun onNext(s: String) {
                println(s)
            }

            override fun onComplete() {
                println("onComplete")
            }

            override fun onError(e: Throwable) {
                println("onError")
            }

            override fun onSubscribe(s: Subscription?) {
                println("onSubscribe")
            }
        }

        return mySubscriber
    }
}

非常感谢您的建议，

Answer 1

密切注意类型。

Observable.subscribe() 具有三个基本变体:

• 不接受任何参数
• 几个接受 io.reactivex.functions.Consumer
• 接受 io.reactivex.Observer

您在示例中尝试订阅的类型是org.reactivestreams.Subscriber(定义为响应式流规范的一部分)。你可以refer to the docs以获得这种类型的更全面的统计，但足以说明它与任何重载的 Observable.subscribe() 方法都不兼容。

这是 createStringSubscriber() 方法的修改示例，该方法将允许您的代码进行编译:

fun createStringSubscriber(): Observer<String> {
        val mySubscriber = object: Observer<String> {
            override fun onNext(s: String) {
                println(s)
            }

            override fun onComplete() {
                println("onComplete")
            }

            override fun onError(e: Throwable) {
                println("onError")
            }

            override fun onSubscribe(s: Disposable) {
                println("onSubscribe")
            }
        }

        return mySubscriber
    }

改变的事情是:

1. 这会返回一个 Observer 类型(而不是 Subscriber)
2. onSubscribe() 传递一个 Disposable(而不是 Subscription)

.. 正如“Vincent Mimoun-Prat”所提到的，lambda 语法确实可以缩短您的代码。

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)

        // Here's an example using pure RxJava 2 (ie not using RxKotlin)
        Observable.create<String> { emitter ->
            emitter.onNext("Hello, World!")
            emitter.onComplete()
        }
                .subscribe(
                        { s -> println(s) },
                        { e -> println(e) },
                        {      println("onComplete") }
                )

        // ...and here's an example using RxKotlin. The named arguments help
        // to give your code a little more clarity
        Observable.create<String> { emitter ->
            emitter.onNext("Hello, World!")
            emitter.onComplete()
        }
                .subscribeBy(
                        onNext     = { s -> println(s) },
                        onError    = { e -> println(e) },
                        onComplete = {      println("onComplete") }
                )
    }

希望有帮助!