RxSwift шпаргалка по операторам (+ PDF)13.04.2016 08:48

public func example(description: String, action: () -> ()) {
    print("\n--- \(description) example ---")
    action()
}

Во всех примерах, где необходимо работать с временными задержками, если этот код будет запускаться в песочнице — необходимо прописать

import XCPlayground
XCPlaygroundPage.currentPage.needsIndefiniteExecution = true

Так же подразумевается, что читатель имеет общее представление о том, что такое реактивное программирование в общем и о RxSwift в частности. Не знаю есть ли смысл городить очередную вводную.

asObservable

Этот метод реализован в классах RxSwift, если они поддерживают конвертацию в Observable. Например: ControlEvent, ControlProperty, Variable, Driver

example("as Observable") {
    let variable = Variable(0)
    
    variable.asObservable().subscribe { e in
        print(e)
    }
    variable.value = 1
}

Консоль:

--- as Observable example ---
Next(0)
Next(1)
Completed

В данном примере мы Variable преобразовали в Observable и подписались на его события

create

Этот метод позволяет создавать Observable с нуля, полностью контролируя какие элементы и когда он будет генерировать.

example("create") {
    let firstSequence = Observable.of(1, 2, 3)
    let secondSequence = Observable.of("A", "B", "C")
    
    let multipleSequence = Observable>.create { observer in
        observer.on(.Next(firstSequence))
        observer.on(.Next(secondSequence))
        return NopDisposable.instance
    }
    let concatSequence = multipleSequence.concat()
    concatSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- create example ---
Next(1)
Next(2)
Next(3)
Next(A)
Next(B)
Next(C)

В данном примере мы создали Observable вручную, он сгенерирует два Observable, элементы которых мы затем объединим командой concat, на получившийся Observable мы и подпишемся

deferred

Этот оператор позволяет отложить создание Observable, до момента подписки с помощью subscribe

example("without deferred") {
    var i = 1
    let justObservable = Observable.just(i)
    i = 2
    _ = justObservable.subscribeNext{ print ("i = \($0)") }
}

example("with deferred") {
    var i = 1
    let deferredJustObservable = Observable.deferred{
        Observable.just(i)
    }
    i = 2
    _ = deferredJustObservable.subscribeNext{ print ("i = \($0)") }
}

Консоль:

--- without deferred example ---
i = 1

--- with deferred example ---
i = 2

В первом случае Observable создается сразу, с помощью Observable.just (i), и изменение значение i уже не влияет на генерируемый элемент это последовательностью. Во втором же случае мы создаем с помощью deferred, и мы можем поменять значение i перед subscribe

empty

Пустая последовательность, заканчивающаяся Completed

example("empty") {
    let sequence = Observable.empty()
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- empty example ---
Completed

error

Создаст последовательность, которая состоит из одного события — Error

example("error") {
    let sequence = Observable
        .error(RxError.Unknown)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- error example ---
Error(Unknown error occured.)

interval

Создает бесконечную последовательность, возрастающую с 0 с шагом 1, с указанной периодичностью

example("interval") {
    let sequence = Observable.interval(1, scheduler: MainScheduler.instance)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- interval example ---
Next(0)
Next(1)
Next(2)
Next(3)
Next(4)
....

just

Создает последовательность из любого значения, которая завершается Completed

example("just") {
    let sequence = Observable.just(100)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- just example ---
Next(100)
Completed

never

Пустая последовательность, чьи observer«ы никогда не вызываются, т.е. не будет сгенерировано ни одно событие

example("never") {
    let sequence = Observable.never()
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- never example ---

of

Последовательность из variadic переменной, после всех элементов генерируется Completed

example("simple of") {
    let sequence = Observable.of(1, 2)

    sequence.subscribe { e in
        print(e)
    }
}

example("of for Observables") {
    let firstSequence = Observable.of(1, 2, 3)
    let secondSequence = Observable.of("A", "B", "C")
    
    let bothSequence = Observable.of(firstSequence, secondSequence)
    let mergedSequence = bothSequence.merge()
    
    mergedSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- simple of example ---
Next(1)
Next(2)
Completed

--- of for Observables example ---
Next(1)
Next(2)
Next(3)
Next(A)
Next(B)
Next(C)
Completed

В первом случае мы создали последовательность из двух чисел. Во втором из двух Observable, а затем их объединили между собой с помощью оператора merge

range

Создает последовательность с конечным числом элементов, возрастающую с шагом 1 от указанного значения указанное число раз, после всех элементов генерируется Completed

example("range") {
    let sequence = Observable.range(start: 5, count: 3)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- range example ---
Next(5)
Next(6)
Next(7)
Completed

Сгенерировались элементы начиная с 5, 3 раза с шагом 1

repeatElement

Бесконечно создавать указанный элемент, без задержек. Никогда не будет сгенерированы события Completed или Error

example("repeatElement") {
    let sequence = Observable.repeatElement(1, scheduler: MainScheduler.instance)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- repeatElement example ---
Next(1)
Next(2)
Next(3)
.....

timer

Бесконечная последовательность, возрастающая с 0 с шагом 1, с указанной периодичностью и возможность задержки при старте. Никогда не будет сгенерированы события Completed или Error

example("timer") {
    let sequence = Observable.timer(2, period: 3, scheduler: MainScheduler.instance)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- timer example ---
Next(0)
Next(1)
Next(2)

В данном примере последовательность начнет генерировать элементы с задержкой в 2 секунды, каждые 3 секунды

amb

SO = [Observable] или SO1, SO2 = Observable
RO = Observable

Из всех Observable SO выбирается тот, который первым начинает генерировать элементы, его элементы и дублируются в RO, остальные SO игнорируются

example("amb") {
    let subjectA = PublishSubject()
    let subjectB = PublishSubject()
    let subjectC = PublishSubject()
    let subjectD = PublishSubject()
    
    let ambSequence = [subjectA, subjectB, subjectC, subjectD].amb()
    ambSequence.subscribe { e in
        print(e)
    }
    
    subjectC.onNext(0)
    subjectA.onNext(3)
    subjectB.onNext(102)
    subjectC.onNext(1)
    subjectD.onNext(45)
}

Консоль:

--- amb example ---
Next(0)
Next(1)

Т.к. первым сгенерировал элемент subjectC, — лишь его элементы дублируются в RO, остальные игнорируются

combineLatest

SO = SO1, SO2,... SON = Observable
RO = Observable 

Как только все Observable сгенерировали хотя бы по одному элементу — эти элементы используются в качестве параметров в переданную функцию, и результат этой функции генерируется RO в качестве элемента. В дальнейшем при генерации любым Observable элемента — генерируется новый результат функции с последними элементами из всех комбинируемых Observable

example("combineLatest") {
    let firstSequence = createSequenceWithWait([1,2,3], waitTime: 2) { element in
        "\(element)"
    }.debug("firstSequence")
    let secondSequence = createSequenceWithWait(["A", "B", "C"], waitTime: 1) { element in
        "\(element)"
        }
        .delaySubscription(3, scheduler: MainScheduler.instance)
        .debug("secondSequence")
    
    let concatSequence = Observable.combineLatest(firstSequence, secondSequence) {
        first, second -> String in
        "\(first) - \(second)"
    }
    concatSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- combineLatest example ---
2016-04-12 16:59:35.421: firstSequence -> subscribed
2016-04-12 16:59:35.422: secondSequence -> subscribed
2016-04-12 16:59:35.434: firstSequence -> Event Next(1)
2016-04-12 16:59:37.423: firstSequence -> Event Next(2)
2016-04-12 16:59:38.423: secondSequence -> Event Next(A)
Next(2 - A)
2016-04-12 16:59:39.423: firstSequence -> Event Next(3)
Next(3 - A)
2016-04-12 16:59:39.522: secondSequence -> Event Next(B)
Next(3 - B)
2016-04-12 16:59:40.622: secondSequence -> Event Next(C)
Next(3 - C)
2016-04-12 16:59:41.722: firstSequence -> Event Completed
2016-04-12 16:59:41.722: firstSequence -> disposed
2016-04-12 16:59:41.722: secondSequence -> Event Completed
2016-04-12 16:59:41.722: secondSequence -> disposed
Completed

В этом примере я создал Observable с помощью createSequenceWithWait, чтобы элементы генерировались с разной задержкой, чтобы было видно как перемешиваются элементы.
firstSequence успел сгенерировать 1 и 2, прежде чем secondSequence сгенерировал A, поэтому 1 отбросили, и первым выводом стало 2 — A

concat

SO = Observable> или SO1, SO2 = Observable
RO = Observable

В RO элементы включают сначала все элементы первого Observable, и лишь затем следующего. Это означает, что если первый Observable никогда не сгенерирует Completed, — элементы второго никогда не поступят в RO. Ошибка в текущем Observable пробрасывается в RO

example("concat object method") {
    let firstSequence = Observable.of(1, 2, 3)
    let secondSequence = Observable.of("A", "B", "C")
    let concatSequence = firstSequence.concat(secondSequence)
    concatSequence.subscribe { e in
        print(e)
    }
}

example("concat from array") {
    let firstSequence = Observable.of(1,2,3)
    let secondSequence = Observable.of(4,5,6)
    let concatSequence = Observable.of(firstSequence, secondSequence)
        .concat()
    
    concatSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- concat object method example ---
Next(1)
Next(2)
Next(3)
Next(A)
Next(B)
Next(C)
Completed

--- concat from array example ---
Next(1)
Next(2)
Next(3)
Next(4)
Next(5)
Next(6)
Completed

В первом примере мы присоединяем второй Observable к первому.
Во втором генерируем последовательность из массива.

merge

SO = Observable>
RO = Observable

Элементы RO включают элементы из исходных Observable в том порядке, в котором они были выпущены в исходных Observable

example("simple merge") {
    let firstSequence = Observable.of(1, 2, 3)
    let secondSequence = Observable.of("A", "B", "C")

    let bothSequence = Observable.of(firstSequence, secondSequence)
    let mergedSequence = bothSequence.merge()
    
    mergedSequence.subscribe { e in
        print(e)
    }
}

example("merge with wait") {
    let firstSequence = createSequenceWithWait([1,2,3]) { element in
        "\(element)"
    }
    let secondSequence = createSequenceWithWait(["A", "B", "C"], waitTime: 2) { element in
        "\(element)"
    }

    let bothSequence = Observable.of(firstSequence, secondSequence)
    let mergedSequence = bothSequence.merge()
    
    mergedSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- simple merge example ---
Next(1)
Next(2)
Next(3)
Next(A)
Next(B)
Next(C)
Completed

--- merge with wait example ---
Next(1)
Next(A)
Next(2)
Next(3)
Next(B)
Next(C)
Completed

В первом примере мы сливаем две последовательности созданные без задержки, в итоге первый Observable успевает сгенерировать все свои элементы перед тем как это начнет делать второй, результат оказывается идентичен concat
Во втором же случае последовательности сделаны с задержкой в генерации, и видно что элементы в RO теперь вперемешку, в том порядке в котором они были сгенерированы в исходных Observable

startWith

SO = Observable
RO = Observable

В начало SO добавляются элементы переданные в качестве аргумента

example("startWith") {
    let sequence = Observable.of(1, 2, 3).startWith(0)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- startWith example ---
Next(0)
Next(1)
Next(2)
Next(3)
Completed

switchLatest

SO = Observable>
RO = Observable 

Изначально подписываемся на O1 генерируемого SO, его элементы зеркально генерируются в RO. Как только из SO генерируется очередной Observable — элементы предыдущего Observable отбрасываются, т.к. происходит отписка от O1, подписываемся на O2 и так далее. Таким образом в RO — элементы лишь из последнего сгенерированного Observable

example("switchLatest") {
    let varA = Variable(0)
    let varB = Variable(100)
    
    let proxyVar = Variable(varA.asObservable())
    let concatSequence = proxyVar.asObservable().switchLatest()
    
    concatSequence.subscribe { e in
        print(e)
    }
    varA.value = 1
    varA.value = 2
    varB.value = 3
    proxyVar.value = varB.asObservable()
    varB.value = 4
    varA.value = 5
}


example("switchLatest") {
    let observableA = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(10))
        }
        delay(3) {
            observer.on(.Next(20))
        }
        delay(5) {
            observer.onCompleted()
        }
        return NopDisposable.instance
        }.debug("oA")
    let observableB = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(100))
        }
        delay(1) {
            observer.on(.Next(200))
        }
        delay(2) {
            observer.onCompleted()
        }
        return NopDisposable.instance
        }.debug("oB")
    
    let observableC = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(1000))
        }
        delay(1) {
            observer.on(.Next(2000))
        }
        delay(2) {
            observer.onCompleted()
        }
        return NopDisposable.instance
        }.debug("oC")
    
        let subjects = [observableA, observableB, observableC]
        let sequence:Observable> = createSequenceWithWait([observableA, observableB, observableC],waitTime:1) {$0}
        let switchLatestSequence:Observable = sequence.switchLatest()
        switchLatestSequence.subscribe { e in
            print(e)
        }
}

Консоль:

--- switchLatest example ---
Next(0)
Next(1)
Next(2)
Next(3)
Next(4)
Completed

--- switchLatest example ---
2016-04-12 17:15:22.710: oA -> subscribed
2016-04-12 17:15:22.711: oA -> Event Next(10)
Next(10)
2016-04-12 17:15:23.797: oA -> disposed // происходит отписка как только сгенерировался oB
2016-04-12 17:15:23.797: oB -> subscribed
2016-04-12 17:15:23.797: oB -> Event Next(100)
Next(100)
2016-04-12 17:15:24.703: oB -> disposed // происходит отписка как только сгенерировался oC
2016-04-12 17:15:24.703: oC -> subscribed 
2016-04-12 17:15:24.703: oC -> Event Next(1000)
Next(1000)
2016-04-12 17:15:25.800: oC -> Event Next(2000)
Next(2000)
2016-04-12 17:15:26.703: oC -> Event Completed
2016-04-12 17:15:26.703: oC -> disposed
Completed

В первом примере показано как команда работает в статике, когда мы руками переподключаем Observable.
Во втором же у нас последовательности с задержками. observableA, observableB, observableC из SO генерируются раз в 1 секунду. Их же элементы генерируются с различными задержками.

withLatestFrom

SO1, SO2 = Observable
RO = Observable

Как только O1 генерирует элемент проверяется сгенерирован ли хоть один элемент в O2, если да, то берутся последние элементы из O1 и O2 и используются в качестве аргументов для переданной функции, результат которой генерируется RO в качестве элемента

example("withLatestFrom") {
    let varA = Variable(0)
    let varB = Variable(10)
    
    let withLatestFromSequence = varA.asObservable().withLatestFrom(varB.asObservable()) {
        "\($0) - \($1)"
    }
    withLatestFromSequence.subscribe { e in
        print(e)
    }
    varA.value = 1
    varA.value = 2 
    varB.value = 20
    
    varB.value = 30
    varA.value = 5
    varA.value = 6
}

Консоль:

--- withLatestFrom example ---
Next(0 - 10)
Next(1 - 10)
Next(2 - 10)
Next(5 - 30)
Next(6 - 30)
Completed

zip

SO = Observable>
RO = Observable 

Элементы RO представляют собой комбинацию из элементов сгенерированных исходными Observable, объединение идет по индексу выпущенного элемента

example("zip with simple Variable") {
    let varA = Variable(0)
    let varB = Variable(10)
    
    let zippedSequence = Observable.zip(varA.asObservable(), varB.asObservable()) { "\($0) - \($1)"
    }
    
    zippedSequence.subscribe { e in
        print(e)
    }
    varA.value = 1
    varA.value = 2
    varB.value = 20
    
    varB.value = 30
    varA.value = 3
    varA.value = 4
}

example("zip with PublishSubject") {
    let subjectA = PublishSubject()
    let subjectB = PublishSubject()

    let zippedSequence = Observable.zip(subjectA, subjectB) { "\($0) - \($1)"
    }
    
    zippedSequence.subscribe { e in
        print(e)
    }
    subjectA.onNext(0)
    subjectA.onNext(1)
    subjectA.onNext(2)
    subjectB.onNext(100)
    subjectB.onNext(101)
    subjectA.onNext(3)
    subjectB.onNext(102)
    subjectA.onNext(4)
}

Консоль:

--- zip with simple Variable example ---
Next(0 - 10)
Next(1 - 20)
Next(2 - 30)
Completed

--- zip with PublishSubject example ---
Next(0 - 100)
Next(1 - 101)
Next(2 - 102)

Из примеров видно, что элементы комбинируются в том порядке, в каком они были сгенерированы в исходных Observable

distinctUntilChanged

Пропускаем все повторяющиеся подряд идущие элементы

example("distinctUntilChanged") {
    let sequence = Observable.of(1, 2, 2, 3, 4, 4, 4, 1).distinctUntilChanged()
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- distinctUntilChanged example ---
Next(1)
Next(2)
Next(3)
Next(4)
Next(1)
Completed

Здесь тонкий момент, что отбрасываются не уникальные для всей последовательности элементы, а лишь те, которые идут подряд.

elementAt

В RO попадает лишь элемент выпущенный N по счету

example("elementAt") {
    let sequence = Observable.of(0, 10, 20, 30, 40)
        .elementAt(2)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- elementAt example ---
Next(20)
Completed

filter

Отбрасываются все элементы, которые не удовлетворяют заданным условиям

example("filter") {
    let sequence = Observable.of(1, 20, 3, 40)
        .filter{ $0 > 10}
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- filter example ---
Next(20)
Next(40)
Completed

ignoreElements

Отбрасывает все элементы, передаёт только терминальные сообщения Completed и Error

example("ignoreElements") {
    let sequence = Observable.of(1, 2, 3, 4)
        .ignoreElements()
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- ignoreElements example ---
Completed

sample

При каждом сгенерированном элементе последовательности семплера (воспринимать как таймер) — брать последний выпущенный элемент исходной последовательности и дублировать его в RO, ЕСЛИ он не был сгенерирован ранее

example("sampler") {
    let sampler = Observable.interval(1, scheduler: MainScheduler.instance).debug("sampler")
    
    let sequence:Observable = createSequenceWithWait([1,2,3,4], waitTime: 3).sample(sampler)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- sampler example ---
2016-04-12 18:28:20.322: sampler -> subscribed
2016-04-12 18:28:21.323: sampler -> Event Next(0)
Next(1)
2016-04-12 18:28:22.324: sampler -> Event Next(1) // элемент в RO не был сгенерирован, т.к. он уже был сгенерирован ранее
2016-04-12 18:28:23.323: sampler -> Event Next(2)
Next(2)
2016-04-12 18:28:24.323: sampler -> Event Next(3) // элемент в RO не был сгенерирован, т.к. он уже был сгенерирован ранее
2016-04-12 18:28:25.323: sampler -> Event Next(4) // элемент в RO не был сгенерирован, т.к. он уже был сгенерирован ранее
2016-04-12 18:28:26.323: sampler -> Event Next(5)
Next(3)
...

single

Из исходной последовательности берется единственный элемент, если элементов > 1 — генерировать ошибку. Есть вариант с предикатом

example("single generate error") {
    let sequence = Observable.of(1, 2, 3, 4).single()
    sequence.subscribe { e in
        print(e)
    }
}

example("single") {
    let sequence = Observable.of(1, 2, 3, 5).single {
        $0 % 2 == 0
    }
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- single generate error example ---
Next(1)
Error(Sequence contains more than one element.)

--- single example ---
Next(2)
Completed

В первом примере в исходной последовательности оказалось больше 1 элемента, поэтому была сгенерирована ошибка в момент генерирования в SO второго элемента
Во втором примере условиям предиката удовлетворил всего 1 элемент, поэтому ошибки сгенерировано не было

skip

Из SO отбрасываем первые N элементов

example("skip") {
    let sequence = Observable.of(1, 2, 3, 4).skip(2)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- skip example ---
Next(3)
Next(4)
Completed

skip (duration)

Из SO отбрасываем первые элементы, которые были сгенерированы в первые N

example("skip duration with wait") {
    let sequence = createSequenceWithWait([1,2,3,4]) { $0 }.skip(2, scheduler: MainScheduler.instance)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- skip duration with wait example ---
Next(3)
Next(4)
Completed

skipUntil

Отбрасываем из SO элементы, которые были сгенерированы до начала генерации элементов последовательностью переданной в качестве параметра

example("skipUntil") {
    let firstSequence = createSequenceWithWait([1,2,3,4]) { $0 }
    let secondSequence = Observable.just(1)
        .delaySubscription(1, scheduler: MainScheduler.instance)
    let skippedSequence = firstSequence.skipUntil(secondSequence)
    
    skippedSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- skipUntil example ---
Next(3)
Next(4)
Completed

Генерация элементов в secondSequence была отложена на 1 секунду с помощью команды delaySubscription, таким образом элементы из firstSequence стали дублироваться в RO лишь через 1 секунду

skipWhile

Отбрасываем из SO элементы до тех пор, пока истинно условие возвращаемой переданное функцией

example("skipWhile") {
    let firstSequence = [1,2,3,4,0].toObservable()
    let skipSequence = firstSequence.skipWhile { $0 < 3 }
    
    skipSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- skipWhile example ---
Next(3)
Next(4)
Next(0)
Completed

skipWhileWithIndex

Отбрасываем из SO элементы до тех пор, пока истинно условие возвращаемой переданное функцией. Отличие от skipWhile в том, что в качестве еще одного параметра переданного в функцию является индекс сгенерированного элемента

example("skipWhileWithIndex") {
    let firstSequence = [1,2,5,0,7].toObservable()
    let skipSequence = firstSequence.skipWhileWithIndex{ value, idx in
        value < 4 || idx < 2
    }
    skipSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- skipWhileWithIndex example ---
Next(5)
Next(0)
Next(7)
Completed

take

Из SO берутся лишь первые N элементов

example("take") {
    let sequence = Observable.of(1, 2, 3, 4).take(2)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- take example ---
Next(1)
Next(2)
Completed

take (duration)

Из SO берутся лишь элементы сгенерированные в первые N секунд

example("take duration with wait") {
    let sequence = createSequenceWithWait([1,2,3,4]) { $0 }
    let takeSequence = sequence.take(2, scheduler: MainScheduler.instance)
    takeSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- take duration with wait example ---
Next(1)
Next(2)
Completed

takeLast

Из SO берутся лишь последние N элементов. Что означает, если SO никогда не закончит генерировать элементы — в RO не попадет ни одного элемента.

example("takeLast") {
    let sequence = Observable.of(1, 2, 3, 4).takeLast(2)
    sequence.subscribe { e in
        print(e)
    }
}

example("takeLast with wait") {
    let sequence = createSequenceWithWait([1,2,3,4]) { $0 }
    let takeSequence = sequence.takeLast(2)
    takeSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- takeLast example ---
Next(3)
Next(4)
Completed

--- takeLast with wait example ---
Next(3)
Next(4)
Completed

Второй пример приведен для иллюстрации в задержке генерации элементов в RO из за ожидания завершения генерации элементов в SO

takeUntil

Из SO берутся элементы, которые были выпущены до начала генерации элементов последовательностью переданной в качестве параметра

example("takeUntil") {
    let stopSequence = Observable.just(1)
        .delaySubscription(2, scheduler: MainScheduler.instance)
    let sequence = createSequenceWithWait([1,2,3,4]) { $0 }
        .takeUntil(stopSequence)
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- takeUntil example ---
Next(1)
Next(2)
Completed

takeWhile

Из SO берутся Элементы до тех пор, пока истинно условие возвращаемой переданной функцией

example("takeWhile") {
    let sequence = [1,2,3,4].toObservable().takeWhile{ $0 < 3 }
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- takeWhile example ---
Next(1)
Next(2)
Completed

takeWhileWithIndex

Из SO берутся Элементы до тех пор, пока истинно условие возвращаемой переданной функцией. Отличие от takeWhile в том, что в качестве еще одного параметра переданного в функцию является индекс сгенерированного элемента

example("takeWhileWithIndex") {
    let sequence = [1,2,3,4,5,6].toObservable()
        .takeWhileWithIndex{ (val, idx) in
            val % 2 == 0 || idx < 3
    }
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- takeWhileWithIndex example ---
Next(1)
Next(2)
Next(3)
Next(4)
Completed

throttle

Из SO берутся лишь элементы, после которых не было новых элементов N секунд.

example("throttle") {
    let sequence = Observable.of(1, 2, 3, 4)
        .throttle(1, scheduler: MainScheduler.instance)
    sequence.subscribe { e in
        print(e)
    }
}

example("throttle with wait") {
    let sequence = createSequenceWithWait([1,2,3,4]) { $0 }
        .throttle(0.5, scheduler: MainScheduler.instance)
    
    sequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- throttle example ---
Next(4)
Completed

--- throttle with wait example ---
Next(1)
Next(2)
Next(3)
Next(4)
Completed

В первом случае SO генерирует элементы без задержек, поэтому лишь последний элемент не имеет после себя новых элементов.
Во втором примере элементы генерируются медленнее, чем N секунд переданные в throttle, поэтому за каждым генерируемым элементом достаточный временной промежуток.

buffer

SO = Observable<>>
RO = Observable<[T]>

Элементы из SO по определенным правилам объединяются в массивы и генерируются в RO. В качестве параметров передается count, — максимальное число элементов в массиве, и timeSpan время максимального ожидания наполнения текущего массива из элементов SO. Таким образом элемент RO, являет собой массив [T], длинной от 0 до count.

example("buffer") {
    let varA = Variable(0)
    
    let bufferSequence = varA.asObservable()
        .buffer(timeSpan: 3, count: 3, scheduler: MainScheduler.instance)
    bufferSequence.subscribe { e in
        print("\(NSDate()) - \(e)")
    }
    varA.value = 1
    varA.value = 2
    varA.value = 3
    delay(3) {
        varA.value = 4
        varA.value = 5
        delay(5) {
            varA.value = 6
        }
    }
}

Консоль:

--- buffer example ---
2016-04-12 16:10:58 +0000 - Next([0, 1, 2])
2016-04-12 16:11:01 +0000 - Next([3]) 
2016-04-12 16:11:04 +0000 - Next([4, 5])
2016-04-12 16:11:07 +0000 - Next([6])
2016-04-12 16:11:07 +0000 - Completed

Длина массива была указана как 3, — как только были сгенерированы 3 элемента — в RO сгенерировался элемент [0, 1, 2]
После генерации элемента 3, — была задержка в 3 секунды, сработал таймаут, и массив не был заполнен полностью.
То же касается и задержки после генерации элемента 5

flatMap

Каждый элемент SO превращается в отдельный Observable, и все элементы из [O1, O2, O3…] объединяются в RO. Порядок генерации элементов в RO зависит от времени их генерации в исходных [O1, O2, O3…] (как в команде merge)

example("flatMap with wait") {
    let sequence:Observable = createSequenceWithWait([0,1,2], waitTime: 1) { $0 }
    let flatMapSequence:Observable = sequence.flatMap{val in
        createSequenceWithWait([10,11,12], waitTime: 2) { element in
            "\(element) - \(val)"
        }
    }
    flatMapSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- flatMap with wait example ---
Next(10 - 0)
Next(10 - 1)
Next(11 - 0)
Next(10 - 2)
Next(11 - 1)
Next(12 - 0)
Next(11 - 2)
Next(12 - 1)
Next(12 - 2)
Completed

flatMapFirst

Каждый элемент SO превращается в отдельный Observable.
1) Изначально подписываемся на O1, его элементы зеркально генерируются в RO. Пока O1 генерирует элементы — все последующие Observable сгенерированные из SO отбрасываются, на них не подписываемся.
2) как только O1 оканчивается, — если будет сгенерирован новый Observable — на него подпишутся и его элементы будут дублироваться в RO.
Повторяем пункт 1, но вместо O1 берем последний сгенерированный Observable

example("flatMapFirst") {
    let sequence:Observable = Observable.of(10, 20, 30)
        .debug("sequence")
    let flatMapSequence:Observable = sequence
        .flatMapFirst{val in
            Observable.of(0, 1, 2)
                .map{"\($0) - \(val)"
            }
    }
    flatMapSequence.subscribe { e in
        print(e)
    }
}


example("flatMapFirst with delay") {
    let subjectA = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(10))
        }
        delay(1) {
            observer.on(.Next(20))
        }
        delay(7) {
            observer.onCompleted()
        }
        return NopDisposable.instance
    }.debug("sA")
    let subjectB = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(100))
        }
        delay(1) {
            observer.on(.Next(200))
        }
        delay(2) {
            observer.onCompleted()
        }
        return NopDisposable.instance
    }.debug("sB")

    let subjectC = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(1000))
        }
        delay(1) {
            observer.on(.Next(2000))
        }
        delay(2) {
            observer.onCompleted()
        }
        return NopDisposable.instance
    }.debug("sC")

    let subjects = [subjectA, subjectB, subjectC]
    let sequence:Observable = createSequenceWithWait([0, 1, 2],waitTime:4){$0}
        .debug("sequence")
    let flatMapSequence:Observable = sequence.flatMapFirst{val in
        return subjects[val].asObservable()
        }.debug("flatMapSequence")
    flatMapSequence.subscribe { e in
        print(e)
    }
}

Консоль:

--- flatMapFirst example ---
2016-04-12 19:19:46.915: sequence -> subscribed
2016-04-12 19:19:46.916: sequence -> Event Next(10)
Next(0 - 10)
Next(1 - 10)
Next(2 - 10)
2016-04-12 19:19:46.918: sequence -> Event Next(20)
Next(0 - 20)
Next(1 - 20)
Next(2 - 20)
2016-04-12 19:19:46.919: sequence -> Event Next(30)
Next(0 - 30)
Next(1 - 30)
Next(2 - 30)
2016-04-12 19:19:46.921: sequence -> Event Completed
Completed
2016-04-12 19:19:46.921: sequence -> disposed

--- flatMapFirst with delay example ---
2016-04-12 19:19:46.925: flatMapSequence -> subscribed
2016-04-12 19:19:46.926: sequence -> subscribed
2016-04-12 19:19:46.935: sequence -> Event Next(0) // SO генерирует 1й элемент
2016-04-12 19:19:46.935: sA -> subscribed // на его основе генерируется Observable sA, на  который мы подписываемся
2016-04-12 19:19:46.936: sA -> Event Next(10)
2016-04-12 19:19:46.936: flatMapSequence -> Event Next(10)
Next(10)
2016-04-12 19:19:47.936: sA -> Event Next(20)
2016-04-12 19:19:47.936: flatMapSequence -> Event Next(20)
Next(20)
2016-04-12 19:19:50.926: sequence -> Event Next(1) // SO генерирует 2й элемент, но на этот момент sA еще генерирует элементы, поэтому подписки на sB не произошло, он просто отбрасывается
2016-04-12 19:19:53.935: sA -> Event Completed 
2016-04-12 19:19:53.936: sA -> disposed // sA закончил генерировать элементы, от него отписались
2016-04-12 19:19:55.137: sequence -> Event Next(2) // SO генерирует 3й элемент
2016-04-12 19:19:55.137: sC -> subscribed // т.к. на этот момент нет активных Observable (от sA мы отписались, sB - мы отбросили) - мы можем подписаться на него
2016-04-12 19:19:55.137: sC -> Event Next(1000)
2016-04-12 19:19:55.137: flatMapSequence -> Event Next(1000)
Next(1000)
2016-04-12 19:19:56.236: sC -> Event Next(2000)
2016-04-12 19:19:56.236: flatMapSequence -> Event Next(2000)
Next(2000)
2016-04-12 19:19:57.335: sC -> Event Completed
2016-04-12 19:19:57.336: sC -> disposed
2016-04-12 19:19:58.926: sequence -> Event Completed
2016-04-12 19:19:58.926: flatMapSequence -> Event Completed
Completed
2016-04-12 19:19:58.926: sequence -> disposed

Первый пример показывает, что т.к. Observable успевают сгенерировать свои элементы, ко времени когда приходит очередь подписаться на новый Observable — это уже разрешено, поэтому в RO попадают элементы со всех Observable
А вот второй пример очень объемный, но позволят в подробностях наблюдать как происходят подписка/отписка и как это влияет на генерацию элементов

flatMapLatest

Каждый элемент SO превращается в отдельный Observable. Изначально подписываемся на O1, его элементы зеркально генерируются в RO. Как только из SO выпускается очередной элемент и на его основе генерируется очередной Observable — элементы предыдущего Observable отбрасываются, т.к. происходит отписка. Таким образом в RO — элементы из последнего генерированного Observable

example("flatMapLatest") {
    let sequence:Observable = Observable.of(10, 20, 30)
    let flatMapSequence = sequence.flatMapLatest{val in Observable.of(0, 1, 2)
        .map{"\($0) - \(val)"
        }
    }
    flatMapSequence.subscribe { e in
        print(e)
    }
}

example("flatMapLatest with delay") {
    let subjectA = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(10))
        }
        delay(3) {
            observer.on(.Next(20))
        }
        delay(5) {
            observer.onCompleted()
        }
        return NopDisposable.instance
        }.debug("sA")
    let subjectB = Observable.create{ observer in
        delay(0) {
            observer.on(.Next(100))
        }
        delay(1) {
            observer.on(.Next(200))
        }
        delay(2) {
            observer.onCompleted()
        }
        return NopDisposabl
    
            
© Habrahabr.ru