fromAsync-5.java

public Observable<SensorEvent> naiveObserveSensorChanged(final SensorManager sensorManager, final Sensor sensor, final int samplingPreiodUs) { 
  return Observable.create(new Observable.OnSubscribe<SensorEvent>() { 
    @Override 
    public void call(final Subscriber<? super SensorEvent> subscriber) { 
      final SensorEventListener sensorEventListener = new SensorEventListener() { 
        @Override 
        public void onSensorChanged(SensorEvent event) { 
          // (3) - checking for subscribers before emitting values 
          if (!subscriber.isUnsubscribed()) { 
            subscriber.onNext(event); 
          } 
        } 

        @Override 
        public void onAccuracyChanged(Sensor sensor, int accuracy) { 
          // ignored for this example 
        } 
      }; 

      // (1) - unregistering listener when unsubscribed 
      subscriber.add(Subscriptions.create(new Action0() { 
        @Override 
        public void call() { 
          try { 
            sensorManager.unregisterListener(sensorEventListener, sensor); 
          } catch (Exception ex) { 
            // (3) - checking for subscribers before emitting values 
            if (!subscriber.isUnsubscribed()) { 
              // (2) - reporting exceptions via onError() 
              subscriber.onError(ex); 
            } 
          } 
        } 
      })); 

      sensorManager.registerListener(sensorEventListener, sensor, samplingPreiodUs); 
    } 
  }); 
}