swift - RxSwift 通过 flatMaps 链传播一个值

Question

所以我有一些 RxSwift 代码，我想在其中执行一系列异步操作，所有操作都使用可观察对象组成。 flatMap 是执行此操作的方法，并且效果很好，但是它似乎无法将变量传递到我能弄清楚的链中。一些伪代码最好地说明了这一点

假设有3个函数

class Connection {
    static func establish(address:String) -> Observable<Connection>
    func sendData(data:String) -> Observable<Int> // num bytes written or something
    func close() -> Observable<Void>
}

我想在链中调用它们，以便我们连接、发送，然后关闭。像这样

Connection.establish(host)
    .flatMap{ connection in connection.sendData("foo") }
    .flatMap{ numBytes in ????.close() }
    .subscribeNext{ /* all done */ }

问题是 flatMap 没有将它的输入参数传递到链下，因此传递给 subscribeNext 的闭包无法访问 connection 对象，因此它不能调用 close。

我可以像下面那样做一些可怕的 hack，但我真的不想这样做!

var connection:Connection?
Connection.establish(host)
    .flatMap{ c in 
        connection = c
        return c.sendData("foo") 
    }
    .flatMap{ numBytes in connection!.close() }
    .subscribeNext{ /* all done */ }

在 Rx 的 C# 版本中，这是通过对 SelectMany 的重载来解决的，它采用第二个闭包，将 2 个值(通常合并到一个元组中)然后 that 事物沿着链传播。我把它写成 RxSwfit 的扩展，它的工作原理如下:

Connection.establish(host)
    .flatMap(
        { connection in connection.sendData("foo") },
        combine: { ($0, $1) }) // tupleify
    .flatMap{ (connection, numbytes) in connection.close() }
    .subscribeNext{ /* all done */ }

一切都很好，但我的主要问题是 - 是否有更好的方法来做到这一点，目前内置在 RxSwift 中？

此外，编写这个扩展方法并不简单也不容易。我基本上是通过复制/粘贴 MiniRxSwift 中的 FlatMap 从头开始​​重新实现的。并修改它。如果我们必须编写这个扩展，是否有更好的方法使用 RxSwift 构造来实现它？

Answer 1

有两种方法可以“使用 RxSwift 结构”来做你想做的事情。

Connection.establish(host)
    .flatMap { Observable.combineLatest(Observable.just($0), $0.sendData("foo")) }
    .flatMap { connection, _ in connection.close() }
    .subscribe(onNext: { /* all done */ })

或者如果您不介意插入 map ，您可以:

Connection.establish(host)
    .flatMap { connection in
        connection.sendData("foo").map { (connection, $0) }
    }
    .flatMap { connection, _ in connection.close() }
    .subscribe(onNext: { /* all done */ })

请注意，combineLatest 和 map 从一开始就在库中。