server-session-clientstreaming.swift 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. // {{ method|methodDescriptorName }} (Client Streaming)
  2. {{ access }} final 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. requestMessage = try? {{ method|input }}(serializedData:requestData)
  16. }
  17. sem.signal()
  18. }
  19. _ = sem.wait(timeout: DispatchTime.distantFuture)
  20. if requestMessage == nil {
  21. throw {{ .|servererror:file,service }}.endOfStream
  22. }
  23. return requestMessage!
  24. }
  25. /// Send a response and close the connection.
  26. {{ access }} func sendAndClose(_ response: {{ method|output }}) throws {
  27. try self.handler.sendResponse(message:response.serializedData(),
  28. statusCode:self.statusCode,
  29. statusMessage:self.statusMessage,
  30. trailingMetadata:self.trailingMetadata)
  31. }
  32. /// Run the session. Internal.
  33. fileprivate func run(queue:DispatchQueue) throws {
  34. try self.handler.sendMetadata(initialMetadata:initialMetadata) {
  35. queue.async {
  36. do {
  37. try self.provider.{{ method|methodDescriptorName|lowercase }}(session:self)
  38. } catch (let error) {
  39. print("error \(error)")
  40. }
  41. }
  42. }
  43. }
  44. }