RxSwift: повторить (завершено) поток

Question

Предположим, у меня есть кнопка, которую можно использовать для запуска и остановки (переключения) действия.

let toggleStream: Observable<Bool> = toggleBtn.rx.tap.scan(false) { state, _ in !state }

У меня есть другой поток, который непрерывно излучает целые числа.

let emitter = Observable<Int>.interval(2.0, scheduler: timerScheduler)

Теперь я хочу использовать поток переключения, чтобы запустить и остановить излучение второго потока. Это мой подход:

Observable.combineLatest(toggleStream, emitter) { shouldEmit, evt in
    return (shouldEmit, evt)
}.takeWhile{ (shouldEmit, evt:Int) in
    return shouldEmit == true
}.map {(_, evt) in
    return evt
}

Это прекрасно работает в первый раз. Я могу нажать кнопку и Observable начинает излучать его Ints, Также остановка работает. К сожалению, я не могу запустить его во второй раз, потому что поток завершен. Как я могу перезапустить / повторить / повторить это, когда пользователь снова переключает кнопку?

Answer 1

Вот как я это сделал на детской площадке. Вы должны быть в состоянии экстраполировать:

//: Playground - noun: a place where people can play

import RxSwift

let toggleButton = PublishSubject<Void>()
let toggleStream: Observable<Bool> = toggleButton
    .scan(false) { state, _ in !state }
    .debug()
    .shareReplayLatestWhileConnected()

let emit = toggleStream
    .filter { $0 }
    .flatMapLatest { _ in
        Observable<Int>.interval(2.0, scheduler: MainScheduler.instance)
            .takeUntil(toggleStream.filter { !$0 })
    }


_ = emit.subscribe( {
    print($0)
})

toggleButton.onNext()
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 5.0) {
    toggleButton.onNext()
}
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 11.0) {
    toggleButton.onNext()
}
DispatchQueue.main.asyncAfter(deadline: DispatchTime.now() + 17.0) {
    toggleButton.onNext()
}

import PlaygroundSupport
PlaygroundPage.current.needsIndefiniteExecution = true