Wrapping event listeners in Observables

Question

I've seen a lot of examples of how to turn finite things like arrays or Iterables into Observables, but I'm not sure I understand how to make an Observable out of something live and effectively unbounded like an event receiver. I studied the RxJava2 docs and came up with this, using an Android LocationListener as an example.

Is there a simpler and/or more correct way to do this? I'm aware of the "RxBus" concept, but it seems like a way of clinging to the old event bus paradigm.

final Observable<Location> locationObservable = Observable.create(new ObservableOnSubscribe<Location>() {
    final LocationManager mLocationManager = (LocationManager) getSystemService(LOCATION_SERVICE);

    @Override
    public void subscribe(final ObservableEmitter<Location> emitter) throws Exception {
        final LocationListener listener = new LocationListener() {
            @Override
            public void onLocationChanged(final Location location) {
                emitter.onNext(location);
            }

            @Override
            public void onStatusChanged(final String s, final int i, final Bundle bundle) {
                // TODO ???
            }

            @Override
            public void onProviderEnabled(final String s) {
                // TODO ???
            }

            @Override
            public void onProviderDisabled(final String s) {
                // TODO ???
            }
        };

        mLocationManager.requestLocationUpdates(LocationManager.GPS_PROVIDER, 0, 0, listener);

        emitter.setCancellable(new Cancellable() {
            @Override
            public void cancel() throws Exception {
                mLocationManager.removeUpdates(listener);
            }
        });

        emitter.setDisposable(new Disposable() {
            private AtomicBoolean mDisposed;

            @Override
            public void dispose() {
                if(mDisposed.compareAndSet(false, true)) {
                    mLocationManager.removeUpdates(listener);
                }
            }

            @Override
            public boolean isDisposed() {
                return mDisposed.get();
            }
        });
    }
});

score 1 · Accepted Answer · answered Jun 29 '17 at 20:13

using Observable.create() is indeed a correct way.

However, with RxJava2 the default way is to extend an Observable, you can see this answer for greater details.

some comments though regarding your implementation:
- there is no point setting both Cancellable and Disposable, as the later one will cancel/dispose the first one, you can see the difference between them here.
- I think it's best practice, to register cancellable/disposable before you start listening to update, in order to prevent weird edge cases races.

Wrapping event listeners in Observables

1 Answers1

Linked