share() 运算符不适用于 Rxjava 中的 Observable

Question

我有一个场景，我有一个 emmiter，它不断发出这样的数据

fun subscribeForEvents(): Flowable<InkChannel> {
    return Flowable.create<InkChannel>({
        if (inkDevice.availableDeviceServices.contains(DeviceServiceType.EVENT_DEVICE_SERVICE)) {
            (inkDevice.getDeviceService(DeviceServiceType.EVENT_DEVICE_SERVICE) as EventDeviceService).subscribe(object : EventCallback {
                override fun onUserActionExpected(p0: UserAction?) {
                    it.onNext(InkChannel.UserActionEvent(p0))
                }

                override fun onEvent(p0: InkDeviceEvent?, p1: Any?) {
                    it.onNext(InkChannel.InkEvents<Any>(p0, p1))
                }

                override fun onUserActionCompleted(p0: UserAction?, p1: Boolean) {

                }
            }

            )
        }
    }, BackpressureStrategy.BUFFER).share()

}

现在我有一个服务，我在应用程序启动时启动并收听它

  inkDeviceBus.subscribeForEvents()
                .filter { it -> (it as InkChannel.InkEvents<*>).event == InkDeviceEvent.STATUS_CHANGED }
                .map { it -> it as InkChannel.InkEvents<*> }
                .map { it -> it.value.toString() }
                .filter { value -> value == "CONNECTED" || value == "DISCONNECTED" }
                .map { it -> it == "CONNECTED" }
                .subscribeBy { b ->
                    if (b) stopSelf()
                }

我有另一个 activity MainActivity，它在启动时调用，我观察到相同的事件。现在的问题是只有服务中的侦听器获取事件，而 activity 没有接收到任何事件。

现在，当我从服务中删除侦听器时，activity 开始接收事件。我已经使用运算符 share 来共享可观察对象，但它似乎不起作用

Answer 1

.share() 只影响调用它的实例。由于每次调用 subscribeForEvents 都会创建一个新实例，因此 .share() 不会改变行为。

您需要调用 subscribeForEvents 一次，然后在您想要接收事件时使用返回值。只要使用同一个对象，它就会共享底层监听器。

share() 运算符不适用于 Rxjava 中的 Observable

share() operator not working for Observable in Rxjava

android

rx-java2

rx-android

rx-kotlin2