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(Observers.both(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 register scopes to track the state of 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); 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. }
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.
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 onEnter() method can wrap any Consumer of the activation data's type:
{ // Without data. observable.subscribe(Observers.onEnter(x -> Log.d(TAG, "activated"))); // With data. observable.subscribe(Observers.onEnter((String data) -> { Log.d(TAG, "activated: data=" + data); })); }
Likewise, onExit() 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 Observers.both(), we can rewrite the above like this:
{ observableA.and(observableB).subscribe(Observers.both((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 made. 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”.
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.
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.