All emitter

AllEmitter.swift

import RxSwift
import RxCocoa

func ignoreNil<A>(x: A?) -> Observable<A> {
    return x.map { Observable.just($0) } ?? Observable.empty()
}

// If there are values return from stream A, B and C then I emit an event on stream X (boolean)
func allEmitedValue<A, B, C>(_ a: Observable<A?>, _ b: Observable<B?>, _ c: Observable<C?>) -> Observable<Bool> {
    Observable.combineLatest(a, b, c) { allContainValue($0, $1, $2)  }
}

// Whenever there is true value returns from streamX, I want to take a take a snapshot of stream A, B and C and do something
func snapshot<A, B, C>(_ a: Observable<A?>, _ b: Observable<B?>, _ c: Observable<C?>) -> Observable<(A, B, C)> {
    Observable.combineLatest(a, b, c)
        .compactMap{ allContainValue($0.0, $0.1, $0.2) ? ($0.0!, $0.1!, $0.2!) : nil }
}

func allContainValue<A, B, C>(_ a: A?, _ b: B?, _ c: C?) -> Bool {
    a != nil && b != nil && c != nil
}

class parent {
    
    let bag = DisposeBag()
    
    let streamA: BehaviorRelay<String?> = BehaviorRelay(value: nil)
    let streamB: BehaviorRelay<String?> = BehaviorRelay(value: nil)
    let streamC: BehaviorRelay<String?> = BehaviorRelay(value: nil)
    var moduleVisibility: Observable<Bool>!
    
    init() {
        moduleVisibility = allEmitedValue(
            streamA.asObservable(),
            streamB.asObservable(),
            streamC.asObservable()
        ).distinctUntilChanged()
        moduleVisibility
        .observe(on: MainScheduler.instance)
        .subscribe(onNext: { [weak self] v in
            guard let self = self else { return }
            // now it gets called.
            print("allEmitedValue \(v)")
            
        })
        .disposed(by: bag)
        
        
        
    }
}

class childClass: parent {
    
    
    override init() {
        super.init()
        moduleVisibility
            .filter({ $0 == true })
            .withLatestFrom(
                snapshot(
                    streamA.asObservable(),
                    streamB.asObservable(),
                    streamC.asObservable()
                )
            )
        
        .observe(on: MainScheduler.instance)
        .subscribe(onNext: { valueA, valueB, valueC in
            // now it gets called.
            print("\(valueA) \(valueB) \(valueC)")
        })
        .disposed(by: bag)
    }
}


var v = childClass()

v.streamA.accept("A")
v.streamB.accept("B")
v.streamC.accept("C")


v.streamA.accept(nil)
v.streamB.accept(nil)
v.streamC.accept(nil)

v.streamA.accept("E")
v.streamB.accept("F")
v.streamC.accept("G")

/*
allEmitedValue false
allEmitedValue true
A B C
allEmitedValue false
allEmitedValue true
E F G

*/