如何在没有 Subjects 的情况下使用 RXJava 和 Observables 抽象出硬件？

Question

我正在做一个项目，我有一个硬件设备通过 USB 电缆与我的 Android 应用程序通信。协议缓冲区，包裹在一个薄序列化层中是我的通信协议。

目前，USB 通信使用监视 USB 的异步侦听器进行。从这里，我读取来自我的硬件的字节，对它们采取行动（例如，反序列化它们，解开 protobuf 等...），并适当地设置任何成员变量等。

我对这个类 + 服务的公共 API 是 2 个 Observables：

Observable<DeviceInfo> getDeviceInfoObservable() 
Observable<DeviceState> getDeviceStateObservable()

这是诀窍：填充 DeviceInfo 需要 5 次来回异步调用（不是我的设计），只要应用程序打开，DeviceState 就会不断从硬件发送出去。

对于设备信息，我等到所有 5 个都准备好，然后通知我的观察者（我知道我可以进行订阅端过滤，但我宁愿推送一个完全准备好的 DeviceInfo），对于状态，我在我收到它们时推送它们。

我的问题：有没有办法在不使用主题作为 DeviceInfo 和 DeviceState 的后备存储的情况下做到这一点？我一直在阅读有关 Subjects 有多邪恶的文章，虽然我不相信它们是邪恶的（但是，它们不一定是“功能性的”），但这似乎是一个我无法控制的外部 IO 的好用处。

mSerial.addReadListener(size -> {
  byte[] data = mSerial.read(size);
  // extract data into something usable
  ...
  if (mDeviceInfo.isReady()) {
    mInfoSubject.onNext(mDeviceInfo);
  }
  ...
  mSubject.onNext(mDeviceState);
}

Answer 1

你可以像这样将你的监听器包裹在 Observable 周围：

Observable.create(subscriber -> mSerial.addReadListener(
        size -> {
            byte[] read = mSerial.read(size);
            subscriber.onNext(read);

            subscriber.add(Subscriptions.create(() -> {
                // Remove listener
            }));
        }));

然后你可以映射这个 Observable 并从字节数组中获取你需要的任何设备信息。