server-session-bidistreaming.swift 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960
  1. // fully streaming
  2. public class {{ .|session:protoFile,service,method }} {
  3. var handler : gRPC.Handler
  4. var provider : {{ .|provider:protoFile,service }}
  5. fileprivate init(handler:gRPC.Handler, provider: {{ .|provider:protoFile,service }}) {
  6. self.handler = handler
  7. self.provider = provider
  8. }
  9. public func Receive() throws -> {{ method|input }} {
  10. let done = NSCondition()
  11. var requestMessage : {{ method|input }}?
  12. try self.handler.receiveMessage() {(requestData) in
  13. if let requestData = requestData {
  14. requestMessage = try! {{ method|input }}(protobuf:requestData)
  15. }
  16. done.lock()
  17. done.signal()
  18. done.unlock()
  19. }
  20. done.lock()
  21. done.wait()
  22. done.unlock()
  23. if requestMessage == nil {
  24. throw {{ .|servererror:protoFile,service }}.endOfStream
  25. }
  26. return requestMessage!
  27. }
  28. public func Send(_ response: {{ method|output }}) throws {
  29. try handler.sendResponse(message:response.serializeProtobuf()) {}
  30. }
  31. public func Close() {
  32. let done = NSCondition()
  33. try! self.handler.sendStatus(statusCode: 0,
  34. statusMessage: "OK",
  35. trailingMetadata: Metadata()) {
  36. done.lock()
  37. done.signal()
  38. done.unlock()
  39. }
  40. done.lock()
  41. done.wait()
  42. done.unlock()
  43. }
  44. fileprivate func run(queue:DispatchQueue) {
  45. do {
  46. try self.handler.sendMetadata(initialMetadata:Metadata()) {
  47. queue.async {
  48. try! self.provider.update(session:self)
  49. }
  50. }
  51. } catch (let callError) {
  52. print("grpc error: \(callError)")
  53. }
  54. }
  55. }