RxSwiftCommunity/RxSwiftExt

pod 'RxSwiftExt', '~> 5'

pod 'RxSwiftExt/Core'

pod 'RxSwiftExt', '~> 3'

github "RxSwiftCommunity/RxSwiftExt"

  Observable.of(1,2,nil,Int?(4))
    .unwrap()
    .subscribe { print($0) }

next(1)
next(2)
next(4)

  Observable.from(["One","Two","Three"])
    .ignore("Two")
    .subscribe { print($0) }

next(One)
next(Three)
completed

  Observable<Int>
    .of(1,2,3,4,5,6)
    .ignoreWhen { $0 > 2 && $0 < 6 }
    .subscribe { print($0) }

next(1)
next(2)
next(6)
completed

  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

Observable.of("a","b","a","c","b","a","d")
    .distinct()
    .subscribe { print($0) }

next(a)
next(b)
next(c)
next(d)
completed

Observable.of(1,2,3)
    .mapTo("Nope.")
    .subscribe { print($0) }

next(Nope.)
next(Nope.)
next(Nope.)
completed

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

Observable.just(false)
    .not()
    .subscribe { print($0) }

next(true)
completed

Observable.of(true, true)
	.and()
	.subscribe { print($0) }

Observable.of(true, false)
	.and()
	.subscribe { print($0) }

Observable<Bool>.empty()
	.and()
	.subscribe { print($0) }

success(true)
success(false)
completed

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)

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

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

// 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

// 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

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

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)

// 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)

// 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)
	}

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())

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)

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")

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

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

Observable.from([1, 2, 3, 4, 5, 6])
    .count { $0 % 2 == 0 }
    .subscribe()

next(3)
completed

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, 5

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]) @ 5

button.rx.tap
    .flatMap {
        animator1.rx.animate()
            .andThen(animator2.rx.animate(afterDelay: 0.15))
            .andThen(animator3.rx.animate(afterDelay: 0.1))
    }

slider.rx.value.map(CGFloat.init)
    .bind(to: animator.rx.fractionComplete)

tableView.rx.reachedBottom(offset: 40)
            .subscribe { print("Reached bottom") }