reactive4java で位置を取得し続ける

//LocationFunctions.java
/**
 * 位置を取得し続ける(finish は呼ばれない)
 */
public static Observable<Location> getCurrentLocationAsObservable(			
		final Context context, final String provider) {
	return Reactive.createWithCloseable(new Func1<Observer<? super Location>, Closeable>() {
		private volatile boolean _stop = false;

		@Override
		public Closeable invoke(final Observer<? super Location> observer) {
			final LocationManager locMan = (LocationManager)context.getSystemService(Context.LOCATION_SERVICE);
			
			final LocationListener listener = new LocationListener() {
				
				@Override
				public void onStatusChanged(String provider, int status, Bundle extras) { }
				
				@Override
				public void onProviderEnabled(String provider) { }
				
				@Override
				public void onProviderDisabled(String provider) {
					observer.error(
						new InvalidParameterException("LocationProvider disabled."));
				}
				
				@Override
				public void onLocationChanged(Location location) {
					if (_stop) {
						return;
					}
					
					// 発火
					observer.next(location);
				}
			};
			
			// 位置取得開始
			locMan.requestLocationUpdates(provider, 0, 0, listener, Looper.getMainLooper());

			return new Closeable() {
				@Override
				public void close() throws IOException {
					if (_stop) {
						return;
					}
					_stop = true;
					locMan.removeUpdates(listener);
					observer.finish();
				}
			};
		}
	});
}