EmbeddedGRPCChannel.swift 5.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181
  1. /*
  2. * Copyright 2020, 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 NIO
  17. import NIOHTTP2
  18. import Logging
  19. import SwiftProtobuf
  20. // This is currently intended for internal testing only.
  21. class EmbeddedGRPCChannel: GRPCChannel {
  22. let embeddedChannel: EmbeddedChannel
  23. let multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>
  24. let logger: Logger
  25. let scheme: String
  26. let authority: String
  27. let errorDelegate: ClientErrorDelegate?
  28. func close() -> EventLoopFuture<Void> {
  29. return embeddedChannel.close()
  30. }
  31. var eventLoop: EventLoop {
  32. return self.embeddedChannel.eventLoop
  33. }
  34. init(
  35. logger: Logger = Logger(label: "io.grpc", factory: { _ in SwiftLogNoOpLogHandler() }),
  36. errorDelegate: ClientErrorDelegate? = nil
  37. ) {
  38. let embeddedChannel = EmbeddedChannel()
  39. self.embeddedChannel = embeddedChannel
  40. self.logger = logger
  41. self.multiplexer = embeddedChannel.configureGRPCClient(
  42. errorDelegate: errorDelegate,
  43. logger: logger
  44. ).flatMap {
  45. embeddedChannel.pipeline.handler(type: HTTP2StreamMultiplexer.self)
  46. }
  47. self.scheme = "http"
  48. self.authority = "localhost"
  49. self.errorDelegate = errorDelegate
  50. }
  51. private func makeRequestHead(path: String, options: CallOptions) -> _GRPCRequestHead {
  52. return _GRPCRequestHead(
  53. scheme: self.scheme,
  54. path: path,
  55. host: self.authority,
  56. options: options,
  57. requestID: nil
  58. )
  59. }
  60. internal func makeUnaryCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  61. path: String,
  62. request: Request,
  63. callOptions: CallOptions
  64. ) -> UnaryCall<Request, Response> {
  65. let call = UnaryCall<Request, Response>.makeOnHTTP2Stream(
  66. multiplexer: self.multiplexer,
  67. serializer: ProtobufSerializer(),
  68. deserializer: ProtobufDeserializer(),
  69. callOptions: callOptions,
  70. errorDelegate: self.errorDelegate,
  71. logger: self.logger
  72. )
  73. call.send(self.makeRequestHead(path: path, options: callOptions), request: request)
  74. return call
  75. }
  76. internal func makeClientStreamingCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  77. path: String,
  78. callOptions: CallOptions
  79. ) -> ClientStreamingCall<Request, Response> {
  80. let call = ClientStreamingCall<Request, Response>.makeOnHTTP2Stream(
  81. multiplexer: self.multiplexer,
  82. serializer: ProtobufSerializer(),
  83. deserializer: ProtobufDeserializer(),
  84. callOptions: callOptions,
  85. errorDelegate: self.errorDelegate,
  86. logger: self.logger
  87. )
  88. call.sendHead(self.makeRequestHead(path: path, options: callOptions))
  89. return call
  90. }
  91. internal func makeServerStreamingCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  92. path: String,
  93. request: Request,
  94. callOptions: CallOptions,
  95. handler: @escaping (Response) -> Void
  96. ) -> ServerStreamingCall<Request, Response> {
  97. let call = ServerStreamingCall<Request, Response>.makeOnHTTP2Stream(
  98. multiplexer: self.multiplexer,
  99. serializer: ProtobufSerializer(),
  100. deserializer: ProtobufDeserializer(),
  101. callOptions: callOptions,
  102. errorDelegate: self.errorDelegate,
  103. logger: self.logger,
  104. responseHandler: handler
  105. )
  106. call.send(self.makeRequestHead(path: path, options: callOptions), request: request)
  107. return call
  108. }
  109. internal func makeBidirectionalStreamingCall<Request: SwiftProtobuf.Message, Response: SwiftProtobuf.Message>(
  110. path: String,
  111. callOptions: CallOptions,
  112. handler: @escaping (Response) -> Void
  113. ) -> BidirectionalStreamingCall<Request, Response> {
  114. let call = BidirectionalStreamingCall<Request, Response>.makeOnHTTP2Stream(
  115. multiplexer: self.multiplexer,
  116. serializer: ProtobufSerializer(),
  117. deserializer: ProtobufDeserializer(),
  118. callOptions: callOptions,
  119. errorDelegate: self.errorDelegate,
  120. logger: self.logger,
  121. responseHandler: handler
  122. )
  123. call.sendHead(self.makeRequestHead(path: path, options: callOptions))
  124. return call
  125. }
  126. }
  127. extension EmbeddedGRPCChannel {
  128. // We need these to conform to `GRPCChannel`. This class is internal and only used for tests so
  129. // it's okay that they're unimplemented for now.
  130. internal func makeUnaryCall<Request: GRPCPayload, Response: GRPCPayload>(
  131. path: String,
  132. request: Request,
  133. callOptions: CallOptions
  134. ) -> UnaryCall<Request, Response> {
  135. fatalError("Not implemented")
  136. }
  137. internal func makeClientStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  138. path: String,
  139. callOptions: CallOptions
  140. ) -> ClientStreamingCall<Request, Response> {
  141. fatalError("Not implemented")
  142. }
  143. internal func makeServerStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  144. path: String,
  145. request: Request,
  146. callOptions: CallOptions,
  147. handler: @escaping (Response) -> Void
  148. ) -> ServerStreamingCall<Request, Response> {
  149. fatalError("Not implemented")
  150. }
  151. internal func makeBidirectionalStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  152. path: String,
  153. callOptions: CallOptions,
  154. handler: @escaping (Response) -> Void
  155. ) -> BidirectionalStreamingCall<Request, Response> {
  156. fatalError("Not implemented")
  157. }
  158. }