server-session-clientstreaming.swift 1.7 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253
  1. // {{ method.name }} (Client Streaming)
  2. public class {{ .|session:protoFile,service,method }} {
  3. private var handler : gRPC.Handler
  4. private var provider : {{ .|provider:protoFile,service }}
  5. /// Create a session.
  6. fileprivate init(handler:gRPC.Handler, provider: {{ .|provider:protoFile,service }}) {
  7. self.handler = handler
  8. self.provider = provider
  9. }
  10. /// Receive a message. Blocks until a message is received or the client closes the connection.
  11. public func Receive() throws -> {{ method|input }} {
  12. let done = NSCondition()
  13. var requestMessage : {{ method|input }}?
  14. try self.handler.receiveMessage() {(requestData) in
  15. if let requestData = requestData {
  16. requestMessage = try? {{ method|input }}(protobuf:requestData)
  17. }
  18. done.lock()
  19. done.signal()
  20. done.unlock()
  21. }
  22. done.lock()
  23. done.wait()
  24. done.unlock()
  25. if requestMessage == nil {
  26. throw {{ .|servererror:protoFile,service }}.endOfStream
  27. }
  28. return requestMessage!
  29. }
  30. /// Send a response and close the connection.
  31. public func SendAndClose(_ response: {{ method|output }}) throws {
  32. try self.handler.sendResponse(message:response.serializeProtobuf(),
  33. statusCode: 0,
  34. statusMessage: "OK",
  35. trailingMetadata: Metadata())
  36. }
  37. /// Run the session. Internal.
  38. fileprivate func run(queue:DispatchQueue) throws {
  39. try self.handler.sendMetadata(initialMetadata:Metadata()) {
  40. queue.async {
  41. do {
  42. try self.provider.collect(session:self)
  43. } catch (let error) {
  44. print("error \(error)")
  45. }
  46. }
  47. }
  48. }
  49. }