Merge branch 'main' into v2/bag-o-bytes

Author: glbrntt

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,40 @@ 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
+
+  /// 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
+
   /// Create a new client interceptor context.
-  public init(descriptor: MethodDescriptor) {
+  public init(
+    descriptor: MethodDescriptor,
+    remotePeer: String,
+    localPeer: String
+  ) {
     self.descriptor = descriptor
+    self.remotePeer = remotePeer
+    self.localPeer = localPeer
   }
 }

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,14 @@ 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, Bytes: GRPCContiguousBytes>(
+    context: ClientContext,
     stream: RPCStream<
       RPCAsyncSequence<RPCResponsePart<Bytes>, any Error>,
       RPCWriter<RPCRequestPart<Bytes>>.Closable
@@ -214,8 +216,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,18 +104,19 @@ 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, Bytes: GRPCContiguousBytes>(
     in group: inout TaskGroup<Void>,
+    context: ClientContext,
     request: StreamingClientRequest<Input>,
-    method: MethodDescriptor,
     attempt: Int,
     serializer: some MessageSerializer<Input>,
     deserializer: some MessageDeserializer<Output>,
@@ -125,7 +126,6 @@ extension ClientRPCExecutor {
       RPCWriter<RPCRequestPart<Bytes>>.Closable
     >
   ) async -> StreamingClientResponse<Output> {
-    let context = ClientContext(descriptor: method)
 
     if interceptors.isEmpty {
       return await ClientStreamExecutor.execute(

Sources/GRPCCore/Transport/ClientTransport.swift

@@ -50,7 +50,7 @@ public protocol ClientTransport<Bytes>: Sendable {
   /// running ``connect()``.
   func beginGracefulShutdown()
 
-  /// 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 alongisde the given context.
   ///
   /// - Important: The opened stream is closed after the closure is finished.
   ///
@@ -62,12 +62,12 @@ public protocol ClientTransport<Bytes>: 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

@@ -104,19 +104,23 @@ extension InProcessTransport {
 
     private let methodConfig: MethodConfigs
     private let state: Mutex<State>
+    private let peer: String
 
     /// Creates a new in-process client transport.
     ///
     /// - Parameters:
     ///   - server: The in-process server transport to connect to.
     ///   - serviceConfig: Service configuration.
+    ///   - peer: The system's PID for the running client and server.
     package init(
       server: InProcessTransport.Server,
-      serviceConfig: ServiceConfig = ServiceConfig()
+      serviceConfig: ServiceConfig = ServiceConfig(),
+      peer: String
     ) {
       self.retryThrottle = serviceConfig.retryThrottling.map { RetryThrottle(policy: $0) }
       self.methodConfig = MethodConfigs(serviceConfig: serviceConfig)
       self.state = Mutex(.unconnected(.init(serverTransport: server)))
+      self.peer = peer
     }
 
     /// Establish and maintain a connection to the remote destination.
@@ -226,12 +230,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<Bytes>.self)
       let response = GRPCAsyncThrowingStream.makeStream(of: RPCResponsePart<Bytes>.self)
@@ -298,11 +302,17 @@ extension InProcessTransport {
         }
       }
 
+      let clientContext = ClientContext(
+        descriptor: descriptor,
+        remotePeer: self.peer,
+        localPeer: self.peer
+      )
+
       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)

Sources/GRPCInProcessTransport/InProcessTransport+Server.swift

@@ -36,7 +36,7 @@ extension InProcessTransport {
 
     private let newStreams: AsyncStream<RPCStream<Inbound, Outbound>>
     private let newStreamsContinuation: AsyncStream<RPCStream<Inbound, Outbound>>.Continuation
-    private let peer: String
+    package let peer: String
 
     private struct State: Sendable {
       private var _nextID: UInt64
@@ -76,6 +76,9 @@ extension InProcessTransport {
     private let handles: Mutex<State>
 
     /// Creates a new instance of ``Server``.
+    ///
+    /// - Parameters:
+    ///   - peer: The system's PID for the running client and server.
     package init(peer: String) {
       (self.newStreams, self.newStreamsContinuation) = AsyncStream.makeStream()
       self.handles = Mutex(State())

Sources/GRPCInProcessTransport/InProcessTransport.swift

@@ -27,6 +27,6 @@ public struct InProcessTransport: Sendable {
   public init(serviceConfig: ServiceConfig = ServiceConfig()) {
     let peer = "in-process:\(System.pid())"
     self.server = Self.Server(peer: peer)
-    self.client = Self.Client(server: self.server, serviceConfig: serviceConfig)
+    self.client = Self.Client(server: self.server, serviceConfig: serviceConfig, peer: peer)
   }
 }