Consuming an reactive version of SensorManager that has been wrapped using RxJava

fromAsync-4.java

public class SensorRxActivity extends Activity { 
  private static final String LOG_TAG = SensorRxActivity.class.getName(); 
  private SensorManager sensorManager; 
  private Sensor accelerometer; 
  private Subscription sensorChangedSubscription; 

  @Override 
  protected void onCreate(Bundle savedInstanceState) { 
    super.onCreate(savedInstanceState); 
    sensorManager = (SensorManager) getSystemService(SENSOR_SERVICE); 
    accelerometer = sensorManager.getDefaultSensor(Sensor.TYPE_ACCELEROMETER); 
  } 

  protected void onResume() { 
    super.onResume(); 
    sensorChangedSubscription = naiveObserveSensorChanged(sensorManager, accelerometer, SensorManager.SENSOR_DELAY_FASTEST) 
      .subscribe(sensorChangedOnNext, sensorChangedOnError); 
  } 

  @Override 
  protected void onPause() { 
    super.onPause(); 
    sensorChangedSubscription.unsubscribe(); 
  } 

  private final Action1<SensorEvent> sensorChangedOnNext = new Action1<SensorEvent>() { 
    @Override 
    public void call(SensorEvent sensorEvent) { 
      Log.d(LOG_TAG, "sensorChangedOnNext - sensorEvent.timestamp=" + sensorEvent.timestamp + ", sensorEvent.values=" + Arrays.toString(sensorEvent.values)); 
    } 
  }; 

  private final Action1<Throwable> sensorChangedOnError = new Action1<Throwable>() { 
    @Override 
    public void call(Throwable throwable) { 
      Log.e(LOG_TAG, "sensorChangedOnError", throwable); 
    } 
  }; 
}