Stop sending requests in call init

Sources/GRPC/ClientCalls/BidirectionalStreamingCall.swift

@@ -134,40 +134,37 @@ public final class BidirectionalStreamingCall<
   }
 
   internal init(
-    path: String,
-    scheme: String,
-    authority: String,
-    callOptions: CallOptions,
-    eventLoop: EventLoop,
+    transport: ChannelTransport<RequestPayload, ResponsePayload>,
+    options: CallOptions
+  ) {
+    self.transport = transport
+    self.options = options
+  }
+
+  internal func sendHead(_ head: _GRPCRequestHead) {
+    self.transport.sendRequest(.head(head), promise: nil)
+  }
+}
+
+extension BidirectionalStreamingCall {
+  internal static func makeOnHTTP2Stream(
     multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>,
+    callOptions: CallOptions,
     errorDelegate: ClientErrorDelegate?,
     logger: Logger,
-    handler: @escaping (ResponsePayload) -> Void
-  ) {
-    let requestID = callOptions.requestIDProvider.requestID()
-    var logger = logger
-    logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
-    logger.debug("starting rpc", metadata: ["path": "\(path)"])
-
-    self.transport = ChannelTransport(
+    responseHandler: @escaping (ResponsePayload) -> Void
+  ) -> BidirectionalStreamingCall<RequestPayload, ResponsePayload> {
+    let eventLoop = multiplexer.eventLoop
+    let transport = ChannelTransport<RequestPayload, ResponsePayload>(
       multiplexer: multiplexer,
-      responseContainer: .init(eventLoop: eventLoop, streamingResponseHandler: handler),
+      responseContainer: .init(eventLoop: eventLoop, streamingResponseHandler: responseHandler),
       callType: .bidirectionalStreaming,
       timeLimit: callOptions.timeLimit,
       errorDelegate: errorDelegate,
       logger: logger
     )
 
-    self.options = callOptions
-
-    let requestHead = _GRPCRequestHead(
-      scheme: scheme,
-      path: path,
-      host: authority,
-      requestID: requestID,
-      options: callOptions
-    )
-
-    self.transport.sendRequest(.head(requestHead), promise: nil)
+    return BidirectionalStreamingCall(transport: transport, options: callOptions)
   }
 }
+

Sources/GRPC/ClientCalls/ClientStreamingCall.swift

@@ -137,40 +137,37 @@ public final class ClientStreamingCall<
   }
 
   internal init(
-    path: String,
-    scheme: String,
-    authority: String,
-    callOptions: CallOptions,
-    eventLoop: EventLoop,
+    response: EventLoopFuture<ResponsePayload>,
+    transport: ChannelTransport<RequestPayload, ResponsePayload>,
+    options: CallOptions
+  ) {
+    self.response = response
+    self.transport = transport
+    self.options = options
+  }
+
+  internal func sendHead(_ head: _GRPCRequestHead) {
+    self.transport.sendRequest(.head(head), promise: nil)
+  }
+}
+
+extension ClientStreamingCall {
+  internal static func makeOnHTTP2Stream(
     multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>,
+    callOptions: CallOptions,
     errorDelegate: ClientErrorDelegate?,
     logger: Logger
-  ) {
-    let requestID = callOptions.requestIDProvider.requestID()
-    var logger = logger
-    logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
-    logger.debug("starting rpc", metadata: ["path": "\(path)"])
-
+  ) -> ClientStreamingCall<RequestPayload, ResponsePayload> {
+    let eventLoop = multiplexer.eventLoop
     let responsePromise: EventLoopPromise<ResponsePayload> = eventLoop.makePromise()
-    self.transport = ChannelTransport(
+    let transport = ChannelTransport<RequestPayload, ResponsePayload>(
       multiplexer: multiplexer,
       responseContainer: .init(eventLoop: eventLoop, unaryResponsePromise: responsePromise),
       callType: .clientStreaming,
       timeLimit: callOptions.timeLimit,
       errorDelegate: errorDelegate,
       logger: logger
     )
-    self.options = callOptions
-    self.response = responsePromise.futureResult
-
-    let requestHead = _GRPCRequestHead(
-      scheme: scheme,
-      path: path,
-      host: authority,
-      requestID: requestID,
-      options: callOptions
-    )
-
-    self.transport.sendRequest(.head(requestHead), promise: nil)
+    return ClientStreamingCall(response: responsePromise.futureResult, transport: transport, options: callOptions)
   }
 }

Sources/GRPC/ClientCalls/ServerStreamingCall.swift

@@ -78,47 +78,38 @@ public final class ServerStreamingCall<
       }
     }
   }
-
-  init(
-    path: String,
-    scheme: String,
-    authority: String,
-    callOptions: CallOptions,
-    eventLoop: EventLoop,
+
+  internal init(
+    transport: ChannelTransport<RequestPayload, ResponsePayload>,
+    options: CallOptions
+  ) {
+    self.transport = transport
+    self.options = options
+  }
+
+  internal func send(_ head: _GRPCRequestHead, request: RequestPayload) {
+    self.transport.sendUnary(head, request: request, compressed: self.options.messageEncoding.enabledForRequests)
+  }
+}
+
+extension ServerStreamingCall {
+  internal static func makeOnHTTP2Stream(
     multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>,
+    callOptions: CallOptions,
     errorDelegate: ClientErrorDelegate?,
     logger: Logger,
-    request: RequestPayload,
-    handler: @escaping (ResponsePayload) -> Void
-  ) {
-    let requestID = callOptions.requestIDProvider.requestID()
-    var logger = logger
-    logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
-    logger[metadataKey: "path"] = "\(path)"
-
-    self.transport = ChannelTransport(
+    responseHandler: @escaping (ResponsePayload) -> Void
+  ) -> ServerStreamingCall<RequestPayload, ResponsePayload> {
+    let eventLoop = multiplexer.eventLoop
+    let transport = ChannelTransport<RequestPayload, ResponsePayload>(
       multiplexer: multiplexer,
-      responseContainer: .init(eventLoop: eventLoop, streamingResponseHandler: handler),
+      responseContainer: .init(eventLoop: eventLoop, streamingResponseHandler: responseHandler),
       callType: .serverStreaming,
       timeLimit: callOptions.timeLimit,
       errorDelegate: errorDelegate,
       logger: logger
     )
 
-    self.options = callOptions
-
-    let requestHead = _GRPCRequestHead(
-      scheme: scheme,
-      path: path,
-      host: authority,
-      requestID: requestID,
-      options: callOptions
-    )
-
-    self.transport.sendUnary(
-      requestHead,
-      request: request,
-      compressed: callOptions.messageEncoding.enabledForRequests
-    )
+    return ServerStreamingCall(transport: transport, options: callOptions)
   }
 }

Sources/GRPC/ClientCalls/UnaryCall.swift

@@ -85,46 +85,37 @@ public final class UnaryCall<
   }
 
   internal init(
-    path: String,
-    scheme: String,
-    authority: String,
-    callOptions: CallOptions,
-    eventLoop: EventLoop,
-    multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>,
-    errorDelegate: ClientErrorDelegate?,
-    logger: Logger,
-    request: RequestPayload
+    response: EventLoopFuture<ResponsePayload>,
+    transport: ChannelTransport<RequestPayload, ResponsePayload>,
+    options: CallOptions
   ) {
-    let requestID = callOptions.requestIDProvider.requestID()
-    var logger = logger
-    logger[metadataKey: MetadataKey.requestID] = "\(requestID)"
-    logger[metadataKey: "path"] = "\(path)"
+    self.response = response
+    self.transport = transport
+    self.options = options
+  }
 
+  internal func send(_ head: _GRPCRequestHead, request: RequestPayload) {
+    self.transport.sendUnary(head, request: request, compressed: self.options.messageEncoding.enabledForRequests)
+  }
+}
+
+extension UnaryCall {
+  internal static func makeOnHTTP2Stream(
+    multiplexer: EventLoopFuture<HTTP2StreamMultiplexer>,
+    callOptions: CallOptions,
+    errorDelegate: ClientErrorDelegate?,
+    logger: Logger
+  ) -> UnaryCall<RequestPayload, ResponsePayload> {
+    let eventLoop = multiplexer.eventLoop
     let responsePromise: EventLoopPromise<ResponsePayload> = eventLoop.makePromise()
-    self.transport = ChannelTransport(
+    let transport = ChannelTransport<RequestPayload, ResponsePayload>(
       multiplexer: multiplexer,
       responseContainer: .init(eventLoop: eventLoop, unaryResponsePromise: responsePromise),
       callType: .unary,
       timeLimit: callOptions.timeLimit,
       errorDelegate: errorDelegate,
       logger: logger
     )
-
-    self.options = callOptions
-    self.response = responsePromise.futureResult
-
-    let requestHead = _GRPCRequestHead(
-      scheme: scheme,
-      path: path,
-      host: authority,
-      requestID: requestID,
-      options: callOptions
-    )
-
-    self.transport.sendUnary(
-      requestHead,
-      request: request,
-      compressed: callOptions.messageEncoding.enabledForRequests
-    )
+    return UnaryCall(response: responsePromise.futureResult, transport: transport, options: callOptions)
   }
 }