Exploration of ObservableCollections about rxjavafx HOT 4 CLOSED

public final class ObservableListSource {

    public static <T> Observable<T> fromObservableListAdds(final ObservableList<T> source) {

        return Observable.create((Observable.OnSubscribe<T>) subscriber -> {

            source.addListener((ListChangeListener<T>) c -> {
                while (c.next()) {
                    if (c.wasAdded()) {
                        c.getAddedSubList().forEach(subscriber::onNext);
                    }
                }
            });
        }).subscribeOn(JavaFxScheduler.getInstance());
    }
    public static <T> Observable<T> fromObservableListRemovals(final ObservableList<T> source) {

        return Observable.create((Observable.OnSubscribe<T>) subscriber -> {

            source.addListener((ListChangeListener<T>) c -> {
                while (c.next()) {
                    if (c.wasRemoved()) {
                        c.getRemoved().forEach(subscriber::onNext);
                    }
                }
            });
        }).subscribeOn(JavaFxScheduler.getInstance());
    }
}

 @Test
    public void testObservableList() {
        JFXPanel panel = new JFXPanel();

        ObservableList<String> items = FXCollections.observableArrayList();

        ObservableListSource.fromObservableListAdds(items).subscribe(v -> System.out.println("ADDED: " + v));
        ObservableListSource.fromObservableListRemovals(items).subscribe(v -> System.out.println("REMOVED: " + v));

        items.add("ABQ");
        items.add("HOU");

        items.setAll("LAX","PHX","FLL");
    }

 @Test
    public void testObservableList() {
        JFXPanel panel = new JFXPanel();

        Platform.runLater(() -> {
            ObservableList<String> items = FXCollections.observableArrayList();

            ObservableListSource.fromObservableListDistinctChanges(items)
                    .subscribe(System.out::println, Throwable::printStackTrace);

            items.add("ABQ");
            items.add("HOU");
            items.add("CHI");
            items.add("ABQ");
            items.add("HOU");

            items.remove("ABQ");
            items.remove("CHI");
            items.remove("ABQ");

            items.add("ABQ");
        });

        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }