Going Reactive: An Architectural Journey

Since its inception, the Android platform has remained challenging to work with when building large event-driven applications in a consistent and reliable way. SoundCloud was among the earliest companies to embrace FRP as a vehicle to tackle this complexity. In this talk at GOTO Conference CPH 2015 Matthias gives an overview of what SoundCloud’s Android application architecture has evolved into, and how the use of RxJava for dealing with asynchronous streams of events has impacted its design, from data sources to business logic and presentation.

---

Introduction (0:00)

I want to take a retrospective look at the way we have been using reactive programming, specifically RxJava on Android, at SoundCloud for the past few years. I’ve talked about RxJava before, and there is plenty of material now on the web you can look at if you are unfamiliar with the library.

I want to focus less on the library itself and what gaps it closes on the Android framework, and focus more on how we use it in our application, and how it has impacted the way we write software for Android applications, i.e. the patterns that emerge and the problems we run into. Hopefully, I can give you some insight into how we solved these problems. Bear in mind that this will be my own opinion, but I hope it’s still going to be valuable, even if you haven’t used RxJava before.

I actually had to remind myself how long we had been using RxJava, so I dug through the archives and I found a commit from March 2013. We’ve been using it for about two and a half years now, starting with a pre-release version. It has been a wild ride since then. It was before API stability, and long before the main release was out. We bumped into a few issues in the beginning, of course, but in hindsight it has had a massive impact on our application and I think everyone was super happy with the choice.

That said, what I want to present here is a walk down the stack in our application, highlighting things that I thought were interesting, talking about in terms of how we split up our code.

Application Composition (2:14)

How do we compose our application, in terms of layers and what constitutes a layer? In 2013, we were following a pretty standard, run-of-the-mill, layered approach to our application architecture. So you would have, at the very top, your presentation objects like views, and your Android fragments and whatnot. You would have your business objects somewhere in the middle, acting as mediators, delivering data from the data layer up into the presentation layer. Finally, at the very bottom, we would have our data objects which would talk to the service API, talk to local storage, and so forth. Nothing terribly interesting there.

By the end of that year, mostly in the back end at SoundCloud, we came to realize that this kind of platform-driven approach we were taking no longer worked well. We were all focused around a specific technology: we had Android teams, we had an iOS team, we had a web team looking after the platform as well. This didn’t align well with the emphasis we wanted to put on particular features that drive business value.

We actually moved to a model where we would split up our organization around features rather than around technology. The purpose of this is really to drive our software into a direction that would more accurately resemble what we want to deliver in terms of value. We had started doing this in the backend already.

It is a transition, so it took awhile for us to go thorugh. Mobile was last in this transition, so we were still a platform team until a few months ago. However, we realized that going forward, we would have to work towards more of a matrix organization.

In a matrix organization, you still want to have separation between your layers: you still want your presentation objects separate from data. But we also wanted to have cleaner vertical lines in the application. If you think about SoundCloud’s discovery features, like exploring new content or a search, in contrast to things like our timeline features which are more like Twitter, we wanted to separate our code along those lines as well.

How Do You Communicate Between Layers? (5:04)

An interesting question that comes up is: How do you communicate? You need to communicate between the layers, in this vertical direction. For instance, how do you transport data from the data layer through your business logic into the UI at the end of the day? Also, now that we have introduced another axis, how do you communicate between features?

You want a particular element of the feature matrix to be cohesive and kind of decoupled from the rest of the system, so that you don’t end up with a “big ball of mud” as it’s sometimes called, where everything knows about everything else.

I mention this because I think this is where RxJava really helped us a lot. There are a number of mechanisms in Android to have components communicate in your application. Think of intents, for instance. Intents are a very common way to communicate. Or handler message, for instance. Often, it’s just callbacks between objects that you have, but it’s a bit of a mess and it’s difficult to get right. Especially if you think about the fact that these edges where we transition between features, or where we transition between layers, often involve concurrency. We have to jump off the main UI thread to do something expensive, like fetching a JSON response. Then the question comes up, “What is a good approach to handle with these without solving the same problem over and over again?”

The Sound Stream (6:38)

Going forward, I want to show you how we’ve worked on this problem in our Sound Stream using RxJava. The Sound Stream is a good example because it’s currently our landing screen, so it’s one of the most prominent screens in the application. You see it when you first open it, and for me this is such a typical screen to implement. At the end of the day, it’s just a list of stuff. If you have been writing mobile applications, I’m pretty sure you’ve written screens like this before, which are lists of stuff. Whether it’s cards, or grids of things or whatever, it’s the same problem.

Another interesting thing about the Sound Stream is that it looks simple. It’s just a list of things. However, there is actually a fair amount of complexity in there. This is basically an endless stream of items. If you follow people, there will be new content on a minute-by-minute basis, or hourly basis. This could grow very long, so you need to start thinking about lazy data deliveries, such as paging. You scroll to the bottom of the page, and then you want to lazily fetch new content in. There are things such as pull-to-refresh, a very common paradigm for a user to refresh content. There are also things such as error states: what if it goes wrong? If you fail loading the next page, you have to think about all these error views that you need to manage now. Additionally, there might be retry logic, if you want to retry fetching a page because it failed loading.

I’m pretty sure I’m not the only one who has written some pretty nasty code in the past try to make all this stuff come together, but RxJava was really helpful in making this a little shinier. ✨

Slicing Up the Sound Stream (8:46)

If we take a closer look at what constitutes this page, there are certain types of objects that we find at the presentation level, the business logic level, and the data level.

At the very top, we have our presentation objects. The first thing to mention are Android views and fragments. We use MVP in our application, so our view is actually very lightweight. Most prominently, we have a screen presenter at the very top which is responsible for steering the presentation logic and talking to our business logic as well. It will do such things as listening for Android LifeCycle events, fetch them further down the stack as well.

Next, we have what we call our feature operations. If you’re familiar with a clean architecture model, these are our “use case classes”, sometimes called interactors, and they contain all the business logic. They are basically responsible for wiring up multiple data sources, and looking after things such as scheduling as well. Typically, if you want to do anything expensive, you want to jump off the main UI thread. I don’t think the presentation layer should be concerned with such things, so we do this on the features operations layer, in addition to whatever business rules you might have for validation of models.

We establish the link between this through RxJava, and to some degree also RxAndroid, although I have to say that we use very little of the RxAndroid library. Mostly, we use the handler thread scheduler which allows us to dispatch callbacks on the Android main thread. We mostly use it downstream where data propagates from the server, or the data source in general to the presentation layer.

At the bottom would be our data layer, which contains all the data relevant to a particular feature. For Sound Stream in particular, this would be like a local database into which we sync content regularly, and of course the service API from which we originally fetch this data. There can be multiple of those involved to backup a certain feature.

One thing I want to highlight is that there is a single edge from the business logic into the UI. I think this is really important because we want the UI to be really dumb. It should not really have to juggle a lot of different objects to pull data from various sources, and then maybe execute logic in between. That would make our views very complicated, views are difficult to test. I think this dumb view approach is a healthy one, because you can push the logic further down the stack where it’s easier to test.

Views and Presenters (12:07)

class SoundStreamFragment extends LightCycleSupportFragment {

  @Inject @LightCycle SoundStreamPresenter presenter;
  
  public SoundStreamFragment() {
    setRetainInstance(true);
    ... 
  }
  ...
}

Here’s what a view looks like. We treat fragments as views, by the way, so they don’t really have a controller aspect to it. I know some people prefer MVC, so they would look at activities or fragments as controllers, but for us they are just basically the glue between the framework and the pieces of code that make up our app. We try to separate everything out, so this is of course a distilled fragment of code. Our fragments are really lean, there’s almost nothing in there, it’s just wiring.

When we inject a presenter who’s responsible for steering these views and talking to the business logic, we use a library called LightCycle, which is something we build in-house. It’s actually not built on RxJava, it’s one of those places where we used a different technology to address a problem. It basically dispatches LifeCycle events that we get from Android to any number of collaborators in the fragment. This could be a presenter, or multiple presenters, and you can also nest LightCycles. It could also do things like view tracking if you want to do something like this. We retain our fragments so they maintain a state that we inject, which is a cheap way to make sure that you don’t lose out on any data that you might be fetching in the background while the screen goes through a rotation change or something like that.

LightCycle works by having a companion object called a dispatcher in your fragment. The fragment can be the dispatcher itself as well, if you choose to do it that way. If the fragment receives a callback from the system such as onCreate, it would simply listen for this event and automatically forward it to any other LightCycles that you inject. The nice thing about this is if you have three LightCycles injected into a fragment, they don’t know each other. It’s very nice to have a cohesive bit of functionality put it in a single class, but it will expose the same API as a fragment. It works for activities as well, by the way.

We can test them in isolation, so any of these things can have their own unit tests. We just rely on LightCycle as kind of the mediator between the framework and our classes to do the wiring, and we just assume it works. Wee don’t write unit tests for fragments at all; we only write unit tests for LightCycles and any of the logic underneath.

class SoundStreamPresenter extends RecyclerViewPresenter<StreamItem> {
  ...
  @Override
  protected CollectionBinding<StreamItem> onBuildBinding(Bundle args) {
    return CollectionBinding.from(
      streamOperations.initialStreamItems())
        .withAdapter(adapter)
        .withPager(streamOperations.pagingFunction())
        .build();
  }
}

Let’s look closer at presenters. This is the screen presenter that we use to render the Sound Stream. I think it’s about 200 lines of code, so it’s not tiny but it’s also not massive. Essentially, what all these screens do if they render a list of things, is to go through a binding step. This is not quite your data binding you know from MVVM or from Android’s data binding library. This is something we did well before then. Binding is essentially the step that connects observable sequence that we build with RxJava to the UI, and it’s kind of like reusable infrastructure code. We do this on many screens.

The thing that sticks out out here is this CollectionBinding that we create. LightCycle would dispatch a callback to our presenter such as, “Oh, this view has just been created, please go and fetch data for this view.” It does so by talking to the stream operations, which is the use case class for this presenter. We can tell it, “Load x into y adapter using z builder syntax,” and we can also do such things as a paging function. That’s really all they do.

Of course, the other direction works as well. The presenter would listen for things such as click events, so if we have a button in your views, it could attach itself or an internal subscriber to a click event and then talk to the business logic to do whatever it’s supposed to do.

Paging (16:59)

Everyone has to deal with this paging problem. I went through a few implementations of this in the past, and I thought they were a bit difficult. If you think about this, it’s very much like one of these problems that sits between the layers. There is a lot of view-specific logic involved of course, but there’s also logic that needs to make a decision, given a particular page, about how it should arrive at the next page. This could be something like traversing an HTTP link to the next page of JSON elements that you fetch from the network, for instance.

These are things you don’t want to have in your views. You don’t want your views to know about how to construct queries, to fetch, to move a database cursor around to the next page of items or something. We were really keen on having something that A) does not create information leaks into the views about how we obtain this data, and B) that is reusable, because clearly the Sound Stream is not the only screen that we want to page.

Building with Rx, we created a pager class, which is like a slight abstraction over an Rx published subject. Basically, you can just think of it like an event channel where you can stick events in one end and you can subscribe other listeners at the other end so you can communicate in a decoupled way. The fragment would subscribe itself to the pager, and if there is already a page available, it would emit right away and deliver this page back into the subscriber through an Rx onNext call.

The second piece of logic that comes in here is if you scroll to the bottom of the page, it will detect, “Oh, it’s time to load the next page.” We have a reusable scroll listener which will issue a signal to the pager called next, which will instruct it to emit the next page. Now, how does it know to load the next page? Here comes the interesting part: an operator in Rx called switchOnNext. With this, given an existing stream of items, you can switch over multiple streams.

Basically, given an initial stream of items, using switchOnNext we can switch to the next page, but from the perspective of the caller that actually listens to the sequence, it looks like a single stream of items. If you want to think about this in more procedural ways or terms, it’s basically moving around pointers to lists. That’s kind of internally what happens. The pager would have a reference to what is the current stream of items, and a reference to what the next stream of items is. Using switchOnNext we can make sure we have it very easy without having to juggle any of the state ourselves. RxJava does all the heavy lifting for us here to emit pages of items into a subscriber.

In order to determine how to obtain the next page, it will ask a paging function. If you look at the code sample from the presenter, where it asks for this paging function, it asks the business logic, “Give me a paging function that tells me, given a certain page of items, how do I arrive at the next one?” In the case of the Sound Stream, it would be a database call, because we sync in the data in the background. So this is totally reusable. Basically, the only piece of information we need to provide to the pager is the paging function. We can just plug in this object in every single screen we need paging on then, and it just works.

Use Cases (21:03)

class SoundStreamOperations {

  Observable<List<StreamItem>> initialStreamItems() {
    return loadFirstPageOfStream()
      .zipWith(
        facebookInvites.loadWithPictures(),
        prependFacebookInvites())
          .subscribeOn(scheduler);
  }

  ...
}

Moving down the stacks a little bit, this would be in our use case classes. Particularly, this is the class that is responsible for making sure of the data delivery into the UI. We needed a page that is basically just a list of streamted items. I think what is interesting to see here is that this is the point where things come together.

First we have vertical communication, as you can see we return an observable sequence again, so it’s really just like a lazy description for the UI to subscribe to and eventually get data delivered. But, it’s also horizontal communication because, as you can see, we use this zipWith operator to reach out to a totally unrelated feature that we compose into this screen, which is a facebookInvites element. If you’re a new user of the application, this is basically a call-to-action to invite your friends to the platform. The nice thing about this is this particular class does not have to know anything about how this works, like where the data comes from, or what the conditions might be for when it is supposed to show up. None of this information is here, which is good, because this is the Sound Stream; it shouldn’t know about this. It’s like a separate feature. I think this is very symptomatic for a clean Rx architecture where. You have a single stream that you compose into somewhere else, and all the complexity is hidden beneath it. And scheduling, of course, we call subscribeOn here, which basically means that we have a scheduler that we inject which tells the sequence where it should execute.

Feature Data (23:05)

class SoundStreamStorage {

  Observable<PropertySet> streamItems(int limit) {
    Query query = Query.from("SoundStreamTable").limit(limit);
    return propellerRx.query(query).map(new StreamItemMapper());
  }

  ...
}

Of course, this is still not the end of the story, because we need to obtain our data somehow. Again, this use case class needs to reach out to some collaborator and ask it for data. We heavily sync data for the user that is locked into the application, so that we can make functionality available to the user as quickly as possible so they don’t have to look at spinners all the time. For the Sound Stream, this means then that in order to load a page of items, we go to the database. There, we have a collaborator which we just refer to as a storage class, which abstracts away the way we talk to the database basically. It’s just an observable sequence. From the perspective of the caller, they have no idea actually where it goes to, and we do not expose cursor objects or anything, so no database information leaks into a layer above.

What we employ internally is a library called Propeller, a light abstraction on top of an Android SQLite database, which lends itself better to reactive programming because it basically looks at tables as collections and iterables. It exposes iterable cursors so that you can use things like Guava or any kind of standard Java collection mechanisms to deal with your data from local storage, and because of the duality principle between observables and iterators, it’s very easy to turn them into Rx observable sequences.

We have this Rx companion object in Propeller which can take a query we write and it turn it into a observable sequence. In this case, it would emit a cursor and then we map this cursor or this row of the table to a particular item that we want to emit.

That basically concludes the vertical way we do things, and it covers all the layers. However, we still quite haven’t answered what we do horizontally, if we want to truly talk between features that maybe should not even know of each other at all.

Cross-Screen Messaging (25:28)

Cross-screen messaging is a really common case that I’m sure most of you have had to deal with. Coming back to the Sound Stream, we have a mixed list of tracks and playlists that other people post or repost, and these are interactive. If I click on a playlist here on the middle of the screen, it launches me into a totally different experience that I can engage with.

If I click the Like button that increments the counter, and then I press the Back button and I go back to the previous screen, the little heart icon in the counter will have increased immediately. So can pick up on the fact that I engaged with this object on a different screen.

The important takeaway here is that the Sound Stream is built by a different team. If you move to a feature team organization the Sound Stream is built by a different team than the screen that renders a playlist and we interact with a playlist. It’s very important, then, that you make sure that you do not couple these things too closely to each other.

Looking at the two screens side-by-side, bearing in mind that they are built by different teams, the problem basically becomes: How do we establish a communication channel between them?

Even without these screens knowing about each other, we can pass around data in an easy and efficient way. Android has ways to do that, like intents, but I’ve always thought that intents are terrible to work with because they force a couple odd complications on you as a developer such as parcelables. The only way you can pass around data in an intent is by putting things in a bundle, which is restricted by this parcelable API, so you need to be able to marshall this data even if no inter-process communication is involved. It’s strange because more often than not, you do not have inter-process communication; you just go from one screen to another, but you use the same vehicle (intents) to communicate this fact. I think is a very awkward design decision in Android.

Example: Screen-to-Screen Updates (27:56)

The way we solve it is by using Rx subjects. (We do still send an intent to open the activity, there’s no way around this, but we do not actually communicate via intents in terms of data passing.) If you know what an EventBus is, a subject is basically that. It’s like a single event queue where there are either subscribers or no subscribers.

Observable<PropertySet> toggleLike(Urn urn,
                                   boolean addLike) {
 return storeLikeCommand.toObservable(urn, addLike)
          .map(toChangeSet(targetUrn, addLike))
          .doOnNext(publishChangeSet);
}

This is what we did. Here’s the use case class for this particular screen, specifically the engagement bar we have there to trigger a Like. In order to target a Like, we need to persist the stack to the database, so we use command classes that know how to persist a certain change set to the database. We turn this into an Rx observable sequence, and if this is successful, we use the map operator to turn this into something that we can publish throughout the application.

Now, we want other screens who might be interested in this fact to know about it, so we use the map operator to turn this into a change set, and then we publish this change set to the application.

publishChangeSet: Action1(PropertySet)

@Override
public void call(PropertySet changeSet) {
  eventBus.publish(
    EventQueue.ENTITY_STATE_CHANGED, /* <-- RxSubject in disguise! */
    EntityStateChangedEvent.fromLike(changeSet)
  );
}

Publishing is basically an Rx action that gets the data passed in. We use this eventBus to publish. It’s really just an Rx subject in disguise; it’s nothing fancy, so if you remove all the fluff you would have an Rx subject. We found it a bit easier to have a light abstraction around that we called by a common name, EventBus, and that we inject as a singleton throughout the application. We have a testable for this as well, so it’s really easy to write unit tests for these things. We can fake events on the EventBus and make sure that other components in the application receive the right events, or subscribe or unsubscribe themselves.

SoundStreamPresenter

protected void onViewCreated(...) {
  eventBus.subscribe(
    EventQueue.ENTITY_STATE_CHANGED,
    new UpdateListSubscriber(adapter)
  );
}

On the receiving side, we are in the view layer. We need to make sure we receive this event so that the views can update themselves accordingly. Here we simply subscribe to the same event queue and update the adapter. If we get this change set, it’s as simple as updating the specific item in the list and rendering the list.

Implementation Patterns (31:41)

Life-Cycle Subscriptions

private CompositeSubscription viewLifeCycle;

protected void onViewCreated(...) {
  viewLifeCycle = new CompositeSubscription();
  viewLifeCycle.add(...);
  ...
}

protected void onDestroyView() {
  viewLifeCycle.unsubscribe();
}

How do you deal with connecting observable sequences to views, specifically of fragments and activities? We chose the simplest way, which is to just bind subscriptions to LifeCycle callbacks we get from Android. We would have something like subscriptions that follow the viewLifeCycle by subscribing sequences in a call to onViewCreated and unsubscribing when we get the onDestroyView callback. This leads to a bit more boilerplate than other solutions. There are more clever solutions out there that go all-in in terms of Rx, and then they basically emit every callback you get from Android as a separate Rx event to an observable. I found this gets very complicated. Instead, I think our way is a very straight-forward way of doing it.

Fast Path & Lazy Updates

Observable<Model> maybeCached() {
  return Observable.concat(cachedModel(), remoteModel()).first()
}

Fast Path or Lazy Update is concat first. You often want to do something like, emitting a cached value to the UI as quickly as possible so that your UI is responsive. But maybe you don’t. Maybe you need to go to the network after all and fetch it, because maybe it isn’t cached. Here, you can use the concat first approach, which is basically using the concatenate operator in RX. This appends two sequences to each other. Either the first one completes right away and doesn’t emit anything and it proceeds to the next sequence, or because of the first that we add at the end, if it does emit a value from the cache, then first we’ll simply take this singular item and unsubscribe so it would never even get to the remote call. This is a really common pattern that you can use to implement this kind of cache event fetch approach.

You can also emit first: then you have kind of like a lazy update, which means you emit something from your cache so the views immediately get the item. However, it will then also fetch an updated version of it and emit it again, so then you would see a number pop in, maybe with a second delay or so which updates itself. It’s kind of the reactive version of getOrElse.

Observable Transformers

Observable<Model> scheduledModel() {
  return Observable.create(...).compose(schedulingStrategy)
}

class HighPrioUiTask<T> extends Transformer<T, T> {
  public Observable<T> call(Observable<T> source) {
    return source
      .subscribeOn(Schedulers.HIGH_PRIO)
      .observeOn(AndroidSchedulers.mainThread())
  }
}

Observable transformers are a super nice things as well. They use the compose operator, and basically they allow you to take any number of composition steps in a sequence and turn them into a single one.

For example, instead of always saying subscribeOn this background thread and observeOn the Android main thread, you could take these two calls and declare a transformer that always performs them for you. Then, you can compose instances of this transformer, like a high-priority UI task, compose into your observables. A nice thing about this is it’s great for testing because if you unit test this scheduled model sequence that you expose, you can just swap this out in a unit test and run it synchronously on the current thread. You remove all concurrency in your unit test, which makes it super easy to deal with threading concerns.

Deferred Execution

Observable<Integer> intSequence() {
  return Observable.create((subscriber) -> {
    List<Integer> ints = computeListOfInts(); /* <-- expensive! */
    for (int n : ints) {
      subscriber.onNext(n);
      subscriber.onCompleted();
    }
  }
}

Observable<Integer> intSequence() {
  return Observable.defer(() -> {
    return Observable.from(computeListOfInts());
  }
}

We use deferred execution quite a bit, especially in Propeller, like the data layer. Let’s say you have a function that computes a list of integers, and it’s expensive. You want to put it on a background thread. Essentially, you want to delay this call until someone actually subscribes to look at this data, and it’s an iterable because it’s a list. Commonly, you would create your observable and then walk over the results of this computation and emit the results into your subscriber. However, there’s a number of problems with this code. First of all, it doesn’t handle errors very well, so the computation might throw, so you might want to catch errors and forward them to the subscriber. It also doesn’t deal with backpressure.

Instead, I recommend just reusing RxJava operators to do all this heavy lifting for you. There’s an operator called defer, which basically allows you to pass a lambda, which is an observable factory, that is only executed when someone subscribes to the sequence, not before. If someone comes and subscribes, you can pass this result of your computation directly into a factory method called from, which turns any Java collection into a observable sequence. It’s a single line, and it’s more resilient because you reuse all the infrastructure that RxJava gives you to deal with things like backpressure.

Common Pitfalls (36:40)

No-args subscribe

Observable.create(...).subscribe(/* no-args */)

–> OnErrorNotImplementedException

First of all, there’s a really tempting overload of the subscribe call on our sequence which takes no arguments. I do not use this because it has following the problem that you might not be aware of: it relies on the fact that you would never receive an error, it will re-throw a fatal exception and crash your application. Even if you want to do a fire-and-forget style call and you’re not actually interested in a result, make sure you always subscribe to sequences and have error handling logic present. Otherwise, this might bite you in production and suddenly crash the application.

ObserveOn: onError

Observable.create((subscriber) -> {
   subscriber.onNext(value); /* <-- gets dropped! */
   subscriber.onError(new Exception());
}.observeOn(mainThread()).subscribe(...)

–> onError cuts ahead of onNext

We have a subscriber here, or an observable, and we immediately emit the value. It doesn’t matter what it is, and we fail. We call onError with an exception, so what you would suspect is that the subscriber on the main thread would get the value first and then the error, but this is actually not what happens. This is because onNext gets dropped, because onError always cuts ahead of onNext. If observeOn recognizes that your sequence has failed, it doesn’t even bother emitting other items anymore because it wants subscribers to know as soon as possible that something went wrong, so that they can detach themselves. This can lead to surprising behavior, so watch out.

ObserveOn: Backpressure (38:30)

public void onStart() {
  request(RxRingBuffer.SIZE); /* 16! */
}

public void onNext(final T t) {
  ...
  if (!queue.offer(on.next(t))) {
    onError(new MissingBackpressureException());
    return;
  }
  schedule();
}

Backpressure is a big topic, so I’m not going to get into too much detail. However, it’s worth noting that it can cause problems.

This is code from the observeOn operator. I want to highlight this request call, because basically RxJava internally isn’t based on a push model. It’s based on what they call “reactive pull”, which basically tries to address the case where a subscriber is too slow to process all of the notifications that a very fast observable emits into the subscriber. Internally, it requests a specific amount of items, but if the observer doesn’t process them and ask for the next batch of items fast enough, it will fail at putting the new items that are produced into the queue. The surprising part of this is that on Android, this is 16. This is the internal buffer size, and clearly you can quickly run out of 16 items that are cached internally for a subscriber.Basically, it will throw a missing backpressure exception that will crash your application.

To deal with this, a pretty obvious way is you can simply take load off your target thread and make sure that you don’t overload, especially the Main Looper on Android. All the draw calls on Android go through the Main Looper. We made the mistake of being a bit defensive, and all of our callbacks we would schedule whatever it was on the main thread. It’s very easy, observeOn main thread, and then you’re done, because you don’t have to deal with concurrency anymore. Of course, this comes at a cost; if you overload the main thread and the Main Looper too much, at some point your sources will emit faster than Android is able process these messages, and you run into issues.

You could also buffer your sequences using things like toList, or the buffer operator so they can catch them and emit them in bigger chunks. There are built in backpressure operators too, and also system property, which is the hackier approach where you can override this value of 16 and bump up the internal buffer size.

Debugging (41:15)

What if something goes wrong? The good news is there is only a certain class of issues that I usually end up debugging, which is something messages not arriving in subscribers. Overall, it’s pretty solid, so it was a while until I found a bug in RxJava.

Observable.just(1, 2, 3)
    .map((n) -> {return Integer.toString(n);}
    .observeOn(AndroidSchedulers.mainThread());

Here is a little example sequence where we just emit a bunch of numbers and turn them into strings. There’s a mapping step involved, and then there’s observeOn. How can we debug this?

You can use the debugger, of course. With all this backpressure stuff internally, if you look at observeOn, it gets a bit difficult to understand though, so we wanted to make this a bit easier to work with.

We ended up writing a small library called Gandalf, which is basically an annotation drived byte code injector. It’s based on a project by Jake Wharton called “Hugo” which initially was meant to only inject logging into your application, but I think we just took it to its logical conclusions and added all sorts of annotations for it. We use it not only for logging, but also to inject all sorts of other instrumentation into the application. It is a byte code injector, for cross-cutting concerns, so it can generate byte code for you and inject it into all kind of different points in your application.

Then we have two annotations that we can use to output logging, @RxLogObservable and @RxLogSubscriber. It looks a little like this:

@RxLogObservable
Observable<String> createObservable() {
  return Observable.just(1, 2, 3)
    .map((n) -> {return Integer.toString(n);}
    .observeOn(mainThread());
}

@RxLogSubscriber
class StringSubscriber extends Subscriber<String> {}

The only thing we have to do is apply the Gandalf plug-in to our gradeable scripts, and then add this annotation, @RxLogObservable. We can do the same thing with a subscriber. What you will get it this:

[@Observable :: @InClass -> MainActivity :: @Method 
    -> createObservable()]
[@Observable#createObservable -> onSubscribe() ::  
    @SubscribeOn -> main]
[@Observable#createObservable -> onNext() -> 1]
[@Observable#createObservable -> onNext() -> 2]
[@Observable#createObservable -> onNext() -> 3]
[@Observable#createObservable -> onCompleted()]
[@Observable#createObservable -> onTerminate() :: 
    @Emitted -> 3 elements :: @Time -> 4 ms]
[@Observable#createObservable -> onUnsubscribe()]

It will simply print the sequence it observed to the command line. Here you can see the class MainActivity (this is not from our application, it’s just a dummy application), and this particular method called createObservable we subscribed on the main thread. First we received the value 1, then 2, then 3, and so forth, then we completed and terminated and it took x amount of time. You can also see there’s an unsubscribe call at the end.

This can be useful when you want to debug a sequence where somewhere along the line it dropped the message, and you’re not really sure where it happened. It’s nice to see, “Oh, it unsubscribed after x step.” It’s kind of a lead into where you can start looking for problems.

I will mention that there is an “official” component as well, called “RxJava Debug”, written by Netflix for the pre-release version of RxJava. It doesn’t look to be maintained though, and it’s actually based on a pre-release version of RxJava-core. It uses the plug-in API of RxJava and gives very noisy output.

Q&A (45:06)

Q: Could you compare the Rx Propeller to the SQL Brite by Square?

Matthias: We’re working on open-sourcing this project, by the way, but it fell behind schedule. We’ve been using it in production for about two years now, so it’s actually not a new project. It started out as a simple abstraction layer on top of Android SQLite, but we added this reactive functionality as our application grew with RxJava. It differs from SQL Brite in the sense that I actually think it’s more powerful and more complete as a fully-functional query builder. I think we covered pretty much all the cases that there are. What we do not do that SQL Brite does, is SQL Brite can register a query with an observable, a table observable, so whenever then the table changes, it will re-execute this query and communicate the result back to you. This is something that we don’t support right now, but we’re working on this. Other than that, it’s a totally unrelated effort we started working on awhile ago. Yahoo! actually put out a library as well for dealing with SQLite databases on Android. I think Propeller compares more to Yahoo!’s library.

Q: Do you plan to open-source LightCycle?

Matthias: Yes, we’re working on it! It’s a very simple library as well: it’s an annotation processor that allows you to hook in all these different classes through a single binding step. It works similar to Dagger in the sense that it generates source code, which is the binding companion basically to LightCycle, which is then instrumented through this binding step. We definitely want to open source this. Right now it’s a question of time and priorities.

If you have any other questions, find me on Twitter!