RxSwift - call sequence of a function after Mapping

Question

I'm trying to call a function after a successful completion of a mapping. The observable is created from an existing stream by mapping. Inside the mapping block, I subscribe to get the output which is mapped to the final type. With the updated stream, I'm looking at calling the function. But the function gets called before I subscribe & map it to the updated stream.

Here is a sample code:

let updatedDataStream = dataStream
            .observeOn(MainScheduler.instance)
            .map { (data) -> ViewData? in

            guard var unknownData = data as? UnknownDataType,
                let location = unknownData.dataPoints.first?.location else  {
                    return data
                }

                let dataModel = DataModel(x: location.x, y: location.y)

                self.dataStore.localStore?
                    .entity(forID: DataModel.entityID)
                    .filter({$0?.status == .Success})
                    .take(1)
                    .unwrap()
                    .subscribe(onNext: { (localDataModel) in

                        guard let value = localDataModel.dataValue, value.count > 0 else {
                            return
                        }

                        unknownData.value = value

                    }).disposed(by: self.disposeBag)
                return unknownData
        }

        self.attachUpdatedData(updatedDataStream)

attachUpdatedData(_:) is expected to be called only after we retrieve the updatedDataStream. With the above implementation attachUpdatedData(_:) is called first before the data is mapped to a new value. Can someone point me in the right direction?

Answer 1

flatMap

Probably what you could use is flatMap instead of map

let updatedDataStream = dataStream
    .observeOn(MainScheduler.instance)
    .flatMap /* or flatMapLatest */ { [weak self] /* avoid memory leak */ data -> ViewData? in
      guard var unknownData = data as? UnknownDataType,
                let location = unknownData.dataPoints.first?.location,
                let self = self else
      { return .just(data) }

      let dataModel = DataModel(x: location.x, y: location.y)
      return self.dataStore.localStore?
                    .entity(forID: DataModel.entityID)
                    .filter({$0?.status == .Success})
                    .take(1)
                    .unwrap()
                    .map { localDataModel in
                       guard let value = localDataModel.dataValue,
                             value.count > 0 else { 
                       return .just(unknownData)
                       }
                       unknownData.value = value
                       return unknownData
                     }
       }
       self.attachUpdatedData(updatedDataStream)