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。

我可以做一些可怕的黑客像下​​面，但我真的不想！

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个值（通常为元组），然后该事情是向下传播的链解决。我写这为扩展RxSwfit，它的工作原理如下：

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

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

此外，写这个扩展方法是不那么容易。我基本上重新实现FlatMap从头开始通过复制/粘贴一个在MiniRxSwift并修改它。如果我们在写这个扩展，有没有更好的方式来实现它使用RxSwift结构？

Answer 1

你可以在你的establish功能。它可能是这样的：

static func establish(address:String) -> Observable<Connection> {
    return Observable.create { observer in
        //create connection
        observer.onNext(connection)
        observer.onCompleted()

        return AnonymousDisposable { connection.close() }
    }
}

当被设置你的observer，它也将关闭连接。 我认为这是这里最好的选择之一。

但是，我们可以通过降低使用combineLatest或flatMap或其它功能虽然链的连接。但是，这将是乏味的:)

---

回复猎户座爱德华兹

不敢肯定这会工作;在我的例子一次性从未实际部署，我们要尽快关闭连接操作完成，而不是

嗯，我敢肯定这是RxSwift的人会怎么做。 如果你有你的结果，要么completed或error了，你处理的观察者。如果你想要做一遍，然后你re-subscribe它。

你可以为了处理它使用这些功能之一：

• 使用take家庭：take(1)，takeUntil ...
• 呼叫dispose()上观察，当你完成。 （不鼓励）
• 使用debug()看到一个观察者处置时
• ...

Answer 2

有两种方法可以做到你想要什么“使用RxSwift结构。”

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

或者，如果你不介意的话插入到地图你可以：

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

需要注意的是combineLatest和map都是从一开始的库。