Wrapping an asynchronous API that uses listeners, by using RxJava Observable.fromAsync

fromAsync-6.java

public Observable<SensorEvent> observeSensorChanged(final SensorManager sensorManager, final Sensor sensor, final int samplingPeriodUs) { 
  return Observable.fromAsync(new Action1<AsyncEmitter<SensorEvent>>() { 
    @Override 
    public void call(final AsyncEmitter<SensorEvent> sensorEventAsyncEmitter) { 
      final SensorEventListener sensorListener = new SensorEventListener() { 
        @Override 
        public void onSensorChanged(SensorEvent sensorEvent) { 
          sensorEventAsyncEmitter.onNext(sensorEvent); 
        } 

        @Override 
        public void onAccuracyChanged(Sensor originSensor, int i) { 
          // ignored for this example 
        } 
      }; 
      // (1) - unregistering listener when unsubscribed 
      sensorEventAsyncEmitter.setCancellation(new AsyncEmitter.Cancellable() { 
        @Override 
        public void cancel() throws Exception { 
          sensorManager.unregisterListener(sensorListener, sensor); 
        } 
      }); 
      sensorManager.registerListener(sensorListener, sensor, samplingPeriodUs); 

    } 
    // (4) - specifying the backpressure strategy to use 
  }, AsyncEmitter.BackpressureMode.BUFFER); 
}