Reactive Programming: How to subscribe and sample an event emitter?

Question

I'm suscribing to an event emitter in a React Native application that is consuming react-native-ble-manager.

handleUpdateValueForCharacteristic(data) {

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

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

I'm dealing with a Bluetooth event Stream which frequency is either 50, 100 or 200 events per second (Hz).

I'm interrested in all events at 50 Hz, half of them at 100 Hz and a quarter of them at 200 Hz. What is the correct way to subscribe to this event Stream with RxJS and which operator should I use to sample the data?

I may be wrong but I can't seem to find a helper method to create an observable from an event emitter.

Answer 1

fromEventPattern should be what you're looking for.

It lets you create an observable based on custom event emissions (like what you've got with this BLE manager).

I've provided a snippet below outlining how you might use it.

Notice the scan() and filter() combination. By using the former operator to keep track of the n^th event, it effectively changes the rate at which events are sampled and thus handled by any subscribers.

In your scenario, you'll want the scan() to also keep track of the event emitted, so you can eventually map() it after the filter() call, so subscribers will receive it. The key point here is to keep track of event state within the scan() (i.e. tick and event data properties, t and data in the snippet respectively) as you accumulate events.

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>