In ReactFX, we work with event streams. It provides various operations on event streams such as filter
, map
, merge
, zip
, combine
, reduceSuccessions
. Of course these cannot cover all the use cases. In such a case, you can either extend LazilyBoundStream to implement your custom stream combinator, or you can use the state machine combinator.
A state machine has a state and multiple input streams. An event emitted by an input stream causes the state machine to update its state and/or to emit an event. We demonstrate ReactFX’s internal DSL (fluent API) for defining state machines on an example.
Example
We implement a stream combinator that takes a source stream and two control streams. An event from one control stream mutes the source stream, an event from the other control stream unmutes the source stream. The resulting stream emits the same events as the source stream, but only when not muted. We are going to represent the internal state as a boolean indicating whether the source stream is muted.
1
2
3
4
5
6
7
8
9
EventStream<T> source = ...;
EventStream<?> muteImpulse = ...;
EventStream<?> unmuteImpulse = ...;
EventStream<T> combined = StateMachine.init(false)
.on(muteImpulse).transition((wasMuted, i) -> true)
.on(unmuteImpulse).transition((wasMuted, i) -> false)
.on(source).emit((muted, t) -> muted ? Optional.empty() : Optional.of(t))
.toEventStream();
- On line 5 we set the initial state to
false
(not muted). - Line 6 says that when an event arrives from the
muteImpulse
stream, we set the internal state totrue
(muted), no matter what the previous state or the value of the impulse is. - Line 7 says that when an event arrives from the
unmuteImpulse
stream, we set the internal state tofalse
(not muted), no matter what the previous state or the value of the impulse is. - Line 8 states that when an event arrives from the
source
stream, it is either emitted or not, based on the internal state (muted or not). - Line 9 completes the creation of the state machine and the associated stream.