server-session-bidistreaming.swift 2.1 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. // {{ method|methodDescriptorName }} (Bidirectional Streaming)
  2. {{ access }} class {{ .|session:file,service,method }} : {{ .|service:file,service }}Session {
  3. private var provider : {{ .|provider:file,service }}
  4. /// Create a session.
  5. fileprivate init(handler:gRPC.Handler, provider: {{ .|provider:file,service }}) {
  6. self.provider = provider
  7. super.init(handler:handler)
  8. }
  9. /// Receive a message. Blocks until a message is received or the client closes the connection.
  10. {{ access }} func receive() throws -> {{ method|input }} {
  11. let sem = DispatchSemaphore(value: 0)
  12. var requestMessage : {{ method|input }}?
  13. try self.handler.receiveMessage() {(requestData) in
  14. if let requestData = requestData {
  15. do {
  16. requestMessage = try {{ method|input }}(serializedData:requestData)
  17. } catch (let error) {
  18. print("error \(error)")
  19. }
  20. }
  21. sem.signal()
  22. }
  23. _ = sem.wait(timeout: DispatchTime.distantFuture)
  24. if let requestMessage = requestMessage {
  25. return requestMessage
  26. } else {
  27. throw {{ .|servererror:file,service }}.endOfStream
  28. }
  29. }
  30. /// Send a message. Nonblocking.
  31. {{ access }} func send(_ response: {{ method|output }}, completion: @escaping ()->()) throws {
  32. try handler.sendResponse(message:response.serializedData()) {completion()}
  33. }
  34. /// Close a connection. Blocks until the connection is closed.
  35. {{ access }} func close() throws {
  36. let sem = DispatchSemaphore(value: 0)
  37. try self.handler.sendStatus(statusCode:self.statusCode,
  38. statusMessage:self.statusMessage,
  39. trailingMetadata:self.trailingMetadata) {
  40. sem.signal()
  41. }
  42. _ = sem.wait(timeout: DispatchTime.distantFuture)
  43. }
  44. /// Run the session. Internal.
  45. fileprivate func run(queue:DispatchQueue) throws {
  46. try self.handler.sendMetadata(initialMetadata:initialMetadata) {
  47. queue.async {
  48. do {
  49. try self.provider.{{ method|methodDescriptorName|lowercase }}(session:self)
  50. } catch (let error) {
  51. print("error \(error)")
  52. }
  53. }
  54. }
  55. }
  56. }