If you're using RxSwift, you may have encountered situations where the built-in operators do not bring the exact functionality you want. The RxSwift core is being intentionally kept as compact as possible to avoid bloat. This repository's purpose is to provide additional convenience operators and Reactive Extensions.
This branch of RxSwiftExt targets Swift 5.x and RxSwift 5.0.0 or later.
- If you're looking for the Swift 4 version of RxSwiftExt, please use version
3.4.0of the framework.
Add to your Podfile:
pod 'RxSwiftExt', '~> 5'This will install both the RxSwift and RxCocoa extensions.
If you're interested in only installing the RxSwift extensions, without the RxCocoa extensions, simply use:
pod 'RxSwiftExt/Core'Using Swift 4:
pod 'RxSwiftExt', '~> 3'Add this to your Cartfile
github "RxSwiftCommunity/RxSwiftExt"
RxSwiftExt is all about adding operators and Reactive Extensions to RxSwift!
These operators are much like the RxSwift & RxCocoa core operators, but provide additional useful abilities to your Rx arsenal.
- unwrap
- ignore
- ignoreWhen
- Observable.once
- distinct
- map
- not
- and
- Observable.cascade
- pairwise
- nwise
- retry
- repeatWithBehavior
- catchErrorJustComplete
- pausable
- pausableBuffered
- apply
- filterMap
- Observable.fromAsync
- Observable.zip(with:)
- Observable.merge(with:)
- count
- partition
- bufferWithTrigger
There are two more available operators for materialize()'d sequences:
Read below for details about each operator.
RxSwift/RxCocoa Reactive Extensions are provided to enhance existing objects and classes from the Apple-ecosystem with Reactive abilities.
Unwrap optionals and filter out nil values.
Observable.of(1,2,nil,Int?(4))
.unwrap()
.subscribe { print($0) }next(1)
next(2)
next(4)
Ignore specific elements.
Observable.from(["One","Two","Three"])
.ignore("Two")
.subscribe { print($0) }next(One)
next(Three)
completed
Ignore elements according to closure.
Observable<Int>
.of(1,2,3,4,5,6)
.ignoreWhen { $0 > 2 && $0 < 6 }
.subscribe { print($0) }next(1)
next(2)
next(6)
completed
Send a next element exactly once to the first subscriber that takes it. Further subscribers get an empty sequence.
let obs = Observable.once("Hello world")
print("First")
obs.subscribe { print($0) }
print("Second")
obs.subscribe { print($0) }First
next(Hello world)
completed
Second
completed
Pass elements through only if they were never seen before in the sequence.
Observable.of("a","b","a","c","b","a","d")
.distinct()
.subscribe { print($0) }next(a)
next(b)
next(c)
next(d)
completed
Replace every element with the provided value.
Observable.of(1,2,3)
.mapTo("Nope.")
.subscribe { print($0) }next(Nope.)
next(Nope.)
next(Nope.)
completed
Transform every element to the value at the provided key path.
struct Person {
let name: String
}
Observable
.of(
Person(name: "Bart"),
Person(name: "Lisa"),
Person(name: "Maggie")
)
.mapAt(\.name)
.subscribe { print($0) }next(Bart)
next(Lisa)
next(Maggie)
completed
Negate booleans.
Observable.just(false)
.not()
.subscribe { print($0) }next(true)
completed
Verifies that every value emitted is true
Observable.of(true, true)
.and()
.subscribe { print($0) }
Observable.of(true, false)
.and()
.subscribe { print($0) }
Observable<Bool>.empty()
.and()
.subscribe { print($0) }Returns a Maybe<Bool>:
success(true)
success(false)
completed
Sequentially cascade through a list of observables, dropping previous subscriptions as soon as an observable further down the list starts emitting elements.
let a = PublishSubject<String>()
let b = PublishSubject<String>()
let c = PublishSubject<String>()
Observable.cascade([a,b,c])
.subscribe { print($0) }
a.onNext("a:1")
a.onNext("a:2")
b.onNext("b:1")
a.onNext("a:3")
c.onNext("c:1")
a.onNext("a:4")
b.onNext("b:4")
c.onNext("c:2")next(a:1)
next(a:2)
next(b:1)
next(c:1)
next(c:2)
Groups elements emitted by an Observable into arrays, where each array consists of the last 2 consecutive items; similar to a sliding window.
Observable.from([1, 2, 3, 4, 5, 6])
.pairwise()
.subscribe { print($0) }next((1, 2))
next((2, 3))
next((3, 4))
next((4, 5))
next((5, 6))
completed
Groups elements emitted by an Observable into arrays, where each array consists of the last N consecutive items; similar to a sliding window.
Observable.from([1, 2, 3, 4, 5, 6])
.nwise(3)
.subscribe { print($0) }next([1, 2, 3])
next([2, 3, 4])
next([3, 4, 5])
next([4, 5, 6])
completed
Repeats the source observable sequence using given behavior in case of an error or until it successfully terminated.
There are four behaviors with various predicate and delay options: immediate, delayed, exponentialDelayed and
customTimerDelayed.
// in case of an error initial delay will be 1 second,
// every next delay will be doubled
// delay formula is: initial * pow(1 + multiplier, Double(currentAttempt - 1)), so multiplier 1.0 means, delay will doubled
_ = sampleObservable.retry(.exponentialDelayed(maxCount: 3, initial: 1.0, multiplier: 1.0), scheduler: delayScheduler)
.subscribe(onNext: { event in
print("Receive event: \(event)")
}, onError: { error in
print("Receive error: \(error)")
})Receive event: First
Receive event: Second
Receive event: First
Receive event: Second
Receive event: First
Receive event: Second
Receive error: fatalError
Repeats the source observable sequence using given behavior when it completes. This operator takes the same parameters as the retry operator.
There are four behaviors with various predicate and delay options: immediate, delayed, exponentialDelayed and customTimerDelayed.
// when the sequence completes initial delay will be 1 second,
// every next delay will be doubled
// delay formula is: initial * pow(1 + multiplier, Double(currentAttempt - 1)), so multiplier 1.0 means, delay will doubled
_ = completingObservable.repeatWithBehavior(.exponentialDelayed(maxCount: 3, initial: 1.0, multiplier: 1.2), scheduler: delayScheduler)
.subscribe(onNext: { event in
print("Receive event: \(event)")
})Receive event: First
Receive event: Second
Receive event: First
Receive event: Second
Receive event: First
Receive event: Second
Completes a sequence when an error occurs, dismissing the error condition
let _ = sampleObservable
.do(onError: { print("Source observable emitted error \($0), ignoring it") })
.catchErrorJustComplete()
.subscribe {
print ("\($0)")
}next(First)
next(Second)
Source observable emitted error fatalError, ignoring it
completed
Pauses the elements of the source observable sequence unless the latest element from the second observable sequence is true.
let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
let trueAtThreeSeconds = Observable<Int>.timer(3, scheduler: MainScheduler.instance).map { _ in true }
let falseAtFiveSeconds = Observable<Int>.timer(5, scheduler: MainScheduler.instance).map { _ in false }
let pauser = Observable.of(trueAtThreeSeconds, falseAtFiveSeconds).merge()
let pausedObservable = observable.pausable(pauser)
let _ = pausedObservable
.subscribe { print($0) }next(2)
next(3)
More examples are available in the project's Playground.
Pauses the elements of the source observable sequence unless the latest element from the second observable sequence is true. Elements emitted by the source observable are buffered (with a configurable limit) and "flushed" (re-emitted) when the observable resumes.
Examples are available in the project's Playground.
Apply provides a unified mechanism for applying transformations on Observable sequences, without having to extend ObservableType or repeating your transformations. For additional rationale for this see discussion on github
// An ordinary function that applies some operators to its argument, and returns the resulting Observable
func requestPolicy(_ request: Observable<Void>) -> Observable<Response> {
return request.retry(maxAttempts)
.do(onNext: sideEffect)
.map { Response.success }
.catchError { error in Observable.just(parseRequestError(error: error)) }
// We can apply the function in the apply operator, which preserves the chaining style of invoking Rx operators
let resilientRequest = request.apply(requestPolicy)A common pattern in Rx is to filter out some values, then map the remaining ones to something else. filterMap allows you to do this in one step:
// keep only even numbers and double them
Observable.of(1,2,3,4,5,6)
.filterMap { number in
(number % 2 == 0) ? .ignore : .map(number * 2)
}The sequence above keeps even numbers 2, 4, 6 and produces the sequence 4, 8, 12.
These operators only apply to observable sequences that have been materialized with the materialize() operator (from RxSwift core). errors returns a sequence of filtered error events, ommitting elements. elements returns a sequence of filtered element events, ommitting errors.
let imageResult = _chooseImageButtonPressed.asObservable()
.flatMap { imageReceiver.image.materialize() }
.share()
let image = imageResult
.elements()
.asDriver(onErrorDriveWith: .never())
let errorMessage = imageResult
.errors()
.map(mapErrorMessages)
.unwrap()
.asDriver(onErrorDriveWith: .never())Turns simple asynchronous completion handlers into observable sequences. Suitable for use with existing asynchronous services which call a completion handler with only one parameter. Emits the result produced by the completion handler then completes.
func someAsynchronousService(arg1: String, arg2: Int, completionHandler:(String) -> Void) {
// a service that asynchronously calls
// the given completionHandler
}
let observableService = Observable
.fromAsync(someAsynchronousService)
observableService("Foo", 0)
.subscribe(onNext: { (result) in
print(result)
})
.disposed(by: disposeBag)Convenience version of Observable.zip(_:). Merges the specified observable sequences into one observable sequence by using the selector function whenever all
of the observable sequences have produced an element at a corresponding index.
let first = Observable.from(numbers)
let second = Observable.from(strings)
first.zip(with: second) { i, s in
s + String(i)
}.subscribe(onNext: { (result) in
print(result)
})next("a1")
next("b2")
next("c3")
Convenience version of Observable.merge(_:). Merges elements from the observable sequence with those of a different observable sequences into a single observable sequence.
let oddStream = Observable.of(1, 3, 5)
let evenStream = Observable.of(2, 4, 6)
let otherStream = Observable.of(1, 5, 6)
oddStream.merge(with: evenStream, otherStream)
.subscribe(onNext: { result in
print(result)
})1 2 1 3 4 5 5 6 6
The ofType operator filters the elements of an observable sequence, if that is an instance of the supplied type.
Observable.of(NSNumber(value: 1),
NSDecimalNumber(string: "2"),
NSNumber(value: 3),
NSNumber(value: 4),
NSDecimalNumber(string: "5"),
NSNumber(value: 6))
.ofType(NSDecimalNumber.self)
.subscribe { print($0) }next(2)
next(5)
completed
This example emits 2, 5 (NSDecimalNumber Type).
Emits the number of items emitted by an Observable once it terminates with no errors. If a predicate is given, only elements matching the predicate will be counted.
Observable.from([1, 2, 3, 4, 5, 6])
.count { $0 % 2 == 0 }
.subscribe()next(3)
completed
Partition a stream into two separate streams of elements that match, and don't match, the provided predicate.
let numbers = Observable
.of(1, 2, 3, 4, 5, 6)
let (evens, odds) = numbers.partition { $0 % 2 == 0 }
_ = evens.debug("even").subscribe() // emits 2, 4, 6
_ = odds.debug("odds").subscribe() // emits 1, 3, 5Collects the elements of the source observable, and emits them as an array when the trigger emits.
let observable = Observable<Int>.interval(1, scheduler: MainScheduler.instance)
let signalAtThreeSeconds = Observable<Int>.timer(3, scheduler: MainScheduler.instance).map { _ in () }
let signalAtFiveSeconds = Observable<Int>.timer(5, scheduler: MainScheduler.instance).map { _ in () }
let trigger = Observable.of(signalAtThreeSeconds, signalAtFiveSeconds).merge()
let buffered = observable.bufferWithTrigger(trigger)
buffered.subscribe { print($0) }
// prints next([0, 1, 2]) @ 3, next([3, 4]) @ 5A live demonstration is available in the Playground.
The animate(afterDelay:) operator provides a Completable that triggers the animation upon subscription and completes when the animation ends.
button.rx.tap
.flatMap {
animator1.rx.animate()
.andThen(animator2.rx.animate(afterDelay: 0.15))
.andThen(animator3.rx.animate(afterDelay: 0.1))
}The fractionComplete binder provides a reactive way to bind to UIViewPropertyAnimator.fractionComplete.
slider.rx.value.map(CGFloat.init)
.bind(to: animator.rx.fractionComplete)reachedBottom provides a sequence that emits every time the UIScrollView is scrolled to the bottom, with an optional offset.
tableView.rx.reachedBottom(offset: 40)
.subscribe { print("Reached bottom") }This library belongs to RxSwift Community.
RxSwiftExt is available under the MIT license. See the LICENSE file for more info.