Adopt ClientContext changes

Sources/GRPCNIOTransportCore/Client/Connection/Connection.swift

@@ -421,6 +421,14 @@ extension Connection {
 
     private let http2Stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>
 
+    var peerInfo: String {
+      self.http2Stream.channel.getRemoteAddressInfo()
+    }
+
+    var localInfo: String {
+      self.http2Stream.channel.getLocalAddressInfo()
+    }
+
     init(
       wrapping stream: NIOAsyncChannel<RPCResponsePart, RPCRequestPart>,
       descriptor: MethodDescriptor

Sources/GRPCNIOTransportCore/Client/Connection/GRPCChannel.swift

@@ -198,11 +198,11 @@ package final class GRPCChannel: ClientTransport {
     self.input.continuation.yield(.close)
   }
 
-  /// Opens a stream using the transport, and uses it as input into a user-provided closure.
+  /// Opens a stream using the transport, and uses it as input into a user-provided closure, alongside the client's context.
   package 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 {
     // Merge options from the call with those from the service config.
     let methodConfig = self.config(forMethod: descriptor)
@@ -218,7 +218,14 @@ package final class GRPCChannel: ClientTransport {
             inbound: RPCAsyncSequence<RPCResponsePart, any Error>(wrapping: inbound),
             outbound: RPCWriter.Closable(wrapping: outbound)
           )
-          return try await closure(rpcStream)
+          let context = ClientContext(
+            descriptor: descriptor,
+            remotePeer: stream.peerInfo,
+            localPeer: stream.localInfo,
+            serverHostname: self.authority ?? "<unknown>",
+            networkTransportMethod: "tcp"
+          )
+          return try await closure(rpcStream, context)
         }
 
       case .tryAgain(let error):

Sources/GRPCNIOTransportCore/Internal/Channel+AddressInfo.swift

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2025, gRPC Authors All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import NIOCore
+
+extension Channel {
+  func getRemoteAddressInfo() -> String {
+    guard let remote = self.remoteAddress else {
+      return "<unknown>"
+    }
+
+    switch remote {
+    case .v4(let address):
+      // '!' is safe, v4 always has a port.
+      return "ipv4:\(address.host):\(remote.port!)"
+
+    case .v6(let address):
+      // '!' is safe, v6 always has a port.
+      return "ipv6:[\(address.host)]:\(remote.port!)"
+
+    case .unixDomainSocket:
+      // The pathname will be on the local address.
+      guard let local = self.localAddress else {
+        // UDS but no local address; this shouldn't ever happen but at least note the transport
+        // as being UDS.
+        return "unix:<unknown>"
+      }
+
+      switch local {
+      case .unixDomainSocket:
+        // '!' is safe, UDS always has a path.
+        return "unix:\(local.pathname!)"
+
+      case .v4, .v6:
+        // Remote address is UDS but local isn't. This shouldn't ever happen.
+        return "unix:<unknown>"
+      }
+    }
+  }
+
+  func getLocalAddressInfo() -> String {
+    guard let local = self.localAddress else {
+      return "<unknown>"
+    }
+
+    switch local {
+    case .v4(let address):
+      // '!' is safe, v4 always has a port.
+      return "ipv4:\(address.host):\(local.port!)"
+
+    case .v6(let address):
+      // '!' is safe, v6 always has a port.
+      return "ipv6:[\(address.host)]:\(local.port!)"
+
+    case .unixDomainSocket:
+      // '!' is safe, UDS always has a path.
+      return "unix:\(local.pathname!)"
+    }
+  }
+}

Sources/GRPCNIOTransportCore/Server/CommonHTTP2ServerTransport.swift

@@ -191,40 +191,6 @@ package final class CommonHTTP2ServerTransport<
     }
   }
 
-  private func peerInfo(channel: any Channel) -> String {
-    guard let remote = channel.remoteAddress else {
-      return "<unknown>"
-    }
-
-    switch remote {
-    case .v4(let address):
-      // '!' is safe, v4 always has a port.
-      return "ipv4:\(address.host):\(remote.port!)"
-
-    case .v6(let address):
-      // '!' is safe, v6 always has a port.
-      return "ipv6:[\(address.host)]:\(remote.port!)"
-
-    case .unixDomainSocket:
-      // The pathname will be on the local address.
-      guard let local = channel.localAddress else {
-        // UDS but no local address; this shouldn't ever happen but at least note the transport
-        // as being UDS.
-        return "unix:<unknown>"
-      }
-
-      switch local {
-      case .unixDomainSocket:
-        // '!' is safe, UDS always has a path.
-        return "unix:\(local.pathname!)"
-
-      case .v4, .v6:
-        // Remote address is UDS but local isn't. This shouldn't ever happen.
-        return "unix:<unknown>"
-      }
-    }
-  }
-
   private func handleConnection(
     _ connection: NIOAsyncChannel<HTTP2Frame, HTTP2Frame>,
     multiplexer: ChannelPipeline.SynchronousOperations.HTTP2StreamMultiplexer,
@@ -233,7 +199,7 @@ package final class CommonHTTP2ServerTransport<
       _ context: ServerContext
     ) async -> Void
   ) async throws {
-    let peer = self.peerInfo(channel: connection.channel)
+    let peer = connection.channel.getRemoteAddressInfo()
     try await connection.executeThenClose { inbound, _ in
       await withDiscardingTaskGroup { group in
         group.addTask {

Sources/GRPCNIOTransportHTTP2Posix/HTTP2ClientTransport+Posix.swift

@@ -120,7 +120,7 @@ extension HTTP2ClientTransport {
     public func withStream<T: Sendable>(
       descriptor: MethodDescriptor,
       options: CallOptions,
-      _ closure: (RPCStream<Inbound, Outbound>) async throws -> T
+      _ closure: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> T
     ) async throws -> T {
       try await self.channel.withStream(descriptor: descriptor, options: options, closure)
     }

Sources/GRPCNIOTransportHTTP2TransportServices/HTTP2ClientTransport+TransportServices.swift

@@ -118,7 +118,7 @@ extension HTTP2ClientTransport {
     public func withStream<T: Sendable>(
       descriptor: MethodDescriptor,
       options: CallOptions,
-      _ closure: (RPCStream<Inbound, Outbound>) async throws -> T
+      _ closure: (RPCStream<Inbound, Outbound>, ClientContext) async throws -> T
     ) async throws -> T {
       try await self.channel.withStream(descriptor: descriptor, options: options, closure)
     }

Tests/GRPCNIOTransportCoreTests/Client/Connection/GRPCChannelTests.swift

@@ -353,7 +353,7 @@ final class GRPCChannelTests: XCTestCase {
         await channel.connect()
       }
 
-      try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream in
+      try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream, _ in
         try await stream.outbound.write(.metadata([:]))
 
         var iterator = stream.inbound.makeAsyncIterator()
@@ -441,7 +441,7 @@ final class GRPCChannelTests: XCTestCase {
       // be queued though.
       for _ in 1 ... 100 {
         group.addTask {
-          try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream in
+          try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream, _ in
             try await stream.outbound.write(.metadata([:]))
             await stream.outbound.finish()
 
@@ -510,7 +510,7 @@ final class GRPCChannelTests: XCTestCase {
           options.waitForReady = false
 
           await XCTAssertThrowsErrorAsync(ofType: RPCError.self) {
-            try await channel.withStream(descriptor: .echoGet, options: options) { _ in
+            try await channel.withStream(descriptor: .echoGet, options: options) { _, _ in
               XCTFail("Unexpected stream")
             }
           } errorHandler: { error in
@@ -780,7 +780,7 @@ final class GRPCChannelTests: XCTestCase {
 
           // Try to open a new stream.
           await XCTAssertThrowsErrorAsync(ofType: RPCError.self) {
-            try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream in
+            try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream, _ in
               XCTFail("Unexpected new stream")
             }
           } errorHandler: { error in
@@ -823,7 +823,7 @@ final class GRPCChannelTests: XCTestCase {
       }
 
       func doAnRPC() async throws {
-        try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream in
+        try await channel.withStream(descriptor: .echoGet, options: .defaults) { stream, _ in
           try await stream.outbound.write(.metadata([:]))
           await stream.outbound.finish()
 
@@ -873,7 +873,7 @@ extension GRPCChannel {
     let values: Metadata.StringValues? = try await self.withStream(
       descriptor: .echoGet,
       options: .defaults
-    ) { stream in
+    ) { stream, _ in
       try await stream.outbound.write(.metadata([:]))
       await stream.outbound.finish()