prototype for a worldwind observable

worldwind observable

public class WorldWindowRx {
 
    private final WorldWindow worldWindow;
    private final Observable<PositionEvent> mousePositions;
 
    public WorldWindowRx(WorldWindow worldWindow){
        this.worldWindow = worldWindow;
 
        this.mousePositions = createMousePositionObservable(); 
    }
 
    private Observable<PositionEvent> createMousePositionObservable(){
        final Observable<PositionEvent> mousePositions =
            Observable.create(new Observable.OnSubscribe<PositionEvent>() {
                @Override
                public void call(final Subscriber<? super PositionEvent> subscriber) {
                    final PositionListener listener = new PositionListener() {
                        @Override
                        public void moved(PositionEvent event) {
                            subscriber.onNext(event);
                        }
                    };
                    worldWindow.addPositionListener(listener);
 
                    subscriber.add(SwingSubscriptions.unsubscribeInEventDispatchThread(new Action0() {
                        @Override
                        public void call() {
                            worldWindow.removePositionListener(listener);
                        }
                    }));
                }
            });
 
        return mousePositions;
    }
 
    public Observable<PositionEvent> observeMousePosition(final Observable<Object> terminator){
        final Observable<PositionEvent> mousePositions = this.mousePositions.takeUntil(terminator);
        return mousePositions;
    }
}