RXSwift ObservableCollection with CombineLatest

Question

I am trying to implement something like this,

    let api1 = Observable.of(["documents"])    //Replace with observable to download docs
    let api2 = Observable.of(["applications"]) //Replace with observable to download apps
    let api3 = Observable.of(["videos"])       //Replace with observable to download videos

    Observable.combineLatest(api1, api2, api3){(docs, apps, videos) in
        return (docs, apps, videos)
    }.skipWhile{ (docs, apps, videos) in
        return docs.count == 0 && apps.count == 0 && videos.count == 0
    }.subscribe(onNext:{(docs, apps, videos) in

    })
    .disposed(by:disposeBag)

In my case, I am trying to create observables dynamically and add it to an array like this,

private var discoverObservables = [Observable<Any>]()


func loadDiscoverFeeds(){
      
        self.feeds.forEach({
            feed in
            switch feed.feedType{
            case "a":
                let observable = self.aObservable(url: feed.feedURL ?? "")
                self.discoverObservables.append(observable)
                break
            case "b":
                let observable = self.bObservable(url: feed.feedURL ?? "")
                self.discoverObservables.append(observable)
                break
            case "c":
                let observable = self.cObservable(url: feed.feedURL ?? "")
                self.discoverObservables.append(observable)
                break
            case "d" :
                let observable = self.dObservable(url: feed.feedURL ?? "")
                self.discoverObservables.append(observable)
                break
            default:
                break
            
            }
        })
}

    private func aObservable(url : String) -> Observable<A?>{
            return APIManager.shared.getA(url: url)
        }

    private func bObservable(url : String) -> Observable<B?>{
            return APIManager.shared.getB(url: url)
        }


    private func cObservable(url : String) -> Observable<C?>{
            return APIManager.shared.getC(url: url)
        }

But this is not working because discoverObservables array is expecting the value of Type Observable<Any> and I am trying to add Observable<A?>

How can I do this correctly, I want to make sure all the observables return data before I start processing the data.

Edit I am trying to load data from different sources before that is added to the view, basically, I have a collectionview, each section loads data from different API, I am trying to get all the required data from all sources before that is added to collection view.

Answer 1

Add the same protocol to A, B and C.

protocol YourProtocol {...}

class A: YourProtocol {...}
class B: YourProtocol {...}
class C: YourProtocol {...}

Then you can make :

private var discoverObservables = [Observable<YourProtocol>]() 

Answer 2

The first code block seems to be doing the job with one exception, the condition checks if all of the (docs, apps, videos) are empty, perhaps you wanted to use || instead of &&.

As for the second code block with an array, I did something that could help.

    struct A {}
    
    let observable1 = Observable.just(A())
    let observable2 = Observable.just(A())
    let observable3 = Observable.just(A())

    let observables: [Observable<A>] = [observable1, observable2, observable3]
    
    Observable.combineLatest(observables).skipWhile { (streams) -> Bool in
        streams.forEach {
            if $0.count == 0 { return true }
        }
        return false
    }.subscribe(...

This subscription will result with Observable<[A]>.

Answer 3

I'm going to specifically address this from your question: "I want to make sure all the observables return data before I start processing the data."

Strictly speaking, you probably don't want an Any structure. Better would be a protocol or enum. I see that other answers have addressed the protocol idea so I will use the enum idea:

enum EndpointResponse {
    case a(A?)
    case b(B?)
    // etc...
}

let responses = Observable.zip(
    feeds.map { (feed) -> Observable<EndpointResponse> in
        switch feed.feedType {
        case "a":
            return aObservable(url: feed.feedURL ?? "").map { EndpointResponse.a($0) }
        case "b":
            return bObservable(url: feed.feedURL ?? "").map { EndpointResponse.b($0) }
        default:
            fatalError()
        }
    }
)

The above responses observable will contain an array of all the responses once they have all emitted values. In other words, the zip operator will gather up all the responses from all the network calls and emit a single array containing all of them.

My Previous answer:

There really isn't a lot of information to go on in the question, but something like this answers the direct question you ask about converting an Observable<X> to an Observable<Any>...

let discoverObservables = Observable.zip(
    feeds.map { (feed) -> Observable<Any> in
        switch feed.feedType {
        case "a":
            return aObservable(url: feed.feedURL ?? "").map { $0 as Any }
        case "b":
            return bObservable(url: feed.feedURL ?? "").map { $0 as Any }
        case "c":
            return cObservable(url: feed.feedURL ?? "").map { $0 as Any }
        case "d":
            return dObservable(url: feed.feedURL ?? "").map { $0 as Any }
        default:
            break
        }
    }
)