ReactFX is an exploration of (functional) reactive programming techniques for JavaFX. These techniques usually result in more concise code, less side effects and less inversion of control, all of which improve the readability of code.

Initial inspiration came from the Principles of Reactive Programming course and the RxJava library. There are, however, important differences from RxJava.

Use reactfx tag on StackOverflow to ask specific questions. For more general discussions about the design of ReactFX and reactive programming for JavaFX, use the reactfx-dev mailing list.

An EventStream emits values (events). You can subscribe to an event stream to get notified each time a value is emitted.

interface EventStream<T> {
    Subscription subscribe(Consumer<T> consumer);
}

Example:

EventStream<T> eventStream = ...;
eventStream.subscribe(event -> System.out.println(event));

To stop receiving notifications, you use the Subscription returned from the subscribe method to unsubscribe:

Subscription subscription = eventStream.subscribe(event -> System.out.println(event));
// ...
subscription.unsubscribe();

Note that you need only the instance of Subscription to stop previously requested notifications. Compare this to JavaFX listeners/event handlers, where you need to keep both the listener/handler and the object you are listening to to be able to unregister the listener/handler.

Multi-valued streams compensate for the lack of language support for tuples in Java. ReactFX has convenience classes for 2- and 3-valued streams, namely BiEventStream and TriEventStream. This allows you to write

BiEventStream<A, B> eventStream = ...;
eventStream.subscribe((a, b) -> f(a, b));

instead of

EventStream<Tuple2<A, B>> eventStream = ...;
eventStream.subscribe(tuple -> f(tuple._1, tuple._2));

JavaFX has a representation of a time-varying value, namely ObservableValue. ObservableValue holds a value at any point in time. This value can be requested with getValue().

Events, on the other hand, are ephemeral—they come and go. You can only be notified of an event when it occurs;—it does not make sense to ask the event stream about the "current event".

JavaFX has means to compose observable values to form new observable values, either using the fluent API (methods of ObservableValue subclasses), or using the Bindings helper class. Some useful compositions of observable values are also provided by the EasyBind library.

JavaFX, however, does not have a nice way to compose streams of events. The user is left with event handlers/listeners, which are not composable and inherently side-effectful. EventStreams try to fill this gap.

Although it has no notion of an event stream, there are many event streams already hiding in JavaFX. ReactFX provides adapter methods to materialize them as EventStream instances.

Every Node is capable of emitting various types of events. We can obtain an event stream for each event type:

EventStream<MouseEvent> clicks = EventStreams.eventsOf(node, MouseEvent.MOUSE_CLICKED);
clicks.subscribe(click -> System.out.println("Click!"));

Every ObservableValue (e.g. property, binding) emits invalidations and changes. We can obtain the respective event streams:

ObservableValue<T> observable = ...;
EventStream<?> invalidations = EventStreams.invalidationsOf(observable);
EventStream<Change<T>> changes = EventStreams.changesOf(observable);
EventStream<T> values = EventStreams.valuesOf(observable);
EventStream<T> nonNullValues = EventStreams.nonNullValuesOf(observable);

The values stream above emits the new value every time the value changes. As opposed to the changes stream above, it avoids creating a Change instance in case we're not interested in the old value.

EventSource is an event stream that emits precisely what you push into it.

EventSource<Integer> numbers = new EventSource<>();
numbers.subscribe(i -> System.out.println(i));
numbers.push(7); // prints "7"

Fun begins with combining streams into new streams.

EventStream<MouseEvent> clicks = EventStreams.eventsOf(node, MouseEvent.MOUSE_CLICKED);
EventStream<MouseEvent> leftClicks = clicks.filter(click -> click.getButton() == MouseButton.PRIMARY);

EventStream<KeyEvent> keysTyped = EventStreams.eventsOf(node, KeyEvent.KEY_TYPED);
EventStream<String> charsTyped = keysTyped.map(keyEvt -> keyEvt.getCharacter());

EventStream<T> stream1 = ...;
EventStream<T> stream2 = ...;
EventStream<T> merged = EventStreams.merge(stream1, stream2);

EventStream<Double> widths = ...;
EventStream<Double> heights = ...;
EventStream<Double> areas = EventStreams.combine(widths, heights).map((w, h) -> w * h);

areas emits a combined value every time either widths or heights emit a value, but only after both widths and heights had emitted at least once.

EventStream<Double> widths = ...;
EventStream<Double> heights = ...;
EventStream<Double> areas = EventStreams.zip(widths, heights).map((w, h) -> w * h);

areas emits a combined value every time both widths and heights emit a value. zip expects all input streams to emit values at the same frequency. In the above example, it would be an IllegalStateException if widths emitted twice while heights did not emit at all.

Accumulates events emitted in close temporal succession into one.

EventSource<Integer> source = new EventSource<>();
EventStream<Integer> accum = source.reduceSuccessions((a, b) -> a   b, Duration.ofMillis(200));

source.push(1);
source.push(2);
// wait 150ms
source.push(3);
// wait 150ms
source.push(4);
// wait 250ms
source.push(5);
// wait 250ms

In the above example, an event that is emitted no later than 200ms after the previous one is accumulated (added) to the previous one. accum emits these values: 10, 5.

See the JavaDoc for more stream combinators.

All the adapters and combinators above subscribe lazily to their inputs - they don't subscribe to their inputs until they themselves have at least one subscriber. When the last subscriber unsubscribes, they unsubscribe from the inputs as well. This behavior has two benefits:

1. unnecessary computation is avoided;
2. composite stream's inputs don't prevent it from being garbage collected (no weak listeners needed).

Notice the difference to composed bindings. Bindings have to keep listening to their inputs all the time, because you can ask for the binding's current value (Binding.getValue()) any time. There is no such thing as the current value (event) of an event stream. This fact allows to automatically disconnect from the inputs when there are no subscribers.

Every event stream can be converted to a Binding that reflects the most recent event emitted from the stream.

EventStream<T> stream = ...;
T initial = ...;
Binding<T> binding = stream.toBinding(initial);

initial is used as the value of binding until stream emits the first value.

binding maintains an active subscription to stream until its dispose() method is called.

SuspendableEventStream is an event stream whose event emission can be temporarily suspended. There are several types of suspendable event streams that differ in what events, if any, are emitted when their emission is resumed.

When a suppressible stream is suspended, all events that would normally be emitted during this period are lost.

EventSource<Integer> src = new EventSource<>();
SuspendableEventStream<Integer> stream = src.suppressible();
stream.subscribe(i -> System.out.println(i));
stream.suspendWhile(() -> {
    src.push(1); // nothing is printed, 1 is never emitted from stream
});

When a pausable stream is suspended, events that would normally be emitted are buffered and emitted when event emission is resumed.

EventSource<Integer> src = new EventSource<>();
SuspendableEventStream<Integer> stream = src.pausable();
stream.subscribe(i -> System.out.println(i));
stream.suspendWhile(() -> {
    src.push(2);
    src.push(3);
    // nothing has been printed so far
});
// now "2" and "3" get printed

When a forgetful stream is suspended, only the latest event that would normally be emitted is remembered. This event is emitted when event emission is resumed.

EventSource<Integer> src = new EventSource<>();
SuspendableEventStream<Integer> stream = src.forgetful();
stream.subscribe(i -> System.out.println(i));
stream.suspendWhile(() -> {
    src.push(4);
    src.push(5);
    // nothing has been printed so far
});
// now "5" gets printed

When a reducible stream is suspended, it keeps reducing the incoming events together. The result of reduction is emitted when event emission is resumed.

EventSource<Integer> src = new EventSource<>();
SuspendableEventStream<Integer> stream = src.reducible((a, b) -> a   b);
stream.subscribe(i -> System.out.println(i));
stream.suspendWhile(() -> {
    src.push(6);
    src.push(7);
    src.push(8);
    // nothing has been printed so far
});
// now "21" gets printed

Note that forgetful() is equivalent to reducible((a, b) -> b).

When an accumulative stream is suspended, it keeps accumulating the incoming events into a cumulative value (accumulator), which may be of a different type than the events. When event emission is resumed, the accumulated value is deconstructed into a sequence of events that are emitted from the stream. This is a generalization of all previous suspendable streams.

reducible(reduction) can be modeled like this:

accumulative(t -> t, reduction, t -> Collections.singletonList(t))

suppressible() can be modeled like this:

accumulative(t -> (Void) null, (a, t) -> a, a -> Collections.emptyList())

pausable() can be modeled like this:

accumulative(ArrayList<T>::new, (l, t) -> { l.add(t); return l; }, l -> l)

InhiBeans are extensions of bindings and properties from javafx.beans.* that help prevent redundant invalidations and recalculations.

See InhiBeans wiki page for details.

Indicator is an observable boolean value that can be turned on temporarily.

Indicator workBeingDone = new Indicator();
Runnable work = ...;
workBeingDone.onWhile(work);

A useful use case for indicator is to signal when a component is changing state.

Consider a rectangle that needs to be repainted every time its width or height changes.

interface Rectangle {
    ObservableDoubleValue widthProperty();
    ObservableDoubleValue heightProperty();
    void setWidth(double width);
    void setHeight(double height);
}

Rectangle rect = ...;
rect.widthProperty().addListener(w -> repaint());
rect.heightProperty().addListener(h -> repaint());

rect.setWidth(20.0); // repaint #1
rect.setHeight(40.0); // repaint #2

Using indicator and stream combinators we can reduce the number of repaints in the above example to 1.

interface Rectangle {
    ObservableDoubleValue widthProperty();
    ObservableDoubleValue heightProperty();
    Indicator beingUpdatedProperty();
    
    // put implementation of setWidth() and setHeight() inside
    // beingUpdatedProperty().onWhile(/* implementation */);
    void setWidth(double width);
    void setHeight(double height);
}

Rectangle rect = ...;
EventStream<?> widthInvalidations = EventStreams.invalidationsOf(rect.widthProperty());
EventStream<?> heightInvalidations = EventStreams.invalidationsOf(rect.heightProperty());
EventStream<?> needsRepaint = EventStreams.merge(widthInvalidations, heightInvalidations);
EventStream<?> doneUpdating = beingUpdatedProperty().offs();
EventStream<?> repaintImpulse = needsRepaint.emitOn(doneUpdating);
repaintImpulse.subscribe(i -> repaint());

rect.beingUpdatedProperty().onWhile(() -> {
    rect.setWidth(20.0);
    rect.setHeight(40.0);
});
// just 1 repaint takes place now

ReactFX has a mechanism to handle errors encountered by event streams. You can read more about this mechanism on the Error Handling wiki page.

Current stable release is 1.4.1.

dependencies {
    compile group: 'org.reactfx', name: 'reactfx', version: '1.4.1'
}

libraryDependencies  = "org.reactfx" % "reactfx" % "1.4.1"

Download the JAR file and place it on your classpath.

Current milestone release is 2.0-M5.

dependencies {
    compile group: 'org.reactfx', name: 'reactfx', version: '2.0-M5'
}

libraryDependencies  = "org.reactfx" % "reactfx" % "2.0-M5"

Download the JAR file and place it on your classpath.

Snapshot releases are deployed to Sonatype snapshot repository.

repositories {
    maven {
        url 'https://oss.sonatype.org/content/repositories/snapshots/' 
    }
}

dependencies {
    compile group: 'org.reactfx', name: 'reactfx', version: '2.0-SNAPSHOT'
}

resolvers  = "Sonatype OSS Snapshots" at "https://oss.sonatype.org/content/repositories/snapshots"

libraryDependencies  = "org.reactfx" % "reactfx" % "2.0-SNAPSHOT"

Download the latest JAR file and place it on your classpath.

API 1.4.1 (Javadoc)
API 2.0-M5 (Javadoc)

BSD 2-Clause License