Home Articles

RxSwift flatMapLatest没有处理以前的observable

Asked
Viewed 1172 times
0

我正在使用RxSwift Playground:https://github.com/ReactiveX/RxSwift

我实现了以下示例代码来建模一个简单的异步任务,将一个Observable of Int转换为一个Observable of String:

let pubishSubject = PublishSubject<Int>()
pubishSubject.asObservable()
    .debug("before")
    .flatMap ({ (value) -> Observable<String> in
        let task: Observable<String> = Observable.create { observer in
            DispatchQueue.main.asyncAfter(deadline: .now() + 1, execute: {
                print("Executed \(value)")
                observer.on(.next("Async \(value)"))
                observer.on(.completed)
            })
            return Disposables.create(with: {
                print("Disposed \(value)")
            })
        }
        return task
    })
    .debug("after")
    .subscribe()

pubishSubject.onNext(1)
pubishSubject.onNext(2)
pubishSubject.onNext(3)

调试输出:

2018-06-01 14:08:35.748: after -> subscribed
2018-06-01 14:08:35.749: before -> subscribed
2018-06-01 14:08:35.751: before -> Event next(1)
2018-06-01 14:08:35.753: before -> Event next(2)
2018-06-01 14:08:35.753: before -> Event next(3)
Executed 1
2018-06-01 14:08:36.785: after -> Event next(Async 1)
Disposed 1
Executed 2
2018-06-01 14:08:36.786: after -> Event next(Async 2)
Disposed 2
Executed 3
2018-06-01 14:08:36.787: after -> Event next(Async 3)
Disposed 3

我需要完成所有异步任务才能继续,这完全正常,但我只需要订阅最新的结果(Async 3) .

我不能使用flatMapLastest运算符,因为它会取消/处理以前的异步调用 .

我不知道如何实现这一点,我觉得完全被阻止,也许我可以使用一个简单的RxSwift运算符,或者我需要以某种方式拆分序列 .

我希望有人知道如何实现这一点,并能帮助我,非常感谢你 .

编辑:对我来说,它看起来更像是一个队列实现,只要在PublishSubject上发布新的Int值并且仍然有Async Tasks运行,它应该跳过所有中间结果并返回基于最后一次发出的Int的最后结果

1 Answer

  • 0

    我在RxJava中有一个解决方案 . 转换为RxSwift应该很简单 .

    使用 switchMap() 运算符连续切换到最新发射 . 为了完成先前的排放,您需要订阅它们,以及发出值 . 为此,请使用 publish() 运算符:

    @Test
    public void testSwitchMap() throws Exception {
        PublishSubject<Integer> source = PublishSubject.create();
    
        source
                .doOnNext( v -> logger.debug( "emission: {}", v ) )
                .switchMap( v -> Observable.just( v )
                        .delay( 100, TimeUnit.MILLISECONDS )
                        .publish( s -> {
                            s.subscribe();
                            return s;
                        } ) )
                .subscribe( v -> logger.debug( "end result {}", v ) );
    
        source.onNext( Integer.valueOf( 1 ) );
        source.onNext( Integer.valueOf( 2 ) );
        source.onNext( Integer.valueOf( 3 ) );
        Thread.sleep( 1_000 );
    }
    

Related