16. Creating Custom Reactive Extensions

Written by Alex Sullivan

After being introduced to RxJava and learning how to create tests, you have yet to see how to create wrappers using RxJava on top of frameworks created by Google or by third parties. Wrapping a Google or third party library component is instrumental in writing reactive applications, so you’ll be introduced to the concept in this chapter.

In this chapter you’ll create a reactive wrapper around an Android Widget, a request for a specific permission, and the process of getting location updates. It’s worth noting here that in a real application you’d probably want to use libraries rather than write these specific wrappers yourself. Later chapters in this book will introduce you to a few of those libraries.

Getting started

You’re going to be creating an app that allows a user to search for gifs through the API for Giphy https://giphy.com, one of the most popular GIF services on the web.

To start, you’ll need a beta key. To get the beta key, navigate to the official docs https://developers.giphy.com/docs/api, and scroll down to “Create an App.”

Follow the instructions there to create an app. You can pick the API key type when prompted. Name your app BestGif:

When you create an app on that page (via the “Create an App” button) you will get a development key, which will suffice to work through this chapter. The API key is displayed under the name of your newly created app like so:

Open the starter project in Android Studio. Then, open GiphyApi.kt and copy the key into the correct place:

private const val API_KEY = "YOUR API KEY HERE"

Once you’ve replaced the API key, run the app. You should see an empty screen with a simple EditText up top. It doesn’t do much yet.

Extending a framework class

It’s often useful to adapt existing framework classes to have a more reactive approach and styling. Luckily, Kotlin’s extension methods allow for a fluid interface to achieve reactive framework classes.

You’re going to start off by extending an EditText widget so you can observe text changes as the user writes them.

Open EditTextUtils.kt and take a look at the EditText.textChanges() method. Right now it returns an empty Observable, but once you’re done it will return an Observable that emits whatever the user types.

To create the actual extension, you’re going to rely on Observable.create() to create an Observable from an existing, non-reactive asynchronous API.

Replace the line returning the empty Observable with the following:

return Observable.create { emitter ->

}

As you’ll recall from Chapter 2, “Observables”, Observable.create() takes in an ObservableOnSubscribe source that allows you to pipe events into an ObservableEmitter object to control how your Observable emits items.

This is a common paradigm when moving from the callback world to the reactive world. Observable.create() provides a convenient interface to wrap existing, callback styled methods.

Now add the following in the Observable.create() block:

val textWatcher = object : TextWatcher {
  override fun afterTextChanged(text: Editable) {
    emitter.onNext(text.toString())
  }

  override fun beforeTextChanged(
    p0: CharSequence?, p1: Int, p2: Int, p3: Int) {}
  override fun onTextChanged(
    p0: CharSequence?, p1: Int, p2: Int, p3: Int) {}
}
addTextChangedListener(textWatcher)

TextWatcher is an Android framework interface that allows you to observe text changes on any TextView or EditText. It allows you to observe the current value before the text changes, as the text changes, and after the text has changed.

While the interface requires methods to do all three things, the only thing you care about is whatever the text is after the EditTexts text has changed. If you wanted to manipulate the text, then you may care about the other two functions, but for now, you can just leave them with empty implementations.

Since all you care about is what the text is after it’s changed, you’re sending the new text received in afterTextChanged() into the Observable via the emitters onNext method. Pretty simple, right?

Last but not least, you’re telling the EditText to use the object you created as a onTextChangedListener. Since you’re creating an extension method, this represents the current instance of EditText.

At this point, you have a fully functioning reactive wrapper. Hooray!

However, it’s not a very responsible wrapper. It never un-registers the text changed listener from the EditText. Since the onTextChangedListener has a strong reference to the EditText, it means your EditText won’t be garbage collected until the Observable is garbage collected, even if the Observable finished long ago. Luckily, the ObservableEmitter class comes with a easy way to trigger cleaning up any resources.

Add the following below the line to add the textChangedListener:

emitter.setCancellable {
  removeTextChangedListener(textWatcher)
}

Now whenever the Observable you’re returning finishes or disposes, it will call that cancellation block and the TextWatcher will be un-registered.

Remember, kids: Safe programming is fun programming. And always wear your seatbelt.

Wiring the extension up

It’s time to use the new extension. Open up GifActivity.kt, and add the following to the bottom of onCreate():

text_input
  // 1
  .textChanges()
  // 2
  .flatMapSingle { GiphyApi.searchForGifs(it) }
  // 3
  .onErrorReturnItem(listOf(GiphyGif(
      "https://media.giphy.com/media/SQ24FpNRW9yRG/giphy.gif")))
  // 4
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  // 5
  .subscribe { adapter.items = it }
  .addTo(disposables)

Here’s what’s happening in the above chain:

1. You’re using the new textChanges() extension to get the changed text from the EditText

2. You’re feeding that received text into GiphyApi.searchForGifs() via flatMapSingle(). Then, searchForGifs() queries the Giphy API and searches for gifs with the given string.

3. You’re using the onErrorReturnItem operator to default to an adorable gif of two kittens if there are any errors.

4. You’re using subscribeOn() and observeOn() to make sure you make the network call from the io thread pool, and you handle the callback on the main thread.

5. You’re subscribing to the whole chain, and updating the adapter’s items to the results received from the API.

Run the app and search for your favorite gif. It should work as expected now, and you should see a list of loading indicators followed by a list of gifs.

There’s one more issue, though. Every time you type a character, the app does a network request for new gifs. In reality, you only really want to start searching once the user has stopped typing for a second or so.

You could update the code in Observable.create() method to add some sort of timer and only emit items every so often, but that sounds like a ton of work. Instead, you can utilize the debounce() operator you learned about in Chapter 6, “Filtering Operators in Practice.”

In case you need a refresher, debounce() limits the items emitted by the source Observable and only emits an item if it isn’t followed by another item after a certain amount of time. It’s perfect for limiting actions taken after typing.

Add the following operator right below textChanges() in the Rx chain:

.debounce(500, TimeUnit.MILLISECONDS)

Now you’ll only receive an item at most once every 500 milliseconds. Run the app again and search for a gif. You should see that you only start seeing loading indicators once you’re done typing.

Wrapping the locations API

The app is looking pretty good, but it’s a bit empty before the user types something in. This seems like a great excuse to wrap some more framework classes!

You’re going to update the app so that it automatically searches for gifs with the name of whatever city the user is currently located in, so you’ll be using the location APIs.

But before you start fetching the location, you need to be a good Android citizen and request permission. You’ll be using a Subject to convert the existing permissions API into a reactive one.

Add the following as instance variables in GifActivity:

private val locationRequestCode = 500
private val permissionsSubject =
    BehaviorSubject.create<Boolean>()

Now add and implement onRequestPermissionsResult(), importing android.Manifest:

override fun onRequestPermissionsResult(
    requestCode: Int,
    permissions: Array<out String>,
    grantResults: IntArray
) {
  super.onRequestPermissionsResult(requestCode, permissions,
      grantResults)
  if (requestCode == locationRequestCode) {
    val locationIndex = permissions.indexOf(
        Manifest.permission.ACCESS_FINE_LOCATION)
    if (locationIndex != -1) {
      val granted = grantResults[locationIndex] ==
          PackageManager.PERMISSION_GRANTED
      permissionsSubject.onNext(granted)
    }
  }
}

Most of the above code is permissions boilerplate that checks the request code and checks that the location permission has been granted or denied. The one interesting piece is the part where you call onNext() on the permissionsSubject with a boolean indicating whether the location permission was granted or denied.

Now that the permissionsSubject is up and running, add the following to the bottom of onCreate():

permissionsSubject
  .doOnSubscribe {
    if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) {
      requestPermissions(arrayOf(
          Manifest.permission.ACCESS_FINE_LOCATION),
          locationRequestCode)
    } else {
      permissionsSubject.onNext(true)
    }
  }
  .filter { it }

You’re building up a new Observable chain here, but not subscribing to it just yet. You’re using doOnSubscribe() to actually kick off the call to request the location permission. If the API version of the device the app is running on is less than M, i.e. before the new permissions model came into effect, you can assume you already have the permission and forward a true event into the permissionsSubject.

You’re also using filter() to filter out any instances where you didn’t receive the location permission. You’ll see why shortly.

You’ve got a new sleek permissions model. It’s time to finish this feature off with a reactive wrapper around the locations API.

Open LocationUtils.kt and look at locationUpdates(). Its return type is Observable<Location>. The location API offers a perfect opportunity for a reactive wrapper, since it works with a constant stream of items in the form of location updates.

For this example you’ll be using the fused location API rather than the raw LocationManager framework API.

To receive location updates, you need to create a LocationRequest object and a FusedLocationProviderClient. Add the following to the top of locationUpdates(), before the return statement:

val currentLocationRequest = LocationRequest()
  .setInterval(500)
  .setFastestInterval(0)
  .setMaxWaitTime(0)
  .setSmallestDisplacement(0f)
  .setPriority(LocationRequest.PRIORITY_HIGH_ACCURACY)

val client = FusedLocationProviderClient(context)

You’re creating that LocationRequest and FusedLocationProviderClient with some configuration options.

Now replace the return Observable.empty() line with the old reliable Observable.create():

 return Observable.create { emitter ->

 }

You’re again using Observable.create() to create a bridge between the callback world and the reactive world.

You need to use requestLocationUpdates() on FusedLocationProvider to actually start the process of getting location updates. To get set up to do that, first add the following in the Observable.create() lambda block:

val callback = object : LocationCallback() {
  override fun onLocationResult(result: LocationResult?) {
    result?.lastLocation?.let { emitter.onNext(it) }
  }
}

This callback will be called whenever the system has a new location update for you, so that’s where you call onNext() on the ObservableEmitter. onLocationResult() can deliver a null location, so that’s why you’re using let.

Below the callback declaration add the following:

client.requestLocationUpdates(currentLocationRequest, callback,
    null)

You learned earlier that you can use setCancellable() on ObservableEmitter to clean up any resources once the Observable terminates. Listening for location updates is an extremely battery intensive task, so it’s doubly important to clean up after yourself when wrapping the location APIs. To that end, add the following right below the requestLocationUpdates() line:

emitter.setCancellable {
  client.removeLocationUpdates(callback)
}

Boom! You’ve wrapped the fused location API with minimal pain, and you’re cleaning up after yourself like a responsible developer.

Time to utilize your newfound location powers.

Head back to GifActivity. It’s time to finish up that Rx chain you started earlier.

Now that you have access to a reactive wrapper around location updates, you can use flatMap() to start receiving location updates after you receive the location permission. Add the following right after the .filter { it } line:

.flatMap { locationUpdates(this) }

Now the type of the Observable switches from Observable<Boolean> to Observable<Location>. Nice! This line of code really shows how powerful it is to create reactive wrappers around traditionally callback based APIs. You can now combine your permission logic and your location logic into one simple declarative stream.

In this app you don’t actually want to keep listening for location updates. All you really care about is the first location you get back. After that first location it will be up to the user to search for their own gif.

You’ve got a few options for limiting the number of location updates to just one. You could try and change up locationUpdates() to only return one Location object and then complete. But that limits the usefulness of locationUpdates().

Instead, you can use take() to only take the first item emitted by your new Observable<Location> object. Add the following operator to the bottom of your chain, after flatMap():

.take(1)

Now you’ll only get one Location object, then the Observable will terminate. And since you were a responsible developer and you used setCancellable() on your ObservableEmitter the app will stop listening for location updates after that first object comes through.

The Giphy API doesn’t accept a Location object. Instead, you want to convert that Location into a String representing the users city. Add the following operator to the bottom of the chain:

.map { cityFromLocation(this, it) }

cityFromLocation() is a method in LocationUtils that uses the Geocoder API to pull out a locality from a Location object.

It might be a good idea to give the user a heads up that the app is searching for their city. You can use the hint attribute on your EditText to show them what’s been searched for. Add the following operator to the chain:

.doOnNext { text_input.hint = it }

It’s time to make the actual network call to fetch some gifs from the city name. Add the following operator:

.flatMapSingle {
  GiphyApi.searchForGifs(it).subscribeOn(Schedulers.io())
}

You need to call subscribeOn() on the actual Observable returned from flatMap() to make sure this this nested Observable is also being run on the correct thread.

Add the following to finish the chain:

.observeOn(AndroidSchedulers.mainThread())
.subscribe { adapter.items = it }
.addTo(disposables)

You’re setting the list of GiphyGifs on your RecyclerView adapter in your subscribe().

The app is ready to go! Give it a run and you should see the app immediately make a request to the Giphy API with whatever city your emulator is set to, after you grant location permissions:

The lift and compose functions

You may come across a few other functions in your Rx travels with regard to custom extensions, especially when interoperating with Java. Since Kotlin supports extension functions, you probably won’t need to use these very often, but it’s still a good idea to understand how they work in case you see them out in the wild in any Java code that you’re interacting with.

The first is compose(), which allows you to write custom RxJava operators that fit inline with the Rx chain.

You will often find that you want to apply a certain set of schedulers when working on reactive apps. For instance, if you’re doing a network call and then you want to display the results of that call to the user, you’ll probably want to use the subscribeOn(Schedulers.io()) and observeOn(AndroidSchedulers.mainThread()) operators and schedulers to make sure you’re running the task on the background thread and applying the results on the main thread. For example:

val observable: Observable<MyModelClass> =
  networkMethodThatReturnsAnObservable()
observable
  .subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe { displayMyResults(it) }

That subscribeOn() and observeOn() combination is so common that lots of people make a Kotlin extension function that applies those two operators:

fun <T> Observable<T>.applySchedulers(): Observable<T> {
  return subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
}

You can use that extension function like an operator in your code:

val observable: Observable<MyModelClass> =
  networkMethodThatReturnsAnObservable()
observable
  .applySchedulers()
  .subscribe { displayMyResults(it) }

However, if you’re still using Java, the above won’t look nearly as clean since Java doesn’t support extension functions, and you have to access those functions through the file they’re created in:

Observable<Integer> observable = Observable.just(1);
Observable<Integer> schedulersApplied =
    FileContaingFunctionKt.applySchedulers(observable);

Pretty gross, huh?

Instead, you can use compose() to keep the chain flowing. All you have to do is create a class that implements the ObservableTransformer<T> interface and override apply():

class ApplySchedulers<T>: ObservableTransformer<T, T> {
  override fun apply(upstream: Observable<T>): ObservableSource<T> {
    return upstream
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
  }
}

Creating that class allows you to write Java code that looks like this:

Observable<Integer> schedulersApplied =
  Observable.just(1).compose(new ApplySchedulers<>());

Much prettier and easier to follow, right?

Fortunately, since you’re using Kotlin you shouldn’t have to mess around with compose() too much!

The last piece of the reactive extensions puzzle is lift(). lift() is an extremely complicated method that many of the internal RxJava operators utilize.

The short explanation is that lift() allows you to create a new operator by reaching into the upstream Observer and directly manipulating its onNext() values.

It’s not worth going too deep into how lift() works. What’s important to know is that if you’re finding yourself in a situation where you feel like you have to use lift(), you’re probably overthinking things. It’s almost always a better choice to make an extension function that utilizes existing RxJava operators.

Testing your custom reactive extension

Testing your custom reactive extensions is just like testing a normal Rx chain. You just need to make sure you’re testing the right thing!

First off, you’ll test the textChanges() extension to EditText you made earlier.

Open EditTextUtilsKtTest.kt and add the following in the body of newStringsReachObserver():

val view = EditText(context)
val testObserver = view.textChanges().test()
view.setText("Test 1")
view.setText("Test 2")
view.setText("Test 3")
view.setText("Test 4")
testObserver.assertValueCount(4)
testObserver
  .assertValues("Test 1", "Test 2", "Test 3", "Test 4")

You’re using the test() method you learned about in Chapter 15, “Testing RxJava Code” to make sure textChanges() emits new text values as expected.

Run the test. It should pass.

Pretty easy right?

Next up you’re going to test that locationUpdates() stops listening for location updates after its associated Observable terminates, ensuring that the locationCallback object is not leaked.

Open LocationUtilsKtTest.kt and add the following to locationUpdatesRemoveOnComplete(). Note that it won’t compile yet until you make some changes in the next step:

val context =
  InstrumentationRegistry.getInstrumentation().targetContext
// 1
val locationProvider =
  mockk<FusedLocationProviderClient>(relaxed = true)

val locationObservable =
  locationUpdates(context, locationProvider)

// 2
verify(exactly = 0) {
  locationProvider.removeLocationUpdates(any<LocationCallback>())
}

locationObservable
  // 3
  .take(0)
  .test()
  .assertComplete()

// 4
verify(exactly = 1) {
  locationProvider
    .removeLocationUpdates(any<LocationCallback>())
}

1. You’re using the mockk library to mock out the FusedLocationProviderClient class so you can verify the location updates are being removed when the Observable completes.
2. You’re using mockk’s verify() to make sure that before the Observable has terminated, the method to remove location updates has not been called.
3. You’re using take(0) to force the location Observable to complete immediately, then validating the Observable has completed.
4. You’re using verify() again to validate that now that the Observable has been completed and the method to remove location updates has been called once (and only once).

Note: For this test, you’re not interested in whether the Observable actually emits any location objects. All you care about here is that the location provider stops listening for the location as soon as the Observable terminates.

You can’t run this code yet because locationUpdates() doesn’t currently accept a FusedLocationProviderClient. But you can fix that easily.

Open LocationUtils.kt, and update locationUpdates() so it takes in the client as a parameter:

fun locationUpdates(
  context: Context,
  client: FusedLocationProviderClient =
    FusedLocationProviderClient(context)
): Observable<Location> {
  ...
}

Next delete the line declaring val client = FusedLocationProviderClient(context) before creating the observer, since client is now being passed in as a parameter.

You should be ready to go. Run the testLocationUpdates test and you should see it pass.

Awesome, you’re not leaking a location callback!

Key points

• You can wrap an existing Android component via Observable.create().
• You should pay attention to any long-lived references inside extensions. Clean up after yourself and cancel any resources when an Observable is disposed.
• You explored compose() and lift() and when to use them (TL;DR avoid lift() unless you know what you’re doing; use compose() if you’re writing Java code).
• Test your reactive wrappers by writing unit tests and mocking any system component.

Where to go from here?

In this chapter, you saw how to implement and wrap the Android framework. Sometimes, it’s very useful to abstract an official Android framework or third party library to better connect with RxJava.

There’s no written rule about when an abstraction is necessary, but the recommendation is to apply this strategy if the code in question meets one or more of these conditions:

• Uses callbacks with success and failure information.
• Needs to inter-operate with other RxJava parts of the application.
• Uses lots of asynchronous constructs to return information.

You also need to know if the code in question has restrictions on which thread the data must be processed. For this reason, it’s a good idea to read the documentation thoroughly before creating an RxJava wrapper.

And don’t forget to look for existing community extensions. There’s a lot of high quality existing reactive wrappers around common Android APIs, some of which you’ll learn about in the next few chapters. If you do write your own wrapper, consider sharing it back with the community!