在一个方法中创建一个临时的 Observable

Question

代码：

static void readCharacteristic( RxBleDevice device, UUID characteristic, Action1<byte[]> callback ) {
    Observable<RxBleConnection> deviceConnObservable = getSharedConnectionObservable( device );
    deviceConnObservable
        .subscribe(
                callback, 
                throwable -> Log.d( TAG, "Error" ),
                () -> {  } // completed the read
        );
}

我推测当方法退出时，deviceConnObservable 被销毁。但是 Subscriber 中的处理程序呢？它们是否也消失了，或者它们继续存在并被调用——也许是 Observable 被销毁时的 onError() 处理程序？

假设他们不这样做，是否存在可以在方法中创建 Observable 的情况？看起来如果你等待它的 onCompleted() 通知应该没问题，不是吗？

Answer 1

如果您的 getSharedConnectionObservable 在当前线程中运行，则订阅在方法退出之前完成。

如果您的 getSharedConnectionObservable 是异步的，则存在指向您的 Observable 的引用链。

Worker(Thread) -> Task(Action0) -> {DownStream(Subscriber) & UpStream(Observable)(OperatorSubscribeOn)}

你的Observable就完成了，你可以自己试试

在一个方法中创建一个临时的 Observable

Creating a temporary Observable inside a method

rx-java

reactive