UnaryResponseCallContext.swift 6.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170
  1. /*
  2. * Copyright 2019, gRPC Authors All rights reserved.
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. import Foundation
  17. import Logging
  18. import NIO
  19. import NIOHPACK
  20. import NIOHTTP1
  21. import SwiftProtobuf
  22. /// Abstract base class exposing a method that exposes a promise for the RPC response.
  23. ///
  24. /// - When `responsePromise` is fulfilled, the call is closed and the provided response transmitted with status `responseStatus` (`.ok` by default).
  25. /// - If `statusPromise` is failed and the error is of type `GRPCStatusTransformable`,
  26. /// the result of `error.asGRPCStatus()` will be returned to the client.
  27. /// - If `error.asGRPCStatus()` is not available, `GRPCStatus.processingError` is returned to the client.
  28. ///
  29. /// For unary calls, the response is not actually provided by fulfilling `responsePromise`, but instead by completing
  30. /// the future returned by `UnaryCallHandler.EventObserver`.
  31. open class UnaryResponseCallContext<ResponsePayload>: ServerCallContextBase, StatusOnlyCallContext {
  32. typealias WrappedResponse = GRPCServerResponsePart<ResponsePayload>
  33. public let responsePromise: EventLoopPromise<ResponsePayload>
  34. public var responseStatus: GRPCStatus = .ok
  35. public convenience init(
  36. eventLoop: EventLoop,
  37. headers: HPACKHeaders,
  38. logger: Logger,
  39. userInfo: UserInfo = UserInfo()
  40. ) {
  41. self.init(eventLoop: eventLoop, headers: headers, logger: logger, userInfoRef: .init(userInfo))
  42. }
  43. @inlinable
  44. override internal init(
  45. eventLoop: EventLoop,
  46. headers: HPACKHeaders,
  47. logger: Logger,
  48. userInfoRef: Ref<UserInfo>
  49. ) {
  50. self.responsePromise = eventLoop.makePromise()
  51. super.init(eventLoop: eventLoop, headers: headers, logger: logger, userInfoRef: userInfoRef)
  52. }
  53. @available(*, deprecated, renamed: "init(eventLoop:headers:logger:userInfo:)")
  54. override public init(eventLoop: EventLoop, request: HTTPRequestHead, logger: Logger) {
  55. self.responsePromise = eventLoop.makePromise()
  56. super.init(eventLoop: eventLoop, request: request, logger: logger)
  57. }
  58. }
  59. /// Protocol variant of `UnaryResponseCallContext` that only exposes the `responseStatus` and `trailingMetadata`
  60. /// fields, but not `responsePromise`.
  61. ///
  62. /// Motivation: `UnaryCallHandler` already asks the call handler return an `EventLoopFuture<ResponsePayload>` which
  63. /// is automatically cascaded into `UnaryResponseCallContext.responsePromise`, so that promise does not (and should not)
  64. /// be fulfilled by the user.
  65. ///
  66. /// We can use a protocol (instead of an abstract base class) here because removing the generic `responsePromise` field
  67. /// lets us avoid associated-type requirements on the protocol.
  68. public protocol StatusOnlyCallContext: ServerCallContext {
  69. var responseStatus: GRPCStatus { get set }
  70. var trailers: HPACKHeaders { get set }
  71. }
  72. extension StatusOnlyCallContext {
  73. @available(*, deprecated, renamed: "trailers")
  74. public var trailingMetadata: HTTPHeaders {
  75. get {
  76. return HTTPHeaders(self.trailers.map { ($0.name, $0.value) })
  77. }
  78. set {
  79. self.trailers = HPACKHeaders(httpHeaders: newValue)
  80. }
  81. }
  82. }
  83. /// Concrete implementation of `UnaryResponseCallContext` used by our generated code.
  84. open class UnaryResponseCallContextImpl<ResponsePayload>: UnaryResponseCallContext<ResponsePayload> {
  85. public let channel: Channel
  86. /// - Parameters:
  87. /// - channel: The NIO channel the call is handled on.
  88. /// - headers: The headers provided with this call.
  89. /// - errorDelegate: Provides a means for transforming response promise failures to `GRPCStatusTransformable` before
  90. /// sending them to the client.
  91. /// - logger: A logger.
  92. public init(
  93. channel: Channel,
  94. headers: HPACKHeaders,
  95. errorDelegate: ServerErrorDelegate?,
  96. logger: Logger
  97. ) {
  98. self.channel = channel
  99. super.init(
  100. eventLoop: channel.eventLoop,
  101. headers: headers,
  102. logger: logger,
  103. userInfoRef: .init(UserInfo())
  104. )
  105. self.responsePromise.futureResult.whenComplete { [self, weak errorDelegate] result in
  106. switch result {
  107. case let .success(message):
  108. self.handleResponse(message)
  109. case let .failure(error):
  110. self.handleError(error, delegate: errorDelegate)
  111. }
  112. }
  113. }
  114. /// Handle the response from the service provider.
  115. private func handleResponse(_ response: ResponsePayload) {
  116. self.channel.write(
  117. self.wrap(.message(response, .init(compress: self.compressionEnabled, flush: false))),
  118. promise: nil
  119. )
  120. self.channel.writeAndFlush(
  121. self.wrap(.end(self.responseStatus, self.trailers)),
  122. promise: nil
  123. )
  124. }
  125. /// Handle an error from the service provider.
  126. private func handleError(_ error: Error, delegate: ServerErrorDelegate?) {
  127. let (status, trailers) = self.processObserverError(error, delegate: delegate)
  128. self.channel.writeAndFlush(self.wrap(.end(status, trailers)), promise: nil)
  129. }
  130. /// Wrap the response part in a `NIOAny`. This is useful in order to avoid explicitly spelling
  131. /// out `NIOAny(WrappedResponse(...))`.
  132. private func wrap(_ response: WrappedResponse) -> NIOAny {
  133. return NIOAny(response)
  134. }
  135. @available(*, deprecated, renamed: "init(channel:headers:errorDelegate:logger:)")
  136. public convenience init(
  137. channel: Channel,
  138. request: HTTPRequestHead,
  139. errorDelegate: ServerErrorDelegate?,
  140. logger: Logger
  141. ) {
  142. self.init(
  143. channel: channel,
  144. headers: HPACKHeaders(httpHeaders: request.headers, normalizeHTTPHeaders: false),
  145. errorDelegate: errorDelegate,
  146. logger: logger
  147. )
  148. }
  149. }
  150. /// Concrete implementation of `UnaryResponseCallContext` used for testing.
  151. ///
  152. /// Only provided to make it clear in tests that no "real" implementation is used.
  153. open class UnaryResponseCallContextTestStub<ResponsePayload>: UnaryResponseCallContext<ResponsePayload> {}