简而言之
我们有一个移动应用程序，可以通过各种双向流将大量数据传输到服务器或从服务器传输。有时需要关闭流(例如，当应用程序处于后台时)。然后根据需要重新打开它们。有时发生这种情况时，会出现问题:
- 据我所知，流已启动并在设备端运行(所涉及的 GRPCProtocall 和 GRXWriter 的状态要么已启动，要么已暂停)
- 设备在流上发送数据正常(服务器接收数据)
- 服务器似乎可以正常将数据发送回设备(服务器的 Stream.Send 调用返回成功)
- 在设备上，从不调用流中接收到的数据的结果处理程序
更多详情
我们的代码在下面得到了极大的简化，但这应该能提供足够的细节来说明我们在做什么。双向流由 Switch 类管理:
class Switch {
/** The protocall over which we send and receive data */
var protocall: GRPCProtoCall?
/** The writer object that writes data to the protocall. */
var writer: GRXBufferedPipe?
/** A static GRPCProtoService as per the .proto */
static let service = APPDataService(host: Settings.grpcHost)
/** A response handler. APPData is the datatype defined by the .proto. */
func rpcResponse(done: Bool, response: APPData?, error: Error?) {
NSLog("Response received")
// Handle response...
}
func start() {
// Create a (new) instance of the writer
// (A writer cannot be used on multiple protocalls)
self.writer = GRXBufferedPipe()
// Setup the protocall
self.protocall = Switch.service.rpcToStream(withRequestWriter: self.writer!, eventHandler: self.rpcRespose(done:response:error:))
// Start the stream
self.protocall.start()
}
func stop() {
// Stop the writer if it is started.
if self.writer.state == .started || self.writer.state == .paused {
self.writer.finishWithError(nil)
}
// Stop the proto call if it is started
if self.protocall?.state == .started || self.protocall?.state == .paused {
protocall?.cancel()
}
self.protocall = nil
}
private var needsRestart: Bool {
if let protocall = self.protocall {
if protocall.state == .notStarted || protocall.state == .finished {
// protocall exists, but isn't running.
return true
} else if writer.state == .notStarted || writer.state == .finished {
// writer isn't running
return true
} else {
// protocall and writer are running
return false
}
} else {
// protocall doesn't exist.
return true
}
}
func restartIfNeeded() {
guard self.needsRestart else { return }
self.stop()
self.start()
}
func write(data: APPData) {
self.writer.writeValue(data)
}
}
就像我说的，大大简化了，但它显示了我们如何启动、停止和重新启动流，以及我们如何检查流是否健康。
当应用进入后台时，我们调用stop()。当它被前台化并且我们再次需要流时，我们调用 start()。我们定期调用 restartIfNeeded()，例如。当使用流的屏幕出现时。
正如我上面提到的，偶尔会发生的情况是我们的响应处理程序 (rpcResponse) 在服务器将数据写入流时停止调用。流似乎是健康的(服务器接收到我们写入的数据，并且 protocall.state 既不是 .notStarted 也不是 .finished)。但即使是响应处理程序第一行的日志也没有执行。
第一个问题:我们是否正确管理流，或者我们停止和重新启动流的方式是否容易出错？如果是这样，做这样的事情的正确方法是什么？
第二个问题:我们如何调试这个？我们能想到的所有我们可以查询状态的东西都告诉我们流已经启动并正在运行，但感觉 objc gRPC 库对我们隐藏了很多它的机制。有没有办法查看来自服务器的响应是否确实到达了我们，但未能触发我们的响应处理程序？
第三个问题:按照上面的代码，我们使用了库提供的GRXBufferedPipe。它的文档建议不要在生产中使用它，因为它没有后推机制。据我们了解，编写器仅用于以同步的、一次一个的方式将数据提供给 gRPC 核心，并且由于服务器可以很好地接收来 self 们的数据，我们认为这不是问题。我们错了吗？作者是否也参与将从服务器接收到的数据提供给我们的响应处理程序？ IE。如果写入器由于过载而崩溃，这是否会表现为从流中读取数据而不是写入数据的问题？
更新:问这个问题一年多后，我们终于在我们的服务器端代码中发现了一个导致客户端出现这种行为的死锁错误。流似乎挂起，因为客户端发送的通信没有被服务器处理，反之亦然，但流实际上是活跃的。接受的答案为如何管理这些双向流提供了很好的建议，我认为这仍然很有值(value)(它帮助了我们很多!)。但问题实际上是由于编程错误。
此外，对于遇到此类问题的任何人，可能值得调查一下您是否遇到了 this known issue当 iOS 更改其网络时， channel 会被静静地丢弃。 This readme提供了使用 Apple 的 CFStream API 而不是 TCP 套接字作为该问题的可能修复方法的说明。
最佳答案
First question: Are we managing the streams correctly, or is our way of stopping and restarting streams prone to errors? If so, what is the correct way of doing something like this?
从我查看您的代码可以看出，start() 函数似乎是正确的。在stop()函数中，不需要调用self.protocall的cancel()；调用将以前面的 self.writer.finishWithError(nil) 结束。
needsrestart() 是它变得有点困惑的地方。首先，您不应该自己轮询/设置 protocall 的状态。这种状态是自己改变的。其次，设置这些状态不会关闭您的流。它只会暂停一个 writer，如果应用程序在后台，暂停一个 writer 就像一个空操作。如果你想关闭一个流，你应该使用 finishWithError 来终止这个调用，并可能在需要的时候开始一个新的调用。
Second question: How do we debug this?
一种方法是打开 gRPC 日志(GRPC_TRACE 和 GRPC_VERBOSITY)。另一种方法是在 here 处设置断点其中 gRPC objc 库从服务器接收 gRPC 消息。
Third question: Is the writer also involved in feeding data received from server to our response handler?
没有。如果您创建一个缓冲管道并将其作为您的调用请求提供，它只会提供要发送到服务器的数据。接收路径由另一个编写器处理(实际上是您的 protocall 对象)。
我看不出有什么地方不鼓励在生产中使用 GRXBufferedPipe。此实用程序的已知缺点是，如果您暂停编写器但继续使用 writeWithValue 向其写入数据，您最终会缓冲大量数据而无法刷新它们，这可能会导致内存问题。
关于ios - 双向 gRPC 流有时会在停止和启动后停止处理响应，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/48944209/