RxJava 3 映射器

如何解决RxJava 3 映射器

我在我的项目中使用 RxJava3 并且我无法编写请求，我已经折腾了几个小时。

我有 List 并且有一个函数，它通过 Stream 返回一个 Single>，我想得到一个 Single

fun getMockTopics(streamId: Long): Single<List<Topic>> {
        return Single.just(listof(Topic(1,"")))
    }

typealias SteamTopics = Map<Stream,List<Topic>>

override fun getTopics(streams: List<Stream?>): Single<SteamTopics> {
        return Observable.fromCallable { streams.filterNotNull() }.flatMapIterable { it }
            .map { stream ->
                Pair(stream,getMockTopics(streamId = stream.streamId))
            }.flatMap {
                Todo("???")
            }
            .toMap({ it.first },{ it.second })
    }

解决方法

fun getMockTopics(streamId: Long): Observable<List<Topic>> {
    return Observable.just(listOf(Topic(1,"")))
}

override fun getTopics(streams: List<Stream?>): Single<SteamTopics> {
    return Observable.fromCallable { streams.filterNotNull() }.flatMapIterable { it }
        .flatMap({ stream ->
            getMockTopics(streamId = stream.streamId)
        },{ stream,topics ->
            Pair(stream,topics)
        }).toMap({ it.first },{ it.second })
}