反应式编程：如何订阅和采样事件发射器？

Question

我正在订阅一个使用 react-native-ble-manager 的 React Native 应用程序中的事件发射器。

handleUpdateValueForCharacteristic(data) {

    console.log('Received data from ' + data.peripheral + ' characteristic ' + data.characteristic, data.value);

  }
bleManagerEmitter.addListener('BleManagerDidUpdateValueForCharacteristic', this.handleUpdateValueForCharacteristic );

我正在处理一个蓝牙事件流，其频率为每秒 50、100 或 200 个事件 (Hz)。

我对 50 赫兹的所有事件感兴趣，其中一半的 100 赫兹，四分之一的 200 赫兹。 使用 RxJS 订阅此事件流的正确方法是什么，我应该使用哪个运算符来采样数据？

我可能错了，但我似乎找不到从事件发射器创建可观察对象的辅助方法。

Answer 1

fromEventPattern 应该是您要查找的内容。

它允许您根据自定义事件发射创建一个可观察对象（就像您使用这个 BLE 管理器所拥有的一样）。

我在下面提供了一个 sn-p，概述了您可以如何使用它。

注意scan() 和filter() 的组合。通过使用前一个运算符来跟踪第 n^th 事件，它有效地改变了事件被采样并因此被任何订阅者处理的速率。

在您的场景中，您希望scan() 也跟踪发出的事件，因此您最终可以在filter() 调用之后map() 它，这样订阅者就会收到它。这里的关键点是在累积事件时跟踪scan() 中的事件状态（即tick 和事件数据属性，分别在sn-p 中的t 和data）。

const { fromEventPattern } = rxjs;
const { filter, map, scan } = rxjs.operators;

// Tweak parameters to vary demo
const hz = 200;
const sample = 4;

function addEmitterHandler(handler) {
  // bleManagerEmitter.addListener('event', handler)
  const intervalId = setInterval(() => {
    handler({ timestamp: Date.now() });
  }, 1000 / hz);
  return intervalId;
}

function removeEmitterHandler(handler, intervalId) {
  // bleManagerEmitter.removeListener(...)
  clearInterval(intervalId);
}

// Simulate emissions using the `setInterval()` call
const emitter = fromEventPattern(
  addEmitterHandler,
  removeEmitterHandler
);

emitter.pipe(
  // Use `scan()` and `filter()` combination to adjust sampling
  scan((state, data) => {
    const t = (state.t % (sample + 1)) + 1;
    return { t, data };
  }, { t: 0, data: null }),
  filter(state => state.t % sample === 0),
  // Use `map()` to forward event data only
  map(state => state.data),
).subscribe(data => console.log(data));

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>