reactivex / rxjavafx Goto Github PK

Is there any plan to migrate Observables to Flowables in version 2 of RxJavaFx, as well as move the maven artifact to the groupId io.reactivex.rxjava2 according to version 2 of RxJava to allow coexistence of versions 1 and 2?

Instead of having the doOnXXXFX() action operators only available on RxKotlinFX as extension functions, I will make them available in RxJavaFX as Transformers as cumbersome as they are. These will be in the 1.0 release.

I added some doOnXXXCount() operators to RxKotlinFX. Specifically these are the doOnNextCount(), doOnErrorCount(), and doOnCompletedCount() operators, which are analogous to doOnNext(), doOnCompleted(), etc already found in RxJava. The only difference is these emit the cumulative emission counts for each of those events. This is helpful to post updates on a UI showing how many items have been "processed" by an Observable.

Here is the example written in Kotlin:

val source = Observable.range(1,1000)
val processedCountLabel = Label()

source.map { it * 10 }
     .doOnNextFx { processedCountLabel.text = "Processed $it items" }
     .subsribe { doSomethingWith(it) }

I added these to RxKotlinFX because extension functions make this more elegant and less cumbersome to use. However, I could port this to RxJava RxJavaFX and allow the Operator to be passed to the lift() method.

Does anybody feel this would be helpful to add/not add?

I may explore creating a Binding factory that returns a Subscriber that implements Binding. This way an Observable<T> can quickly be subscribed to a JavaFX Binding.

Similar to JavaFxObservable#eventsOf(Node, EventType<T>), JavaFxObservable should have a method to observe Event objects during the capturing phase of event delivery. Internally, such method would use Node#addEventFilter(EventType, EventHandler) instead of Node#addEventHandler(EventType, EventHandler).

I made a typo. The Binding's addto() method should be addTo(). I will deprecate and correct it later.

The CompositeObservable was somewhat of a moving target after I first had the idea. It was a tool to safely decouple RxJava streams in a JavaFX application without exposing a Subject, which can encourage antipatterns especially with Rx newbies.

Although I haven't changed my sentiments on Subjects, I may have softened my stance a little when it comes to decoupling. The CompositeObservable ultimately came to resemble a Subject without exposing the Observer methods. Not to mention, it is somewhat inconvenient having to call its toObservable() method to get the Observable.

Therefore, I am considering deprecating the CompositeObservable and telling people to use PublishSubject, ReplaySubject, etc instead. If anybody has thoughts on this topic, please feel free to share.

Hi. I'm trying to create an Observable out of a button click but apparently it's running only on the first click. The other clicks does nothing. Any ideas why this would happen?

    Button loadUsersButton = new Button("Load users");
    BorderPane root = new BorderPane();
    root.setTop(loadUsersButton);
    HBox githubInfo = new HBox();
    List<VBox> listOfUserInfo = Arrays.asList(new VBox(), new VBox(), new VBox());
    githubInfo.getChildren().addAll(listOfUserInfo);
    root.setCenter(githubInfo);
    final Subject<ActionEvent> loadButtonSubject = PublishSubject.create();

    JavaFxObservable.actionEventsOf(loadUsersButton).subscribe(loadButtonSubject);

    loadButtonSubject
            .map(ae -> githubExternalService.getUsers())
            .flatMapIterable(x -> x)
            .take(3)
            .map(this::createText)
            .map(text -> {
                VBox vBox = new VBox();
                vBox.getChildren().add(text);
                return vBox;
            }).subscribe(
            vbox -> githubInfo.getChildren().add(vbox),
            System.err::println);
    return root;

I'm thinking of creating a CompositeBinding in the same spirit as CompositeSubscription. This way multiple bindings created off Observable items can quickly be disposed as a group.

The README and Learning RxJava with JavaFX eBook need to be updated to reflect RxJavaFX 2.0's features and API.

I am noticing a lot of ObservableList-related factories, particularly emitOnChanged(), are not triggered on a SimpleListProperty when the backing list is changed.

Maybe SimpleListProperty needs its own factories.

As title

In RxJavaFX 2.0, I have set up JavaFxSubscriber to have toBinding() factories overloaded for both Observable and Flowable. Should I separate these two overloads into two separate static utility classes: JavaFxObserver and JavaFxSubscriber? Just to follow RxJava 2.0 terminology?

Is the name of this project correct? Anything better?

What should the artifact and module names be?

What do you think about adding FxCompletableTransformer, FxMaybeTransformer and FxSingleTransformer to complement the existing FxObservableTransformer and FxFlowableTransformer?

There should be an option to make the RxJavaFX Binding implementation lazy. That way it delays subscribing to the Observable until the Binding is first called for the value.

Maybe there needs to be two implementations: EagerBindingSubscriber and LazyBindingSubscriber. The first one is what exists currently as BindingSubscriber.

I think I'll add another factory method to JavaFxObservable that works with Menu and MenuItem action events. Since MenuItem does not extend Node, I cannot use the existing factories.

I am beginning to think that backing the CompositeObservable with a Subject may not be a bad idea. A feature that is missing is the ability to "unsubscribe" the CompositeObservable from a source without an emission to trigger the takeWhile(). If the add() method returns a Subscription, the client can easily remove streams going into the CompositeObservable. I am also thinking of making it share() by default.

None of this should change the signatures of the CompositeObservable.

I'd like to think this library is maturing and almost ready for a 1.0 release. After I complete a project with O'Reilly this fall, I'm going to put together a GitBook guide covering RxJavaFX/RxKotlinFX in detail. This will very much be in the spirit of the TornadoFX Guide which I'm also going to finish before October.

I may use this book to teach RxJava but from a UI perspective with JavaFX. It will also cover design patterns and tricks you can do with certain operators. I'll include Java and Kotlin versions of code examples.

Here is the working outline:

I. RxJava Basics
II. UI Events
III. Reactive Binding
IV. Data Controls and Collection Events
V. Concurrency
VI. Model Patterns with CompositeObservable
VII. Kotlin and TornadoFX

RxJava added Automatic-Module-Name: io.reactivex.rxjava2 to the manifest, in order to have a small step into the Java 11 full module transition. This was in the following clause of the build.graddle file:

jar {
    manifest {
        name = "rxjava"
        instruction "Bundle-Vendor", "RxJava Contributors"
        instruction "Bundle-DocURL", "https://github.com/ReactiveX/RxJava"
        instruction "Import-Package", "!org.junit,!junit.framework,!org.mockito.*,!org.testng.*,*"
        instruction "Eclipse-ExtensibleAPI", "true"
        instruction "Automatic-Module-Name", "io.reactivex.rxjava2"
    }
}

Can this be made also for RxJavaFX?

I have noticed the JavaFxScheduler does not always get along when the Platform thread is setting up the scheduling. In some cases I have seen it hang up and values never get emitted when subscribed on the JavaFxScheduler.

My initial tests indicate this can be fixed by checking if the Platform thread is already calling the scheduler setup, and if so just execute the task right then and there (bold below).

       @Override
        public Subscription schedule(final Action0 action) {
            final BooleanSubscription s = BooleanSubscription.create();

            if (Platform.isFxApplicationThread()) {
                if (!(innerSubscription.isUnsubscribed() && s.isUnsubscribed())) {
                    action.call();
                    innerSubscription.remove(s);
                }
            }
            else{
                Platform.runLater(() -> {
                    if (innerSubscription.isUnsubscribed() || s.isUnsubscribed()) {
                        return;
                    }
                    action.call();
                    innerSubscription.remove(s);
                });
            }
            innerSubscription.add(s);
            // wrap for returning so it also removes it from the 'innerSubscription'
            return Subscriptions.create(new Action0() {

                @Override
                public void call() {
                    s.unsubscribe();
                    innerSubscription.remove(s);
                }

            });
        }

I've not kept up with Java 9/10 releases and chose to align RxJavaFX with TorndoFX's approach of waiting for Java 11.

@protogenes has already started filing issues and PR's, and I know other libraries have already transitioned as well.

I'm going to try my best to get RxJavaFX moved to Java 11 in the next few weeks.

EDIT Rx works fine. Described issue is caused by weak refernces.
I wanted to bind two properties and listen for changes on both, using RxJavaFx.
What I expected was - bind two properties, write to only one of them and listen for changes on both. It turned out it worked as I expected only using pure fx. Using RxJavaFx (please uncomment lines) causes firing change event only on directly updated property (even fx listener stopped to work).
Another strange thing is that at first everything seems to work perfectly ("wow" line).

import javafx.application.Application;
import javafx.beans.binding.Bindings;
import javafx.beans.property.ObjectProperty;
import javafx.beans.property.SimpleObjectProperty;
import javafx.beans.value.ChangeListener;
import javafx.scene.Scene;
import javafx.scene.control.Button;
import javafx.scene.layout.HBox;
import javafx.stage.Stage;

public class BidingsTestApp extends Application {

    public static void main(String[] args) {
        launch(args);
    }
    
    @Override
    public void start(Stage stage) throws Exception {
        Button button1 = new Button("one");
        Button button2 = new Button("two");
        HBox hBox = new HBox(button1, button2);
        
        stage.setScene(new Scene(hBox));
        stage.show();
        
        ObjectProperty<String> property1 = new SimpleObjectProperty<>();
        ObjectProperty<String> property2 = new SimpleObjectProperty<>();
        
        Bindings.bindBidirectional(property1, property2);
        
        property1.addListener((ChangeListener<String>) (observable, oldValue, newValue) -> System.out.println("fx property1 changed to value=" + newValue));
        property2.addListener((ChangeListener<String>) (observable, oldValue, newValue) -> System.out.println("fx property2 changed to value=" + newValue));
//        JavaFxObservable.valuesOf(property1).subscribe(text -> System.out.println("rx property1 changed to value=" + text));
//        JavaFxObservable.valuesOf(property2).subscribe(text -> System.out.println("rx property2 changed to value=" + text));
        
        property2.set("wow");
        
        button1.setOnMouseClicked(event -> property2.setValue("one"));
        button2.setOnMouseClicked(event -> property2.setValue("two"));
    }
}

The JavaFxObservable.fromObservableList() factory should probably emit the ObservableList on subscription as the initial emission. I'm finding myself putting a startsWith() after it to accomplish this all the time.

Merging #20 seems to have caused a build issue caused by testPeriodicScheduling() failing

I found a need to emit changes of an ObservableList as an Observable, such as dealing with multiple row selections in a TableView. It would be nice to emit those selection events even though they come in an ObservableList.
#6 brought up a similar idea a while back, but I think it might be helpful to emit changes from an ObservableList with each of those changes categorized. I'll document as I explore in the coming weeks.

The usage of internal com.sun classes breaks on newer Java versions.

./src/main/java/io/reactivex/rxjavafx/observers/BindingObserver.java
import com.sun.javafx.binding.ExpressionHelper;
./src/main/java/io/reactivex/rxjavafx/observers/BindingSubscriber.java
import com.sun.javafx.binding.ExpressionHelper;

I read the documentation on the release candidate of RxJava 2.0, and it is pretty clear disruptive (but good) changes are on the horizon that directly affect RxJavaFX.

https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0

I haven't dived deep into this yet or even messed with the released candidate, but it does seem a major refactoring is necessary since RxJava was basically rewritten from scratch with the coming 2.0 release.

I'll investigate this further, but it seems RxJavaFX would want to use the Flowable rather than the Observable. I am not a backpressure expert, but my understanding is backpressure cannot be helped with UI events since a person cannot be directed to slow down emissions. Therefore, a Flowable makes sense, and I can't think of any factories in RxJavaFX that shouldn't use it.

I think the best approach is to release RxJavaFX 1.0 and 2.0 as separate maintained versions, aligned with the respective RxJava counterparts. The same will likely happen with RxKotlinFX.

Again, I didn't dig deep in RxJava 2.0 but I think I have a high level of understanding. If anybody has any concerns, ideas, feedback, or just anything they would want to see in RxJavaFX 1.0 and 2.0, please let us know here on this issue.

I discovered a factory to implement reactive dialogs and alerts is quite simple.

Implementation

public static <T> Observable<T> fromDialogSource(final Dialog<T> dialog) {
    return Observable.fromCallable(dialog::showAndWait)
            .subscribeOn(JavaFxScheduler.getInstance())
            .filter(Optional::isPresent)
            .map(Optional::get);
}

Usage

JavaFxObservable.fromDialog(alert)
    .filter(response -> response.equals(ButtonType.OK))
    .subscribe(System.out::println,Throwable::printStackTrace);

I think I'm going to do a release with this and the fix for #36. Did I oversimplify this factory? Is there anything else that needs to be added?

I think this was mentioned before in an old PR, but I'm rethinking it might be beneficial to add another factory to handle WindowEvent emissions.

Here is what it might look like when called.

Observable<WindowEvent> winEvents = 
    JavaFXObservable.fromWindowEvents(myWindow, WindowEvent.WINDOW_HIDING);

I mocked this up in Kotlin real quick. Will implement this in Java later.

fun <T : Window> T.windowEvents(windowEvent: EventType<WindowEvent>): Observable<WindowEvent> {
    val window = this
    return Observable.create(Observable.OnSubscribe<WindowEvent> { subscriber ->
        val handler = EventHandler<WindowEvent> { subscriber.onNext(it) }

        window.addEventHandler(windowEvent, handler)

        subscriber.add(JavaFxSubscriptions.unsubscribeInEventDispatchThread { window.removeEventHandler(windowEvent, handler) })

    }).subscribeOn(JavaFxScheduler.getInstance())
}

Currently the JavaFxScheduler depends on the keyframe finish callback of JavaFX animations.
But animations with a duration of zero will never start, thus the callback (of the keyframe) will not be called.
See javafx.animation.Animation.java (JDK 8u112):

boolean impl_startable(boolean forceSync) {
    return (fromDuration(getCycleDuration()) > 0L)
        || (!forceSync && clipEnvelope.wasSynched());
}

One possible solution is to use the finish callback of the timeline instead: timer.setOnFinished(executeOnce);

I am going to add Observable source factories to emit events from ObservableMap and ObservableSet. I wonder if it is time to start thinking about a 1.0 release. I can't really think of anything else that needs to be added.

I think I may add another factory in JavaFxObservable that emits an old value with the new value in a change object.

I'm having some build issues probably related to deployment tasks. I have a strong guess the Nebula plugin is no longer supported by Netflix, which breaks with newer versions of Gradle.

@akarnokd et al, can you help me get my build files set up correctly?

I think I'm going to implement null/non-null factories for RxJavaFX in both the 1.0 and 2.0 releases. ReactFX took this approach and it seems to work quite well. It will also simplify things since RxJava 2.0 does not support null emissions, and only nullable factories will need to emit Optional<T> types.

https://github.com/TomasMikula/ReactFX#event-streams-in-javafx

This is not yet building ... someone who works with JavaFX needs to figure this out.

I can make it build with Java 8, but there must be a way to get the correct JavaFX dependencies or build config for Java 7?

Also, when building with Java 8 some unit tests fail.

Rather than this implementation:

    public static <T> Observable<T> fromDialogSource(final Dialog<T> dialog) {
        return Observable.fromCallable(dialog::showAndWait)
                .subscribeOn(JavaFxScheduler.platform())
                .filter(Optional::isPresent)
                .map(Optional::get);
    }

It should instead be returning a Maybe:

    public static <T> Maybe<T> fromDialogSource(final Dialog<T> dialog) {
        return Single.fromCallable(dialog::showAndWait)
                .subscribeOn(JavaFxScheduler.platform())
                .filter(Optional::isPresent)
                .map(Optional::get);
    }

Hi,
javafx.collections.ListChangeListener among other methods also contains getFrom and getTo methods, indicating array indexes of added/modified deleted entries sub lists. Would it be possible to extend io.reactivex.rxjavafx.sources.ListChange to provide an array index of the changed item? Thank you!

Hello,

RxJava gave functional names to Schedulers accessor : Schedulers.io(), Schedulers.computation(), Schedulers.newThread(), ...

RxJavaFX has a technical name for the JavaFX Scheduler accessor : JavaFxScheduler.getInstance().
Even if you can understand it at the first sight, it can be more revelant to use a fonctional name instead, like : JavaFxScheduler.ui()

RxAndroid use the same logic, and hidding the singleton instance though RxAndroid.mainThread() : https://github.com/ReactiveX/RxAndroid/blob/1.x/rxandroid/src/main/java/rx/android/schedulers/AndroidSchedulers.java#L56

TLDR : Is renaming JavaFxScheduler.getInstance() to JavaFxScheduler.ui() can be an option ?

Regards.

I am encountering a need heavily where I want to merge() multiple Observable event sources, but I don't have the sources available upon initialization. I need to add them later at separate places in the application.

With David Karnok's help, I came up with a utility. Here is the Kotlin implementation.

class ObservableBus<T>(cacheCount: Int = 0) {
​
    private val subject: SerializedSubject<Observable<T>, Observable<T>> = PublishSubject<Observable<T>>().toSerialized()
    private val live: MutableSet<Observable<T>> = ConcurrentHashMap.newKeySet<Observable<T>>()
​
    private val observable = subject.flatMap { obs: Observable<T> -> obs.takeWhile { live.contains(obs) } }
            .observeOnFx().let { if (cacheCount > 0) it.cacheWithInitialCapacity(cacheCount) else it }
​
    fun toObservable(): Observable<T> = observable
​
    operator fun plusAssign(observable: Observable<T>) = add(observable)
    operator fun minusAssign(observable: Observable<T>) = remove(observable)
​
    fun add(observable: Observable<T>) {
        live.add(observable)
        subject.onNext(observable)
    }
    fun remove(observable: Observable<T>) {
        live.remove(observable)
    }
}

This is also helpful for creating event-driven models. So if you have a MenuItem, Button, and key combo that all trigger a refresh request, you can merge all three Observables later.

object MyEventModel { 
    val refreshRequests = ObservableBus<ActionEvent>
}

val button = Button("Refresh")
MyEventModel.refresh += button.actionEvents()

//later
val menuItem = MenuItem("Refresh")
MyEventModel.refresh += button.actionEvents()

//and later again
val ctrlRPress = someNode.events(KeyEvent.KEY_PRESSED,)
     .filter { it.isControlDown && it.keyCode = KeyCode.R }
     .map { ActionEvent() }

MyEventModel.refresh += ctrlRPress

Subscription

MyEventModel.refresh.subscribe { performRefresh }

Of course, you can push anything and not just ActionEvent items. Does anybody else see value in this? I'll write this in Java if there is interest.

In my opinion the first/initial emission of nullSentinel is missing:

ObjectProperty<String> stringProperty = new SimpleObjectProperty<>();
JavaFxObservable.valuesOf(stringProperty, "N/A").subscribe(System.out::println);  
stringProperty.set("one");  
stringProperty.set(null);  
stringProperty.set("two");

Output:
one
N/A
two

I would add "else" block just after "if" at ObservableValueSource:50

if (fxObservable.getValue() != null) {
    emitter.onNext(fxObservable.getValue());
} else {
    emitter.onNext(nullSentinel);
}

Add a factory for invalidations of any FX observable.
I think propagating the FX observable instead of some constant is of better value.

public static Observable<javafx.beans.Observable> invalidationsOf(final javafx.beans.Observable fxObservable) {
    return ObservableSource.fromInvalidations(fxObservable);
}

public class ObservableSource {
	private ObservableSource() {}

	public static Observable<javafx.beans.Observable> fromInvalidations(javafx.beans.Observable fxObservable) {
		return Observable.create((ObservableEmitter<javafx.beans.Observable> emitter) -> {
			final InvalidationListener listener = emitter::onNext;
			fxObservable.addListener(listener);
			emitter.setDisposable(JavaFxSubscriptions.unsubscribeInEventDispatchThread(() -> fxObservable.removeListener(listener)));
		});
	}
}

Hey, I got java.lang.IllegalStateException when using Schedulers.newThread() or Schedulers.computation(), and when using JavaFxScheduler.platform() got no exception but the process is not asynchronously executed, I have to wait for the process to finish so i can access the UI. Any information, please?

I was learning about concurrency from your book (btw it's great) and ran into this.
When creating an Observable using JavaFxObservable factory methods and following it with subscribeOn(Scheduler) the Scheduler should be suppressed by default JavaFxScheduler.platform().
It is exectly what happens when using JavaFxObservable.actionEventsOf() but there is strange behaviour using it with JavaFxObservable.additionsOf().
It works fine (both subscribers receive their emisions) when new list element is added from inside button click handler but adding element straight from the main method causes lack of emissions to Observer[2].

import io.reactivex.Observable;
import io.reactivex.rxjavafx.observables.JavaFxObservable;
import io.reactivex.schedulers.Schedulers;
import javafx.application.Application;
import javafx.collections.FXCollections;
import javafx.collections.ObservableList;
import javafx.scene.Scene;
import javafx.scene.control.Button;
import javafx.stage.Stage;

public class SubscribeOnTestApp extends Application {

    public static void main(String[] args) {
        launch(args);
    }
    
    @Override
    public void start(Stage stage) throws Exception {
        
        Button button = new Button("Click me!");
        stage.setScene(new Scene(button));
        stage.show();
        
        ObservableList<String> strings = FXCollections.observableArrayList();
        
        JavaFxObservable.actionEventsOf(button)
        .subscribeOn(Schedulers.newThread())
        .subscribe(actionEvent -> {
            System.out.println("[" + Thread.currentThread().getName() + "] button actionEvent");
            strings.add("item");
        });
        
        Observable<String> observable = JavaFxObservable.additionsOf(strings);
        
        observable
        .subscribe(next -> System.out.println("[" + Thread.currentThread().getName() + "] Observer[1] next=" + next));
        
        observable
        .subscribeOn(Schedulers.newThread())
        .subscribe(next -> System.out.println("[" + Thread.currentThread().getName() + "] Observer[2] next=" + next));
        
        strings.add("initialItem");
    }
}

When using JavaFxObservable.emitOnChanged outside of the running application (specifically - to unit-test my view models with JUnit) I've experienced the unexpected "Toolkit not initilized exception" on the first emission.
The following sample code throws the exception:

    private ListProperty<String> listProperty = new SimpleListProperty<>(FXCollections.observableArrayList());

    @Test
    public void test() {
        JavaFxObservable.emitOnChanged(listProperty)
                //.subscribeOn(Schedulers.io())
                .subscribe(list -> {
                    Observable.fromIterable(list).subscribe(s -> System.out.println(s));
                });


        listProperty.add("list");
    }

The walkaround for the issue is to subscribe on a specific non-JavaFx scheduler (see the commented line).

I was inspired by Kotlin and the TornadoFX project, and believe this library should be represented as extension functions for Kotlin. This is why I started the RxKotlinFX library to represent the RxJavaFX static factories as extension functions.

But using Kotlin's extension features, it can go further than RxJavaFX is able to. It will augment existing JavaFX component functionality with helpful extension functions where it makes sense (for example, emitting TableView selected column and row indexes as Observables through extension functions).

If you are interested, please check out this project and provide input over the next week. After that I will look to get a first release on Maven Central.

https://github.com/thomasnield/RxKotlinFX

I won't make these changes right now. Just documenting them.

* Add vararg add() and remove() methods
* Add clear() method to remove all Observables
* Back with ObservableSet instead of ObservableList to prevent duplicates
* Add getBackingList() to return unmodifiable ObservableList<Observable<T>> for currently tracked Observables.

I'll need to do some exploring and see if RxJavaFX can support animation a little better.

http://www.java2s.com/Tutorials/Java/JavaFX/1010__JavaFX_Timeline_Animation.htm

RxJavaFx has been extracted from RxJava prior to 1.0 as per ReactiveX/RxJava#589

I have done this with a simply copy/paste/commit for now. Commit history remains in the 0.20.x branch of RxJava: https://github.com/ReactiveX/RxJava/tree/0.20.x

If anyone wishes to replace the code in this project with files that retain commit history, go ahead and make the changes.

Over the coming weeks the build and release process will be updated.

This project does not need to release immediately as 1.0. It can continue with 0.x releases until it makes sense to hit 1.0. Its lifecycle is now decoupled from RxJava.

Hi,

i have a quit complex application which has a ObservableList inside the data model.
Many ui controls and other parts of the business logic are bound to the list to update when changes are made.
Until now all changes of the list are triggered inside the JavaFX application thread. Which was good because otherwise i get an exception that it is the wrong thread when a bound control of the ui wants to reflect the changes.

Is it possible to use RXJavaFX to wrap the list and create another binding which notifies all these observer on the JavaFx Thread. It is important not to change the code parts where the original ObserableList is modified.

I thought about something like that:

1. Create an rx obserable from the change events of the original list.
2. Use oberserveOn to notify observers on the specific thread.
3. Create an javafx oberserable list from this rx obersable which will be used to bind the ui controls.

I have problems with point 3. how is it possible to create an javafx obersablelist from this obserable?

Thanks and best regards