chromecast/base/reactive_java.md
[TOC]
Ever notice how Android methods often come in pairs? For every onCreate(),
there is an onDestroy(), for every onStart(), there is an onStop(). The
Android SDK commonly asks clients to register callbacks or extend base classes
that override pairs of methods that correspond to reversible changes in state.
State is often expressed in Java code as mutable variables. A state changes
when you assign a new value to the variable. If a variable can be one value at
some points and another value at other points, that means there are two states
that the variable can have. Everything that interacts with that variable needs
to work correctly for each state the variable can be in. For example, if an
instance variable is null in a class's constructor, and set to a value by some
method in that class, then every method that tries to call a method on that
variable needs to check whether the value of the variable is null before
handling it, because there is no guarantee which state the variable is in. You
will see a lot of code that looks like this when using this pattern for
representing state:
if (mFoo != null) {
mFoo.doSomething();
}
Additionally, mutator methods may need to check the state at runtime. For example, lazy initialization often looks like this:
if (mFoo == null) {
mFoo = new Foo(...);
}
This is not bad in and of itself, if the states are well-defined and it's easy to reason about the set of possible states by looking at the code. However, it very, very quickly becomes difficult to reason about states when there are any of the following:
Connection class that reads and writes data over a socket might disconnect
on a socket error from any read() or write() call. That means that
before any read() or write() call, the state must be checked. (Real Java
objects will often use Exceptions to short-circuit code blocks that enter
an exceptional state).initialize() method may
have methods that should only be called after initialize(), but the
compiler will not be able to check whether initialize() has been called.
This includes every method that has an assert statement on a mutable
instance variable.Consider this seemingly-simple task: you have two variables, mA and mB, each
of which could be either null or some real value of types A and B,
respectively. Furthermore, you want to initialize a new variable, mC of type
C, when the values of mA and mB are non-null, perhaps because the C
constructor takes an A and a B. Finally, if mA or mB becomes null again
after creating mC, reset mC to null. Also, you need to invoke a close()
method on mC whenever mC is reset. And if mA or mB changes while mC
exists, you need to call mC.close() and re-create mC with the new mA and
mB.
class MyClass {
private A mA = null;
private B mB = null;
private C mC = null;
public void setA(A a) {
mA = a;
recalculateC();
}
public void setB(B b) {
mB = b;
recalculateC();
}
private void recalculateC() {
// This method is always called when A or B changes, so if C exists, it
// must first be reset.
if (mC != null) {
mC.close();
mC = null;
}
if (mA != null && mB != null) {
mC = new C(mA, mB);
}
}
}
This may be fine on its own. But chances are, you will want to do something with
mC outside these methods. Every read will have to null-check, there's an
undocumented but critical requirement that every write to mA and mB is
done through setA() and setB(), and that recalculateC() is only called
when mA or mB is mutating, or else it will implicitly close.
These undocumented dependencies can only be protected against regression by testing. The compiler will not tell you if you made a mistake, so there must be unittests covering every possible state change. And in this case, we have two variables, each with two states that each have two possible state transitions, so 8 test cases are needed to cover everything. And this is the simplest case of composing two independent nullable mutable variables.
Meanwhile, if you use the Observables framework:
class MyClass {
private final Controller<A> mA = new Controller<>();
private final Controller<B> mb = new Controller<>();
{
mA.and(mB).subscribe(Both.adapt(C::new));
}
public void setA(A a) {
mA.set(a);
}
public void setB(B b) {
mB.set(b);
}
}
In the instance initializer, we set up a simple state machine with two
Controllers, which correspond to the mutable instance variables from the
previous example, and an event that observes the state of the Controllers and
invokes some logic on certain state changes.
The and() call composes the mA and mB, returning a new Observable that
is only activated when both sources are activated, and then deactivated if
either source is deactivated. mA and mB are activated or deactivated by
set() and reset() calls, respectively. The set() method deactivates the
state if the argument is null (the reset() method can also be used to
deactivate the state).
The subscribe() call makes it so that when the composed Observable formed by
mA.and(mB) is activated, a new C object will be created. When deactivated,
that C object's close() method will be called.
This really does cover all the cases we need. If multiple set*() calls are
made, an implicit reset() call will be made to the relevant Controller and
the C object associated with the first scope will be close()d.
What's better about this? First, notice we don't need mutable variables.
Both Controller objects are final, and are never null. We don't at any
point need to know what state the object is in inside any method
implementations; the Controllers and the pipeline set up by the and() and
subscribe() calls handle that for you.
Second, notice how the concerns of mutating and reacting to state are cleanly
separated. The mutator methods setA() and setB() are concerned only with
their respective Controllers, and the lifetime of the C object is managed in
one place in the instance initializer.
Finally, the relationship between the A, B, and C objects is
self-documenting. In the first approach with mutable variables, to
understand that the lifetime of C is associated with the intersection of the
lifetimes of A and B, one has to examine both setters and trace through
recalculateC() from the perspective of both of its call sites. In the second
approach, using Controllers, the relationship between A, B, and C is
expressed holistically in one line.
Think of an Observable as a container, with one very important feature: the
ability to register observers that will be notified when the contents of the
container change. The contents of the container at a given time is the state
of the Observable. The Observable base class alone does not expose any state
mutators, but it provides ways to register events that will be invoked when
state changes.
All state transitions of an Observable is either an activation or a
deactivation. An activation refers to putting some data into the container,
and a deactivation represents removing some data from the container. The data
that is contained in an Observable is thus called activation data.
To register events that should be invoked on these state transitions, we
subscribe observers. An Observer works by opening a Scope when an
activation occurs. If a deactivation occurs, the Scope opened by the
Observer will be closed. The name scope comes from how its lifetime is the
time that the activation data exists within the Observable, similar to how
variables are in scope when using
RAII in
languages like C++.
The one-to-one mapping of activations in an Observable to opened Scopes for
each subscribed Observer affords many useful properties. Foremost among these
is that the Scope can capture the data it is associated with as an immutable
variable: the activation data is the same when it is deactivated as when it was
activated. It also ensures that every deactivation requires an activation to
have occurred first (since you cannot close() an object that hasn't been
constructed). The implications of these properties will be explored further in
later sections.
To subscribe an Observer to an Observable, we call subscribe()
on the Observable. The subscribe() method takes a single argument, an
Observer, which has an open() method that returns a Scope. The
Observer's open() method is called when the Observable activates,
and the resulting Scope's close() method is called when the Observable
deactivates.
Lambda syntax can be used to easily construct Observer objects without
much boilerplate. For instance, if we want to simply log the transitions of an
Observable, we might do it like this:
void logStateTransitions(Observable<?> observable) {
observable.subscribe(x -> {
Log.d(TAG, "activated");
return () -> Log.d(TAG, "deactivated");
});
};
This is equivalent to the following, much more verbose version:
void logStateTransitions(Observable<?> observable) {
observable.subscribe(new Observer<Object>() {
@Override
public Scope create(Object x) {
Log.d(TAG, "activated");
return new Scope() {
@Override
public void close() {
Log.d(TAG, "deactivated");
}
};
}
});
}
As you can see, the version that uses lambdas is much more readable, as long as
you understand what an Observer is. It can help to think of the return () ->
as a separator between what happens when the data is activated and what happens
when the data is deactivated.
Either way, when logStateTransitions() is called on an Observable,
"activated" will be printed to the log when that Observable is activated,
and "deactivated" will be printed to the log when that Observable is
deactivated.
Though the above Observer does not use the x parameter, normally
Observer implementations will use the data that open() is given, so
that the behavior of the scope can depend on what data the Observable is
activated with.
Say we have an Observable<String> and we want to log the data it is activated
with:
void logStateTransitionsWithData(Observable<String> observable) {
observable.subscribe((String s) -> {
Log.d(TAG, "activated with data: %s", s);
return () -> Log.d(TAG, "deactivated");
});
}
The Observable interface does not provide any way to directly change the state
of the Observable. However, subclasses of Observable exist that do provide
mutators. The Controller class provides set() and reset().
Controllers are basically nullable, mutable variables that let you register
callbacks, through the Observable interface, that are run when the variable
changes.
With this in mind, the set() method on Controller is like setting a mutable
variable to a value. The reset() method is like setting the mutable variable
to null.
Here are some guarantees that Controllers provide:
set() is non-null, it enters the activated state.set() is null, it enters the deactivated state.reset() and set(null) enter the deactivated
state.reset() and set(null) do nothing.data1, set(data2) no-ops if
data1.equals(data2).equal() to the current
data, set() implicitly deactivates and reactivates with the new data.As corollaries, any registered Observer objects will:
open() methods invoked exactly once for each non-null
set() callScopes close()d exactly once when reset() or
set() to nullThis means this:
void helloGoodbye(Controller<String> message) {
message.set("hello");
message.set("goodbye");
message.set(null);
}
has the same behavior as this:
void helloGoodbye(Controller<String> message) {
message.set("hello");
message.reset();
message.set("goodbye");
message.reset();
}
Essentially, a Controller adapts two states with two possible actions each:
deactivated: set, resetactivated: set, resetinto a well-defined state machine with two states and two transitions:
deactivated: setactivated: reset... by dropping redundant reset() calls and inserting implicit reset() calls
between contiguous set() calls. This cuts the number of state transitions you
need to worry about in half!
Since Controllers implement Observable, you can register Observer
objects with subscribe() the same way as in the previous section, or inject a
Controller into any method that takes an Observable of the same parametric
type.
Remember that an Observable can be thought of as a container of data. In
this frame of mind, a Controller is a container of one or zero instances of
some data type.
The state of a Controller<T> is isomorphic to that of a nullable T variable
for all types T. But there are many cases where what we really want is a
representation of a boolean state: on or off, active or inactive, and don't need
any activation data.
For these cases, the org.chromium.chromecast.base.Unit class is used to denote
the fact that there is no data associated with the controller. The Unit type
is inspired by the type of the same name in many functional programming
languages, and represents a type with only one possible instance (aka
Singleton).
To make a controller without data, you can therefore use Controller<Unit>.
Since Unit means "no data," and there's only one way to get a Unit instance
(through the Unit.unit() method), this maps correctly to the concept of a
mutable boolean value.
Note that because the instance of Unit equals itself, calling set() on a
Controller<Unit> when it is already activated will no-op, making the behavior
of set(Unit.unit()) and reset() symmetric.
Example:
{
Controller<Unit> onOrOff = new Controller<>();
onOrOff.subscribe(x -> {
Log.d(TAG, "on");
return () -> Log.d(TAG, "off");
});
onOrOff.set(Unit.unit()); // Turns on.
onOrOff.set(Unit.unit()); // Does nothing because it's already on.
onOrOff.reset(); // Turns off.
onOrOff.reset(); // Does nothing because it's already off.
}
It's common for observers and APIs that refer to Observables with no data to
use Observable<?> in their interface. This is an easy way to make an
Observable's data "opaque" to observers, even when the data are not Unit.
{
Controller<Foo> foos = new Controller<>();
// Subscribers to opaqueFoos will get notified of state changes, but will
// not be able to access the data through the Foo interface.
Observable<?> opaqueFoos = foos;
// The observer can use methods like toString() that exist on all Objects,
// but cannot use Foo's API.
subscribe(opaqueFoos, x -> {
Log.d(TAG, "got Foo: %s", x.toString());
return () -> Log.d(TAG, "lost Foo: %s", x.toString());
});
}
and()In the motivating example, we wanted to invoke a callback once two independent
Observables have been activated.
Let's say we have a set of states:
Anot AWith the transitions not A <-> A.
And introduce another set of states:
Bnot BWith the transitions not B <-> B.
We can then describe the combinations of those state spaces with four states:
neitherjust Ajust BA and B...and eight transitions:
neither <-> just Aneither <-> just Bjust A <-> A and Bjust B <-> A and BThe Observable interface gives us a convenient way to get the (A and B)
state with a simple call:
public void logWhenBoth(Observable<A> observableA, Observable<B> observableB) {
observableA.and(observableB).subscribe(...);
}
The and() method takes the calling Observable and the given other
Observable and returns a new Observable that is only activated when both
input Observables are activated.
One way to think about it is that the and() call collapses the three states
(neither), (just A), and (just B) into one deactivated state, and
treats the state both as activated. For observers of the and()-composition
of states, one needs only worry about the two states, deactivated and
activated, same as with any other observer.
So how do we get the data in the subscribe() call? Let's say we want to log
when both Observables are activated:
public void logWhenBoth(Observable<A> observableA, Observable<B> observableB) {
observableA.and(observableB).subscribe((Both<A, B> data) -> {
A a = data.first;
B b = data.second;
Log.d(TAG, "both activated: a=" + a + ", b=" + b);
return () -> Log.d(TAG, "deactivated");
});
}
The type of the activation data for an and()-composed Observable is Both.
The Both type has two generic parameters, and first and second public
fields to access the data it encapsulates. It is essentially a trick to box
multiple values into a single object, so we only ever need Observer
interfaces that take a single argument.
Since the and() method returns an Observable, the result can itself call
and(), whose result can itself call and(), ad infinitum:
observableA.and(observableB).and(observableC).and(observableD)...
But beware, as the associated type of the Observable gets uglier and uglier:
a.and(b).and(c).and(d).subscribe((Both<Both<Both<A, B>, C>, D> data) -> {
A aData = data.first.first.first;
B bData = data.first.first.second;
C cData = data.first.second;
D dData = data.second;
Log.d(TAG, "a=%s, b=%s, c=%s, d=%s", aData, bData, cData, dData);
return () -> Log.d(TAG, "exit");
});
One one hand, it's kind of neat that you can do that at all. But it does come at
a cost to readability. The compiler can catch you if you mess up the
first.first.second chains if the types are different, but it is regrettable
that this much work is required to read the compound data. Some methods for
alleviating this are described below.
andThen()The composition of state spaces stateA.and(stateB) doesn't care if stateA or
stateB was activated first, so it can be activated by either activating
stateA and then stateB, or by activating stateB and then stateA.
This means the state (A and B), extracted by the and() method on
Observable, is too ambiguous for knowing the order of activation. If we want
to know whether A was activated before B, we must partition the state
(A and B) into (A and then B) and (B and then A). The time-dependent state
space for two boolean variables looks like this, with five states:
neitherjust Ajust BA and then BB and then A...and ten transitions:
neither <-> just Aneither <-> just Bjust A <-> A and then Bjust B <-> B and then AA and then B --> just BB and then A --> just ACalling stateA.andThen(stateB) returns an Observable representing the
(A and then B) state from above. The resulting Observable will only activate
on the transition between (just A) and (A and then B), and will not activate
on the transition between (just B) and (B and then A).
Sometimes you might want to only subscribe() to an Observable for a limited
time, for instance, until some other Observable is activated. So how do you
remove an observer?
The subscribe() method returns a Subscription, which, when close()d, will
unregister the Observer registered in the subscribe() call. To subscribe()
for a limited time, simply store the Subscription somewhere, and call
close() on it when you're done.
private final Observable<String> mMessages = ...;
private final List<String> mLog = ...;
private Subscription mSubscription = null;
public void startRecording() {
if (mObserver != null) stopRecording();
mObserver = mMessages.subscribe(Observers.onEnter(mLog::add));
}
public void stopRecording() {
if (mSubscription == null) return;
mSubscription.close();
}
... wait a minute, are those null-checks? And a mutable variable? I thought
this framework was supposed to get rid of those!
And indeed we can! Since mObserver is a Subscription, which is a kind of
Scope, that means we can use it in another subscribe() call!
private final Observable<String> mMessages = ...;
private final List<String> mLog = ...;
private final Controller<Unit> mRecordingState = ...;
{
// When mRecordingState is activated, an Observer is registered to
// mMessages.
mRecordingState.subscribe(x -> {
// When mRecordingState is deactivated, the Subscription is closed,
// so new messages will stop being added to the log.
return mMessages.subscribe(Observers.onEnter(mLog::add));
});
}
public void startRecording() {
mRecordingState.set(Unit.unit());
}
public void stopRecording() {
mRecordingState.reset();
}
Now we have removed the mutable variable and delegated all management of state
to Observables.
But wait, we could have done the same thing with and():
{
mRecordingState.and(mMessages).subscribe(Observers.onEnter(
(Both<Unit, String> data) -> mLog.add(data.second)));
}
But here we can see the drawbacks of that approach. We need to deconstruct the
Both object. Though the below section shows a way to circumvent that when only
using a single and() call, it gets much harder to work with longer chains of
and()-composed Observables.
Recall that deconstructing larger Both trees is ugly:
stateA.and(stateB).and(stateC).and(stateD).subscribe(data -> {
A a = data.first.first.first;
B b = data.first.first.second;
C c = data.first.second;
D d = data.second;
...
});
If we only care about registering a Scope for when all four Observables are
activated, then we can use nested subscribe() calls instead:
stateA.subscribe(a -> stateB.subscribe(b -> stateC.subscribe(c -> stateD.subscribe(d -> {
...
}))));
This is called subscription-currying, and is a useful alternative to and()
calls when registering Observer objects for the intersection of many
Observables.
To show why this works, let's simplify to just this:
stateA.subscribe(a -> stateB.subscribe(b -> ...));
Notice that:
subscribe to stateB until stateA is activated.stateB is already activated when an Observer is subscribed to it, the
observer will be notified of the data immediately.stateA is activated, the Observer subscribed to stateB will open
and close its scopes normally as stateB mutates.stateA is deactivated, the Subscription to stateB is closed. Closing
a Subscription also closes the Scopes from the Observer.In other words, the inner Observer is only opened if both stateA and
stateB are activated, and that Observer's Scope will be closed if either
stateA or stateB is deactivated. This is the same as the and() operator!
This even works for Observables that have multiple activations. Each
activation of stateA will produce a unique Subscription to stateB, so this
pattern can be compared to a nested for loop -- one that operates reactively
whenever the states update! More precisely, there will be a 1:1 mapping of the
Cartesian product of activations of stateA and stateB and Scopes from the
inner Observer subscribed to stateB.
One should use the and() operator when more operations like map() and
filter() are needed on the resulting Observable, but subscription-currying
can be used in some other situations as an alternative in situations where
dealing with Both objects becomes cumbersome. It is generally recommended to
prefer the and() operator if all else is equal, because it's easier to add
more operators to the pipeline later on if needed.
An alternative to using and() in situations that call for map() and
filter() on the result is to use flatMap()-currying:
stateA.flatMap(a -> stateB.flatMap(b -> Observable.just(new C(a, b))))
.filter(c -> isValid(c))
.subscribe(d -> ...);
The Observers class contains several helper methods to increase the
fluency and readability of common cases that Observer objects might be
used for.
Every Observer returns a Scope, but sometimes clients do not care about
when the state deactivates, only when it activates. It's possible to create a
Observer with lambda syntax to do the job like this:
{
observable.subscribe((String data) -> {
Log.d(TAG, "activated: data=" + data);
return () -> {};
});
}
The return () -> {}; statement in the lambda corresponds to having no
side-effects to handle the destructor, but this is not very readable.
To make intentions clearer, the onOpen() method can wrap any Consumer of
the activation data's type:
{
// Without data.
observable.subscribe(Observer.onOpen(x -> Log.d(TAG, "activated")));
// With data.
observable.subscribe(Observer.onOpen((String data) -> {
Log.d(TAG, "activated: data=" + data);
}));
}
Likewise, onClose() is used the same way to transform any Consumer of the
activation data's type into a Observer that only has side effects when the
Observable is deactivated.
When you use the and() method on Observable to create an Observable<Both>,
recall that the Observer passed to subscribe() must look like this:
{
observableA.and(observableB).subscribe((Both<A, B> data) -> {
A a = data.first;
B b = data.second;
Log.d(TAG, "on enter: a = " + a + "; b = " + b);
return () -> Log.d(TAG, "on exit: a = " + a + "; b = " + b);
});
}
Observers provides a helper method to turn any function that takes two
arguments and returns a Scope into a Observer<Both>, which deconstructs
the Both object for you and passes the constituent parts into the function.
Using Both.adapt(), we can rewrite the above like this:
{
observableA.and(observableB).subscribe(Both.adapt((A a, B b) -> {
Log.d(TAG, "on enter: a = " + a + "; b = " + b);
return () -> Log.d(TAG, "on exit: a = " + a + "; b = " + b);
}));
}
When using onEnter() or onExit(), which take Consumers of the data type,
it can be useful to use Both.adapt on a BiConsumer to turn it into a
Consumer<Both>.
{
// Before:
Observable<Both<A, B>> both = observableA.and(observableB);
both.subscribe(Observers.onEnter((Both<A, B> data) -> {
Log.d(TAG, "on enter: a = " + data.first + "; b = " + data.second);
}));
both.subscribe(Observers.onExit((Both<A, B> data) -> {
Log.d(TAG, "on exit: a = " + data.first + "; b = " + data.second);
}));
// After:
both.subscribe(Observers.onEnter(Both.adapt((A a, B b) -> {
Log.d(TAG, "on enter: a = " + a + "; b = " + b);
})));
both.subscribe(Observers.onExit(Both.adapt((A a, B b) -> {
Log.d(TAG, "on exit: a = " + a + "; b = " + b);
})));
}
The Both.adapt() helpers are also able to turn BiFunctions into
Function<Both> objects and BiPredicates into Predicate<Both> objects,
which makes them useful when using map() and filter() operators on
Observable<Both> objects.
{
both.map(Both.adapt((A a, B b) -> {
return new ThingBuilder().setA(a).setB(b).build();
}));
both.filter(Both.adapt((A a, B b) -> a.contains(b)));
}
There are numerous instances where one may want to take the activation data of
some Observable and use it to set the state of a Controller, and reset that
Controller when the Observable is deactivated. A shortcut to doing this
without having to instantiate any Controller is provided with the map()
method in the Observable interface.
For example, we might have an Activity that overrides onNewIntent(), and
extracts some data from the Intent it receives. We might want to register
observers on the extracted data rather than the Intent itself, as some work
needs to be done to unparcel the data we care about from the Intent.
public class MyActivity extends Activity {
private final Controller<Intent> mIntentState = new Controller<>();
{
Observable<Uri> uriState = mIntentState.map(Intent::getData);
Observable<String> instanceIdState = uriState.map(Uri::getPath);
...
}
public void onCreate() {
super.onCreate();
mIntentState.set(getIntent());
}
public void onNewIntent(Intent intent) {
super.onNewIntent(intent);
mIntentState.set(intent);
}
}
The map() method takes any function on the Observable's activation data and
creates a new Observable of the result of that function applied to the
original Observable's activation data. So the activation lifetime of
uriState and instanceIdState are the same as mIntentState in this example.
The instance initializer can then call subscribe() on uriState or
instanceIdState to register callbacks for when we get a new URI or instance
ID, and the process of extracting the URI from the Intent and the instance ID
from the Uri is delegated to methods with no side-effects.
If a function provided to a map() method returns null, then the resulting
Observable will be put in a deactivated state, even if the source Observable
is activated. This can be used to filter invalid data from Observables in
the pipeline:
{
mIntentState.map(Intent::getExtras)
.map((Bundle bundle) -> bundle.getString(INTENT_EXTRA_FOO))
.subscribe((String foo) -> ...);
}
The bundle.getString() call might return null if the source Intent does
not have the correct extra data field set. When this happens, the resulting
Observable simply does not activate, so the Observer registered in the
subscribe() call does not need to worry that foo might be null.
One may wish to construct an Observable that is only activated if some
predicate on some other Observable's activation data is true. This is easily
done using the filter() method on Observable.
This example will only log "Got FOO intent" if mIntentState was set() with
an Intent with action "org.my.app.action.FOO":
{
String ACTION_FOO = "org.my.app.action.FOO";
mIntentState.map(Intent::getAction)
.filter(ACTION_FOO::equals)
.subscribe(Observers.onEnter(action -> {
Log.d(TAG, "Got FOO intent");
}));
}
Since Observable<T>#filter() takes any Predicate<T>, which is a functional
interface whose method takes a T and returns a boolean, the parameter can be
an instance of a class that implements Predicate<T>:
class InRangePredicate implements Predicate<Integer> {
private final int mMin;
private final int mMax;
private InRangePredicate(int min, int max) {
mMin = min;
mMax = max;
}
@Override
public boolean test(Integer value) {
return mMin <= value && value <= mMax;
}
}
InRangePredicate inRange(int min, int max) {
return new InRangePredicate(min, max);
}
Controller<Integer> hasIntState = new Controller<>();
Observable<Integer> hasValidIntState = hasIntState.filter(inRange(0, 10));
... or a method reference for a method that takes the activation data and returns a boolean:
class Util {
static boolean inRange(int i) {
return 0 <= i && i <= 10;
}
}
Controller<Integer> hasIntState = new Controller<>();
Observable<Integer> hasValidIntState = hasIntState.filter(Util::inRange);
... or a lambda that takes the activation data and returns a boolean:
Controller<Integer> hasIntState = new Controller<>();
Observable<Integer> hasValidIntState =
hasIntState.filter(i -> 0 <= i && i <= 10);
Consider this code:
Controller<String> c = new Controller<>();
c.set("hi");
c.reset();
c.subscribe(Observers.onEnter(s -> Log.d(TAG, s)));
Will the callback registered in the subscribe() call get fired? It turns out
that it will not, since c is deactivated when subscribe() is called. But if
the subscribe() call is made before the set() call, then the callback is
fired.
Sometimes this is what you want, but it's best to avoid any ambiguity like this.
Generally, Observable methods like subscribe() should be called before any
Controller methods. A couple of things that one can do to help with this:
Controller objects in field initializers, not the constructor.subscribe(), and(), map(), etc.) in an instance
initializer. This is run before anything else when creating an instance,
including the constructor, and is the same regardless of which constructor
is being used. This also removes the potential of accidentally depending on
constructor parameters or mutable instance variables in the pipeline, which
can be dangerous compared to adapting them to Observables.Observables composed from other Observables
can usually be local variables rather than instance variables. This prevents
code outside the initializer from subscribe()ing these Observables after
the instance has been initialized.Controller mutator methods (set() or reset()) inside the
instance initializer. They may be called in the constructor or any instance
methods.Controllers can be separated by using a factory
function.What happens here?
Controller<Object> c = new Controller<>();
c.subscribe(x -> {
Log.d(TAG, "enter");
c.reset();
return () -> Log.d(TAG, "exit");
});
c.set("ding");
Here, we reset() the same Controller in an activation observer for that very
Controller!
This is in fact safe, though there should be few places you need to do something
like this. Currently, Controllers notify all observers synchronously on the
thread that set() or reset() was called in (so they are not thread safe),
but if an observer calls set() or reset() again while observers are still
being notified, the set() or reset() call gets queued and handled only after
all observers have been notified. This allows a deterministic and unastonishing
order of execution for the above example: the log will show "enter", followed
immediately by "exit".
We call this property re-entrancy safety. Controllers are re-entrant-safe.
What this guarantees is that all observers are notified of all changes, and any
observer that imposes its own change (including indirect changes) will not have
that change take effect until all observers are notified of the current change.
Note that if you set() a controller with a value that is never null inside
an activation handler, you will get an infinite loop.
Controller<Integer> c = new Controller<>();
c.subscribe(Observers.onEnter(x -> c.set(x + 1))); // Danger!
c.set(0); // Infinite loop!
Whenever the Controller is set with a value, the observing scope immediately
sets it with a new value, recurring infinitely.
It's possible to still be safe if you can guarantee that set() isn't called or
set(null) is called in some base case for all recursive stacks of activation
handlers, but if you do that, it's your job to solve the halting problem.
It is good practice to avoid calling set() or reset() on Controllers
inside Observer event handlers altogether, but there are many safe ways
that are useful.
One of the most important aspects of using Observables is that they are very
testable. Although Observers themselves are not pure-functional (i.e. they
tend to mutate program state), this is done in such a way that the mutations in
the form of state transitions in Observables are easy to track, and therefore
easy to test.
If you write a class that implements Observable or returns an Observable in
one of its methods, it's easy to test the events it emits by using the
ReactiveRecorder test utility module. This class, which is only allowed in
tests, provides a fluent interface for describing the expected output of an
Observable.
To use this in your tests, add //chromecast/base:cast_base_test_utils_java to
your JUnit test target's GN deps.
As an example, imagine we want to test a class called FlipFlop, which
implements Observable and changes from deactivated to activated every time its
flip method is called. The tests might look like this:
import org.chromium.chromecast.base.ReactiveRecorder;
... // other imports
public class FlipFlopTest {
@Test
public void testStartsDeactivated() {
FlipFlop f = new FlipFlop();
ReactiveRecorder recorder = ReactiveRecorder.record(f);
// No events should be emitted.
recorder.verify().end();
}
@Test
public void testFlipOnceActivatesObserver() {
FlipFlop f = new FlipFlop();
ReactiveRecorder recorder = ReactiveRecorder.record(f);
f.flip();
// A single activation should have been emitted.
recorder.verify().opened(Unit.unit()).end();
}
@Test
public void testFlipTwiceActivatesThenDeactivates() {
FlipFlop f = new FlipFlop();
ReactiveRecorder recorder = ReactiveRecorder.record(f);
f.flip();
f.flip();
// Expect an activation followed by a deactivation.
recorder.verify().opened(Unit.unit()).closed(Unit.unit()).end();
}
}
The ReactiveRecorder class works by calling subscribe() on the given
Observable and storing the activations and deactivations it observes in a
list. The verify() method opens a domain-specific language for performing
assertions on the activation data, using opened() and closed() to check
which data has been activated and deactivated. The transitions recorded must
occur in the same order as the opened() and closed() calls to pass the test.
The end() method asserts that no more transitions occurred.
You can test behaviors that should occur when closing a subscribe() scope by
calling recorder.unsubscribe(). For example, every Observable implementation
should close all existing Scopes emitted from an Observer when that
Observer's subscribe() scope is closed:
@Test
public void testUnsubscribeCloses() {
FlipFlop f = new FlipFlop();
ReactiveRecorder recorder = ReactiveRecorder.record(f);
f.flip();
// Clear the record; we don't care about the activation from flip().
recorder.reset();
recorder.unsubscribe();
// Unsubscribing should implicitly close the scope.
recorder.verify().closed(Unit.unit()).end();
}
Once a ReactiveRecorder unsubscribes, it will not get any new events from the
Observable it was recording.
Sometimes, particularly in long chains of and() or andThen() or map() or
filter(), you might get confused about the ultimate state of the system in
some circumstances. When confused about these complex Observables, it's often
helpful to use the debug() operator:
// Before
foos.andThen(bars).subscribe(...);
// After
foos.debug(msg -> Log.d(TAG, "foos: %s", msg))
.andThen(bars.debug(msg -> Log.d(TAG, "bars: %s", msg)))
.debug(msg -> Log.d(TAG, "foosAndThenBars: %s", msg))
.subscribe(...);
The debug() operator doesn't by itself log anything; you need to give it a
Consumer<String> that logs the debug messages it generates. This makes the
file/line number information in the log is more helpful, as it indicates which
file you're debugging, rather than the body of the debug() operator itself.
The messages in question will show when Observers subscribe and unsubscribe,
and whenever data are added or removed from the Observable. The debug()
operator will call the toString() method on all data to format messages about
state transitions.
Observables and Controllers are intended to succinctly adapt common Android
SDK method pairs, whether they're callbacks for entering and exiting a state, or
mutators to perform state changes, into a common pattern that better separates
concerns and is composable.
Every mutable, nullable variable is a variable that you constantly have to
null-check before using. A Controller can be used to refactor these variables
into a final Controller.
The important insight is that you tend to only read a variable when state changes, either after the variable itself is known to change, or when some other state changes.
First, let's consider an Activity that registers a BroadcastReceiver in
onStart() and unregisters the BroadcastReceiver in onStop().
We will ignore for now that Android tries to guarantee that pathological call
sequences like multiple onStart()s in a row or an onStop() before the first
onStart() will not occur, because these guarantees are not known to the Java
compiler and similar guarantees can't be relied on for all events.
class MyActivity extends Activity {
private BroadcastReceiver mReceiver = null;
@Override
public void onStart() {
super.onStart();
if (mReceiver != null)
unregisterReceiver(mReceiver);
mReceiver = new BroadcastReceiver(...);
registerReceiver(mReceiver);
}
@Override
public void onStop() {
if (mReceiver != null)
unregisterReceiver(mReceiver);
mReceiver = null;
super.onStop();
}
}
Without the assumption that Android will call onStart() and onStop() in
reasonable orders, we need to check the state of the mReceiver variable each
time before it is used. And making that assumption is prone to backfiring, as
it's a recipe for NullPointerExceptions, IllegalStateExceptions, and other
horrors in general practice.
Here's the refactored version that uses a Controller:
class MyActivity extends Activity {
private final Controller<Unit> mStartedState = new Controller<>();
{
mStartedState.subscribe(x -> {
final BroadcastReceiver receiver = new BroadcastReceiver(...);
registerReceiver(receiver);
return () -> unregisterReceiver(receiver);
});
}
@Override
public void onStart() {
super.onStart();
mStartedState.set(Unit.unit());
}
@Override
public void onStop() {
mStartedState.reset();
super.onStop();
}
}
The refactored version better separates concerns. BroadcastReceiver
registration and unregistration is handled in a small area of the code, rather
than spread throughout the class, and the BroadcastReceiver doesn't need to be
stored in a mutable variable. No code outside the scope in which the
BroadcastReceiver object is relevant can touch it, and the onStart() and
onStop() methods have no logic except toggling the Controller that
represents whether the Activity is started. Best of all, there are no null
checks, and no need for any.
Some methods run asynchronously and take a callback that is run when the work is
complete. We can set() Controllers in such callbacks to adapt this pattern
to Observables, which can be used to create asynchronous initialization
pipelines.
This example shows how one can link up the outputs of multiple asynchronous
functions that use the callback-passing style using Controllers, and
encapsulating the complicated setup into a single function that returns an
Observable.
public class AsyncExample {
private static final String TAG = "AsyncExample";
// Adapts the callback-style asynchronous Baz function to an Observable.
// Shows how a
public static Observable<Baz> createBazDefault() {
// Begin constructing a Foo.
Controller<Foo> fooState = new Controller<>();
// Set the fooState controller when created, reset on errors.
Foo.createAsync(fooState::set, fooState::reset);
// Bar requires Foo to initialize.
Controller<Bar> barState = new Controller<>();
fooState.subscribe((Foo foo) -> {
Bar.createAsync(foo, barState::set);
// If fooState is reset, then barState is also reset.
return barState::reset;
});
// Baz requires Foo and Bar to initialize.
Controller<Baz> bazState = new Controller<>();
fooState.and(barState).subscribe(Observers.both((Foo foo, Bar bar) -> {
Baz.createAsync(foo, bar, bazState::set);
// If fooState or barState is reset, then bazState is also reset.
return bazState::reset;
}));
return bazState;
}
public static void demo() {
Observable<Baz> bazState = createBazDefault();
bazState.subscribe(Observers.onEnter((Baz baz) -> {
// This runs when the full initialization pipeline is complete.
Log.d(TAG, "Baz created!");
}));
}
public static class Foo {
static void createAsync(Consumer<Foo> callback, Runnable onError) {...}
}
public static class Bar {
static void createAsync(Foo foo, Consumer<Bar> callback) {...}
}
public static class Baz {
static void createAsync(Foo foo, Bar bar, Consumer<Baz> callback) {...}
}
}
This way, Observables can be used similarly to Promises, where a callback
handling the underlying value can be registered before the underlying value is
available. But unlike Promises, Observables provide a way to also handle
teardowns, and to transitively tear down everything down stream when something
is torn down.