19. RxSwiftExt

Written by Florent Pillet

The RxSwift framework offers a large choice of operators, which can be overwhelming for beginners.

But when you start using a lot of Rx functionality in your applications, you quickly find yourself needing even more operators to solve cases not covered by the core framework.

While creating your own operators isn’t too complicated, you may want to rely on existing, well-tested code that guarantees you’ll always get the results you want.

RxSwiftExt https://git.io/JJXNh is another project living under the RxSwiftCommunity flagship. It is a collection of convenience operators to help cover a variety of situations not handled by core RxSwift. You’re going to learn about a few of them here.

distinct

When processing a sequence of values, it can be useful to only see each element once. For this, you can use the distinct() operator:

_ = Observable.of("a", "b", "a", "c", "b", "a", "d")
  .distinct()
  .toArray()
  .subscribe(onNext: { print($0) })

Only one of each distinct values will be kept, then toArray() (a core RxSwift operator) will group them all in a single array. When the source completes, you‘ll get a single array:

["a","b","c","d"]

Remember, though, that distinct() needs some sort of storage to detect duplicates, so make sure you have an idea of the variety of unique values your sequence will emit.

mapAt

Keypaths in Swift are a powerful way of specifying where to get data from. The mapAt(_:) operator takes advantage of keypaths to let you extract data from a larger object:

struct Person {
    let name: String
}

Observable
  .of(Person(name: "Bart"),
      Person(name: "Lisa"),
      Person(name: "Maggie"))
  .mapAt(\.name)

This comes handy when working with sequences of JSON data!

retry and repeatWithBehavior

Ramping up the powerful operators, you can do a lot with retry(_:) and repeatWithBehavior(_:). (RxSwiftExt’s retry(_:) operator is a more sophisticated version of RxSwift’s retry() family of operators.) Use these new operators to resubscribe to a sequence once it completes or errors, while precisely controlling how and when resubscription occurs. Specify this using one of the following enum cases:

• .immediate(maxCount:) subscribes immediately, up to maxCount times.
• .delayed(maxCount: UInt, time: Double) does the same, but delays resubscription by time seconds.
• .exponentialDelayed(maxCount: UInt, initial: Double, multiplier: Double) is even more powerful, as it automatically multiplies the resubscription delay at every cycle.
• .customTimerDelayed(maxCount: UInt, delayCalculator: (UInt) -> Double) lets you provide a closure that receives the current iteration, and returns the number of seconds by which resubscription should be delayed.

There are many use cases for this. When querying a server, you can ease up over time if your request keeps failing, making sure not to hog the resources on your user’s device:

// try request up to 5 times
// multiply the delay by 3.0 at each attemps
// so retry after 1, 3, 9, 27 and 81 seconds before giving up
let request = URLRequest(url: url)
let tryHard = URLSession.shared.rx.response(request: request)
  .map { response in
      // process response here
  }
  .retry(.exponendialDelayed(maxCount: 3, inital: 1.0, multiplier: 3.0))

catchErrorJustComplete

Sometimes you just want to ignore errors. That’s when the catchErrorJustComplete() operator comes handy, and it does exactly what its name suggests!

let neverErrors = someObservable.catchErrorJustComplete()

pausable and pausableBuffered

When composing observable sequences, you may need to conditionally put one on hold based on another trigger sequence. The pausable(_:) operator takes a pauser sequence of Booleans. The source sequence is ignored until the pauser sequence emits a true value. If your source sequence should keep running while pauser has not yet emitted anything, use startWith(true) like so:

let pausableSequence = source.pausable(pauser.startWith(true))

The pausableBuffered(_:limit:flushOnCompleted:flushOnError:) operator does the same thing, and additionally stores incoming values from the source sequence while paused, with a variety of options to control how much should be stored and whether memorized values should be sent upon source sequence completion or error.

bufferWithTrigger

Another trigger-using operator, bufferWithTrigger(_:) is related to the core buffer(timeSpan:count:scheduler) operator. It also buffers the values emitted by the source sequence, and emits an array of the buffered values every time the trigger sequence emits something. The type of the values emitted by the trigger sequence does not matter.

withUnretained

Resource management (object ownership and lifetime) is always a major concern for developers, particularly in a language like Swift where closures easily capture accidental strong references.

RxSwiftExt has an interesting take on it with an observable sequence that can stop when an object is no longer in use.

The withUnretained(_:) operator keeps a weak reference to the given object, and tests whether it still exists every time the source sequence emits something. If the object is still referenced, the closure you pass is executed and the value it returns is the new emitted value:

var anObject: SomeClass! = SomeClass()

_ = Observable
    .of(1, 2, 3, 5, 8, 13, 18, 21, 23)
    .withUnretained(anObject)
    .debug("Combined Object with Emitted Events")
    .do(onNext: { _, value in
        if value == 13 {
            // When anObject becomes nil, the next value of the source
            // sequence will try to retain it and fail.
            // As soon as it fails, the sequence will complete.
            anObject = nil
        }
    })
    .subscribe()

This operator is also highly useful inside View Controllers, where you want to safely capture self while avoiding the guard let self = self dance:

message
  .withUnretained(self) { vc, message in
    vc.showMessage(message)
  }

partition

As you learned in the chapters on filtering operators, you often find yourself in the need to filter only specific elements of a stream. In many other cases though, it can be useful to partition a stream to two streams based on a condition, so elements not matching the predicate are part of a second stream. This is what the partition operator is all about:

let (evens, odds) = Observable
                      .of(1, 2, 3, 5, 6, 7, 8)
                      .partition { $0 % 2 == 0 }

_ = evens.debug("evens").subscribe() // Emits 2, 6, 8
_ = odds.debug("odds").subscribe() // Emits 1, 3, 5, 7

mapMany

You‘ve used map many times throughout this book, so you probably remember that its used to transform individually emitted elements. mapMany is a specialization of map for collection types. It would map every individual element inside a collection-typed observable sequence, such as an Array:

_ = Observable.of(
  [1, 3, 5],
  [2, 4]
)
.mapMany { pow(2, $0) }
.debug("powers of 2")
.subscribe() // Emits [2, 8, 32] and [4, 16]