EmbeddedGRPCChannel.swift 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133
  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. // This is currently intended for internal testing only.
  20. class EmbeddedGRPCChannel: GRPCChannel {
  21. let embeddedChannel: EmbeddedChannel
  22. let multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>
  23. let logger: Logger
  24. let scheme: String
  25. let authority: String
  26. let errorDelegate: ClientErrorDelegate?
  27. func close() -> EventLoopFuture<Void> {
  28. return embeddedChannel.close()
  29. }
  30. var eventLoop: EventLoop {
  31. return self.embeddedChannel.eventLoop
  32. }
  33. init(errorDelegate: ClientErrorDelegate? = nil) {
  34. let embeddedChannel = EmbeddedChannel()
  35. self.embeddedChannel = embeddedChannel
  36. let logger = Logger(subsystem: .clientChannel)
  37. self.logger = logger
  38. self.multiplexer = embeddedChannel.configureGRPCClient(
  39. errorDelegate: errorDelegate,
  40. logger: logger
  41. ).flatMap {
  42. embeddedChannel.pipeline.handler(type: HTTP2StreamMultiplexer.self)
  43. }
  44. self.scheme = "http"
  45. self.authority = "localhost"
  46. self.errorDelegate = errorDelegate
  47. }
  48. private func makeRequestHead(path: String, options: CallOptions) -> _GRPCRequestHead {
  49. return _GRPCRequestHead(
  50. scheme: self.scheme,
  51. path: path,
  52. host: self.authority,
  53. requestID: options.requestIDProvider.requestID(),
  54. options: options
  55. )
  56. }
  57. internal func makeUnaryCall<Request: GRPCPayload, Response: GRPCPayload>(
  58. path: String,
  59. request: Request,
  60. callOptions: CallOptions
  61. ) -> UnaryCall<Request, Response> {
  62. let call = UnaryCall<Request, Response>.makeOnHTTP2Stream(
  63. multiplexer: self.multiplexer,
  64. callOptions: callOptions,
  65. errorDelegate: self.errorDelegate,
  66. logger: self.logger
  67. )
  68. call.send(self.makeRequestHead(path: path, options: callOptions), request: request)
  69. return call
  70. }
  71. internal func makeClientStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  72. path: String,
  73. callOptions: CallOptions
  74. ) -> ClientStreamingCall<Request, Response> {
  75. let call = ClientStreamingCall<Request, Response>.makeOnHTTP2Stream(
  76. multiplexer: self.multiplexer,
  77. callOptions: callOptions,
  78. errorDelegate: self.errorDelegate,
  79. logger: self.logger
  80. )
  81. call.sendHead(self.makeRequestHead(path: path, options: callOptions))
  82. return call
  83. }
  84. internal func makeServerStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  85. path: String,
  86. request: Request,
  87. callOptions: CallOptions,
  88. handler: @escaping (Response) -> Void
  89. ) -> ServerStreamingCall<Request, Response> {
  90. let call = ServerStreamingCall<Request, Response>.makeOnHTTP2Stream(
  91. multiplexer: self.multiplexer,
  92. callOptions: callOptions,
  93. errorDelegate: self.errorDelegate,
  94. logger: self.logger,
  95. responseHandler: handler
  96. )
  97. call.send(self.makeRequestHead(path: path, options: callOptions), request: request)
  98. return call
  99. }
  100. internal func makeBidirectionalStreamingCall<Request: GRPCPayload, Response: GRPCPayload>(
  101. path: String,
  102. callOptions: CallOptions,
  103. handler: @escaping (Response) -> Void
  104. ) -> BidirectionalStreamingCall<Request, Response> {
  105. let call = BidirectionalStreamingCall<Request, Response>.makeOnHTTP2Stream(
  106. multiplexer: self.multiplexer,
  107. callOptions: callOptions,
  108. errorDelegate: self.errorDelegate,
  109. logger: self.logger,
  110. responseHandler: handler
  111. )
  112. call.sendHead(self.makeRequestHead(path: path, options: callOptions))
  113. return call
  114. }
  115. }