每次发射的映射——SwitchMap 保证至少 1 个发射/ConcatMap 混合？

Question

我想知道如何在 RX 中做到这一点。

T他的实际用例是LowerLevelEvent(val userId: String)到HigherLevelEvent(val user: User)的映射，其中User是由observable提供的，所以它可以发出n次，所以例子输出

LowerLevelEvent1(abc) -> HigherLevelEvent1(userAbc(nameVariation1)
LowerLevelEvent2(abc) -> HigherLevelEvent2(userAbc(nameVariation1)
LowerLevelEvent3(abc) -> HigherLevelEvent3(userAbc(nameVariation1)
LowerLevelEvent4(abc) -> HigherLevelEvent4(userAbc(nameVariation1)
                         HigherLevelEvent4(userAbc(nameVariation2)
                         HigherLevelEvent4(userAbc(nameVariation3)

所以我天真的解决方案是使用 combineLatest。因此，虽然 userId 未更改，但用户 observable 已订阅，即当新的 lowerLevelEmits 及其 userId 未更改时未重新订阅

val _lowerLevelEventObservable: Observable<LowerLevelEvent> = lowerLevelEventObservable
                        .replayingShare()

val _higherLevelEventObservable: Observable<HigherLevelEvent> = Observables
                        .combineLatest(
                _lowerLevelEventObservable, 
                _lowerLevelEventObservable
                            .map { it.userId }
                            .distinctUntilChanged()
                            .switchMap { userRepository.findByIdObservable(it)
            ) { lowerLevelEvent, user -> createHigherLevelInstance... }

但是，这存在故障问题，因为 combineLatest 中的两个源都来自同一个 observable。

然后我想到了

lowerLevelObservable.
   .switchMap { lowerLevelEvent ->
      userRepository.findByIdObservable(lowerLevelEvent.userId)
          .map { user -> createHigherLevelInstance... }
   }

但是，如果 lowerLevelObservable 快速发射，这可能会中断，并且由于用户 observable 可能需要一些时间，因此可以跳过给定的 lowerLevelX 事件，而我不能这样做。它还重新订阅用户可观察的每个发射，这是浪费的，因为它很可能不会改变

那么，也许是 concatMap？这存在用户可观察对象未完成的问题，因此 concatMap 不起作用。

有人知道吗？

非常感谢

// 说明： 基本上它的 A 变体 (A1, A2..) 到 A' 变体 (A1', A2'..) 的映射，同时将查询对象附加到它，其中查询是可观察的，因此它可能会在映射后重新发出，所以AX' 需要重新发送新的查询结果。但是查询很冷，没有完成

例如A1(1) -> A1'(user1), A2(1) -> A2'(user1), A3(1) -> A3'(user1) -- 现在有人在应用程序的其他地方更改了 user1，所以下一个发出是A3'(user1')

Answer 1

根据您制作的 cmets，以下内容可以在 RxSwift 中使用。我不知道如何将它翻译成 RxJava。不过老实说，我认为这里存在对 Rx 的根本性误用。祝你好运。

它是如何工作的：如果允许订阅它就会订阅，否则它会将事件添加到缓冲区以供以后使用。如果当前未订阅内部事件，或者当前订阅的内部 Observable 已发出元素，则允许订阅。

警告：它不能正确处理完成。我会把它留给你作为练习。

func example(lowerLevelEventObservable: Observable<LowerLevelEvent>, userRepository: UserRepository) {
    let higherLevelEventObservable = lowerLevelEventObservable
        .flatMapAtLeastOnce { event in // RxSwift's switchLatest I think.
            Observable.combineLatest(
                Observable.just(event),
                userRepository.findByIdObservable(event.userId),
                resultSelector: { (lowLevelEvent: $0, user: $1) }
            )
        }
        .map { createHigherLevelInstance($0.lowLevelEvent, $0.user) }

    // use higherLevelEventObservable
}

extension ObservableType {
    func flatMapAtLeastOnce<U>(from fn: @escaping (E) -> Observable<U>) -> Observable<U> {
        return Observable.create { observer in
            let disposables = CompositeDisposable()
            var nexts: [E] = []
            var disposeKey: CompositeDisposable.DisposeKey?
            var isAllowedToSubscribe = true
            let lock = NSRecursiveLock()
            func nextSubscription() {
                isAllowedToSubscribe = true
                if !nexts.isEmpty {
                    let e = nexts[0]
                    nexts.remove(at: 0)
                    subscribeToInner(e)
                }
            }

            func subscribeToInner(_ element: E) {
                isAllowedToSubscribe = false
                if let key = disposeKey {
                    disposables.remove(for: key)
                }
                let disposable = fn(element).subscribe { innerEvent in
                    lock.lock(); defer { lock.unlock() }
                    switch innerEvent {
                    case .next:
                        observer.on(innerEvent)
                        nextSubscription()
                    case .error:
                        observer.on(innerEvent)
                    case .completed:
                        nextSubscription()
                    }
                }
                disposeKey = disposables.insert(disposable)
            }

            let disposable = self.subscribe { event in
                lock.lock(); defer { lock.unlock() }
                switch event {
                case let .next(element):
                    if isAllowedToSubscribe == true {
                        subscribeToInner(element)
                    }
                    else {
                        nexts.append(element)
                    }
                case let .error(error):
                    observer.onError(error)
                case .completed:
                    observer.onCompleted()
                }
            }
            _ = disposables.insert(disposable)
            return disposables
        }
    }
}