Add more properties to ClientContext and have the ClientTransport provide it

Sources/GRPCCodeGen/Internal/StructuredSwift+ServiceMetadata.swift

@@ -45,6 +45,7 @@ extension VariableDescription {
   /// static let descriptor = GRPCCore.MethodDescriptor(
   ///   service: GRPCCore.ServiceDescriptor(fullyQualifiedServiceName: "<literalFullyQualifiedService>"),
   ///   method: "<literalMethodName>"
+  /// )
   /// ```
   package static func methodDescriptor(
     accessModifier: AccessModifier? = nil,

Sources/GRPCCore/Call/Client/ClientContext.swift

@@ -19,8 +19,50 @@ public struct ClientContext: Sendable {
   /// A description of the method being called.
   public var descriptor: MethodDescriptor
 
+  /// A description of the remote peer.
+  ///
+  /// The format of the description should follow the pattern "<transport>:<address>" where
+  /// "<transport>" indicates the underlying network transport (such as "ipv4", "unix", or
+  /// "in-process"). This is a guideline for how descriptions should be formatted; different
+  /// implementations may not follow this format so you shouldn't make assumptions based on it.
+  ///
+  /// Some examples include:
+  /// - "ipv4:127.0.0.1:31415",
+  /// - "ipv6:[::1]:443",
+  /// - "in-process:27182".
+  public var remotePeer: String
+
+  /// The hostname of the RPC server.
+  public var serverHostname: String
+
+  /// A description of the local peer.
+  ///
+  /// The format of the description should follow the pattern "<transport>:<address>" where
+  /// "<transport>" indicates the underlying network transport (such as "ipv4", "unix", or
+  /// "in-process"). This is a guideline for how descriptions should be formatted; different
+  /// implementations may not follow this format so you shouldn't make assumptions based on it.
+  ///
+  /// Some examples include:
+  /// - "ipv4:127.0.0.1:31415",
+  /// - "ipv6:[::1]:443",
+  /// - "in-process:27182".
+  public var localPeer: String
+
+  /// The transport in use (e.g. "tcp", "udp").
+  public var networkTransportMethod: String
+
   /// Create a new client interceptor context.
-  public init(descriptor: MethodDescriptor) {
+  public init(
+    descriptor: MethodDescriptor,
+    remotePeer: String,
+    localPeer: String,
+    serverHostname: String,
+    networkTransportMethod: String
+  ) {
     self.descriptor = descriptor
+    self.remotePeer = remotePeer
+    self.localPeer = localPeer
+    self.serverHostname = serverHostname
+    self.networkTransportMethod = networkTransportMethod
   }
 }

Sources/GRPCCore/Call/Client/Internal/ClientRPCExecutor+HedgingExecutor.swift

@@ -322,7 +322,7 @@ extension ClientRPCExecutor.HedgingExecutor {
       return try await self.transport.withStream(
         descriptor: method,
         options: options
-      ) { stream -> _HedgingAttemptTaskResult<R, Output>.AttemptResult in
+      ) { stream, context -> _HedgingAttemptTaskResult<R, Output>.AttemptResult in
         return await withTaskGroup(of: _HedgingAttemptTaskResult<R, Output>.self) { group in
           group.addTask {
             do {
@@ -348,8 +348,8 @@ extension ClientRPCExecutor.HedgingExecutor {
 
               let response = await ClientRPCExecutor._execute(
                 in: &group,
+                context: context,
                 request: request,
-                method: method,
                 attempt: attempt,
                 serializer: self.serializer,
                 deserializer: self.deserializer,

Sources/GRPCCore/Call/Client/Internal/ClientRPCExecutor+OneShotExecutor.swift

@@ -98,11 +98,13 @@ extension ClientRPCExecutor.OneShotExecutor {
   ) async -> Result<R, any Error> {
     return await withTaskGroup(of: Void.self, returning: Result<R, any Error>.self) { group in
       do {
-        return try await self.transport.withStream(descriptor: method, options: options) { stream in
+        return try await self.transport.withStream(descriptor: method, options: options) {
+          stream,
+          context in
           let response = await ClientRPCExecutor._execute(
             in: &group,
+            context: context,
             request: request,
-            method: method,
             attempt: 1,
             serializer: self.serializer,
             deserializer: self.deserializer,

Sources/GRPCCore/Call/Client/Internal/ClientRPCExecutor+RetryExecutor.swift

@@ -118,7 +118,7 @@ extension ClientRPCExecutor.RetryExecutor {
           let attemptResult = try await self.transport.withStream(
             descriptor: method,
             options: options
-          ) { stream in
+          ) { stream, context in
             group.addTask {
               var metadata = request.metadata
               // Work out the timeout from the deadline.
@@ -127,6 +127,7 @@ extension ClientRPCExecutor.RetryExecutor {
               }
 
               return await self.executeAttempt(
+                context: context,
                 stream: stream,
                 metadata: metadata,
                 retryStream: retry.stream,
@@ -194,6 +195,7 @@ extension ClientRPCExecutor.RetryExecutor {
 
   @inlinable
   func executeAttempt<R: Sendable>(
+    context: ClientContext,
     stream: RPCStream<ClientTransport.Inbound, ClientTransport.Outbound>,
     metadata: Metadata,
     retryStream: BroadcastAsyncSequence<Input>,
@@ -211,8 +213,8 @@ extension ClientRPCExecutor.RetryExecutor {
 
       let response = await ClientRPCExecutor._execute(
         in: &group,
+        context: context,
         request: request,
-        method: method,
         attempt: attempt,
         serializer: self.serializer,
         deserializer: self.deserializer,

Sources/GRPCCore/Call/Client/Internal/ClientRPCExecutor.swift

@@ -104,25 +104,25 @@ extension ClientRPCExecutor {
   ///
   /// - Parameters:
   ///   - request: The request to execute.
-  ///   - method: A description of the method to execute the request against.
+  ///   - context: The ``ClientContext`` related to this request.
   ///   - attempt: The attempt number of the request.
   ///   - serializer: A serializer to convert input messages to bytes.
   ///   - deserializer: A deserializer to convert bytes to output messages.
   ///   - interceptors: An array of interceptors which the request and response pass through. The
   ///       interceptors will be called in the order of the array.
+  ///   - stream: The stream to excecute the RPC on.
   /// - Returns: The deserialized response.
   @inlinable  // would be private
   static func _execute<Input: Sendable, Output: Sendable>(
     in group: inout TaskGroup<Void>,
+    context: ClientContext,
     request: StreamingClientRequest<Input>,
-    method: MethodDescriptor,
     attempt: Int,
     serializer: some MessageSerializer<Input>,
     deserializer: some MessageDeserializer<Output>,
     interceptors: [any ClientInterceptor],
     stream: RPCStream<ClientTransport.Inbound, ClientTransport.Outbound>
   ) async -> StreamingClientResponse<Output> {
-    let context = ClientContext(descriptor: method)
 
     if interceptors.isEmpty {
       return await ClientStreamExecutor.execute(

Sources/GRPCCore/Transport/ClientTransport.swift

@@ -59,12 +59,12 @@ public protocol ClientTransport: Sendable {
   /// - Parameters:
   ///   - descriptor: A description of the method to open a stream for.
   ///   - options: Options specific to the stream.
-  ///   - closure: A closure that takes the opened stream as parameter.
+  ///   - closure: A closure that takes the opened stream and the client context as its parameters.
   /// - Returns: Whatever value was returned from `closure`.
   func withStream<T: Sendable>(
     descriptor: MethodDescriptor,
     options: CallOptions,
-    _ closure: (_ stream: RPCStream<Inbound, Outbound>) async throws -> T
+    _ closure: (_ stream: RPCStream<Inbound, Outbound>, _ context: ClientContext) async throws -> T
   ) async throws -> T
 
   /// Returns the configuration for a given method.

Sources/GRPCInProcessTransport/InProcessTransport+Client.swift

@@ -225,12 +225,12 @@ extension InProcessTransport {
     /// - Parameters:
     ///   - descriptor: A description of the method to open a stream for.
     ///   - options: Options specific to the stream.
-    ///   - closure: A closure that takes the opened stream as parameter.
+    ///   - closure: A closure that takes the opened stream and the client context as its parameters.
     /// - Returns: Whatever value was returned from `closure`.
     public func withStream<T>(
       descriptor: MethodDescriptor,
       options: CallOptions,
-      _ closure: (RPCStream<Inbound, Outbound>) async throws -> T
+      _ closure: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> T
     ) async throws -> T {
       let request = GRPCAsyncThrowingStream.makeStream(of: RPCRequestPart.self)
       let response = GRPCAsyncThrowingStream.makeStream(of: RPCResponsePart.self)
@@ -297,11 +297,19 @@ extension InProcessTransport {
         }
       }
 
+      let clientContext = ClientContext(
+        descriptor: descriptor,
+        remotePeer: "<in-process>",
+        localPeer: "<in-process>",
+        serverHostname: "<in-process>",
+        networkTransportMethod: "in-process"
+      )
+
       switch acceptStream {
       case .success(let streamID):
         let streamHandlingResult: Result<T, any Error>
         do {
-          let result = try await closure(clientStream)
+          let result = try await closure(clientStream, clientContext)
           streamHandlingResult = .success(result)
         } catch {
           streamHandlingResult = .failure(error)

Tests/GRPCCoreTests/GRPCServerTests.swift

@@ -48,7 +48,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.get,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message([3, 1, 4, 1, 5]))
         await stream.outbound.finish()
@@ -75,7 +75,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.collect,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message([3]))
         try await stream.outbound.write(.message([1]))
@@ -106,7 +106,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.expand,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message([3, 1, 4, 1, 5]))
         await stream.outbound.finish()
@@ -135,7 +135,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.update,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         for byte in [3, 1, 4, 1, 5] as [UInt8] {
           try await stream.outbound.write(.message([byte]))
@@ -166,7 +166,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: MethodDescriptor(fullyQualifiedService: "not", method: "implemented"),
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         await stream.outbound.finish()
 
@@ -187,7 +187,7 @@ final class GRPCServerTests: XCTestCase {
             try await client.withStream(
               descriptor: BinaryEcho.Methods.get,
               options: .defaults
-            ) { stream in
+            ) { stream, _ in
               try await stream.outbound.write(.metadata([:]))
               try await stream.outbound.write(.message([i]))
               await stream.outbound.finish()
@@ -225,7 +225,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.get,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         await stream.outbound.finish()
 
@@ -250,7 +250,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: MethodDescriptor(fullyQualifiedService: "not", method: "implemented"),
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         await stream.outbound.finish()
 
@@ -277,7 +277,7 @@ final class GRPCServerTests: XCTestCase {
         try await client.withStream(
           descriptor: BinaryEcho.Methods.get,
           options: .defaults
-        ) { stream in
+        ) { stream, _ in
           XCTFail("Stream shouldn't be opened")
         }
       } errorHandler: { error in
@@ -291,7 +291,7 @@ final class GRPCServerTests: XCTestCase {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.update,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         var iterator = stream.inbound.makeAsyncIterator()
         // Don't need to validate the response, just that the server is running.
@@ -364,7 +364,7 @@ final class GRPCServerTests: XCTestCase {
     try await transport.withStream(
       descriptor: BinaryEcho.Methods.get,
       options: .defaults
-    ) { stream in
+    ) { stream, _ in
       try await stream.outbound.write(.metadata([:]))
       try await stream.outbound.write(.message([0]))
       await stream.outbound.finish()
@@ -407,7 +407,7 @@ struct ServerTests {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.get,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message(Array("hello".utf8)))
         await stream.outbound.finish()
@@ -437,7 +437,7 @@ struct ServerTests {
       try await client.withStream(
         descriptor: HelloWorld.Methods.sayHello,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message(Array("Swift".utf8)))
         await stream.outbound.finish()
@@ -494,7 +494,7 @@ struct ServerTests {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.get,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message(Array("hello".utf8)))
         await stream.outbound.finish()
@@ -524,7 +524,7 @@ struct ServerTests {
       try await client.withStream(
         descriptor: BinaryEcho.Methods.collect,
         options: .defaults
-      ) { stream in
+      ) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
         try await stream.outbound.write(.message(Array("hello".utf8)))
         await stream.outbound.finish()

Tests/GRPCCoreTests/Test Utilities/Transport/AnyTransport.swift

@@ -24,7 +24,7 @@ struct AnyClientTransport: ClientTransport, Sendable {
     @Sendable (
       _ method: MethodDescriptor,
       _ options: CallOptions,
-      _ body: (RPCStream<Inbound, Outbound>) async throws -> (any Sendable)
+      _ body: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> (any Sendable)
     ) async throws -> Any
   private let _connect: @Sendable () async throws -> Void
   private let _close: @Sendable () -> Void
@@ -34,8 +34,8 @@ struct AnyClientTransport: ClientTransport, Sendable {
   where Transport.Inbound == Inbound, Transport.Outbound == Outbound {
     self._retryThrottle = { transport.retryThrottle }
     self._withStream = { descriptor, options, closure in
-      try await transport.withStream(descriptor: descriptor, options: options) { stream in
-        try await closure(stream) as (any Sendable)
+      try await transport.withStream(descriptor: descriptor, options: options) { stream, context in
+        try await closure(stream, context) as (any Sendable)
       }
     }
 
@@ -67,7 +67,7 @@ struct AnyClientTransport: ClientTransport, Sendable {
   func withStream<T>(
     descriptor: MethodDescriptor,
     options: CallOptions,
-    _ closure: (RPCStream<Inbound, Outbound>) async throws -> T
+    _ closure: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> T
   ) async throws -> T {
     let result = try await self._withStream(descriptor, options, closure)
     return result as! T

Tests/GRPCCoreTests/Test Utilities/Transport/StreamCountingTransport.swift

@@ -54,15 +54,15 @@ struct StreamCountingClientTransport: ClientTransport, Sendable {
   func withStream<T>(
     descriptor: MethodDescriptor,
     options: CallOptions,
-    _ closure: (RPCStream<Inbound, Outbound>) async throws -> T
+    _ closure: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> T
   ) async throws -> T {
     do {
       return try await self.transport.withStream(
         descriptor: descriptor,
         options: options
-      ) { stream in
+      ) { stream, context in
         self._streamsOpened.increment()
-        return try await closure(stream)
+        return try await closure(stream, context)
       }
     } catch {
       self._streamFailures.increment()