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);
    }
  };
}