reduce-net.swift

/// Using `blockURLComponent`, requests relenvant data from FRING API for the given blocks. Before
/// requesting, it will check the `fetchedBlockIds` set, attempting to avoid requesting duplicate 
/// information. After data is fetched, `processOperation` will be called.
func requestData(for blockIds: [(MKShape, String)]) -> Observable<[[String: Any]]> {
    assert(parse != nil, "parse closure was never set; this is a requirement of use")
    
    let rv = PublishSubject<[[String: Any]]>()
    
    // bail if we already have all block data
    let filteredBlockIds = blockIds
        .flatMap { _, blockId -> String? in
            if self.fetchedBlockIds.contains(blockId) {
                return nil
            } else {
                return blockId
            }
        }
    
    guard filteredBlockIds.count > 0 else {
        rv.onCompleted()
        return rv
    }
    
    let bodies = prepareHttpBodies(from: filteredBlockIds)
    
    if bodies.count == 1 {
      // not relevant
      // ... 
    } else if bodies.count > 1 {
        queue.async { [rv] in
            let group = DispatchGroup()
            let innerRv = PublishSubject<[[String: Any]]>()
            
            let innerDisposable = innerRv
                .reduce([], accumulator: +)
                .observeOn(MainScheduler.instance)
                .subscribe(onNext: { final in
                    Logger.verbose("processing \(final.count) pieces of data")
                    self.process(data: final)
                    rv.onNext(final)
                    rv.onCompleted()
                })
            
            let nextQueue = DispatchQueue(label: "com.housecanary.HouseCanary:BlockProcessor:Next")
            bodies.forEach { self.makeGroupedRequest(with: $0, subject: innerRv, group: group, nextQueue: nextQueue) }
            
            group.wait()
            innerRv.onCompleted()
            
            innerDisposable.dispose()
            innerRv.dispose()
        }
        
    }
    
    return rv
}