api: Add ClientStreamTracer.inboundHeaders(Metadata)

This will be used by the metadata exchange of CSM. When recording
per-attempt metrics, we really need per-attempt data and can't leverage
ClientInterceptors.

Author: ejona86

api/src/main/java/io/grpc/ClientStreamTracer.java

@@ -70,10 +70,23 @@ public void inboundHeaders() {
   }
 
   /**
-   * Trailing metadata has been received from the server.
+   * Headers has been received from the server. This method does not pass ownership to {@code
+   * headers}, so implementations must not access the metadata after returning. Modifications to the
+   * metadata within this method will be seen by interceptors and the application.
    *
-   * @param trailers the mutable trailing metadata.  Modifications to it will be seen by
-   *                 interceptors and the application.
+   * @param headers the received header metadata
+   */
+  public void inboundHeaders(Metadata headers) {
+    inboundHeaders();
+  }
+
+  /**
+   * Trailing metadata has been received from the server. This method does not pass ownership to
+   * {@code trailers}, so implementations must not access the metadata after returning.
+   * Modifications to the metadata within this method will be seen by interceptors and the
+   * application.
+   *
+   * @param trailers the received trailing metadata
    * @since 1.17.0
    */
   public void inboundTrailers(Metadata trailers) {

binder/src/main/java/io/grpc/binder/internal/Inbound.java

@@ -579,7 +579,7 @@ boolean countsForInUse() {
     @GuardedBy("this")
     protected void handlePrefix(int flags, Parcel parcel) throws StatusException {
       Metadata headers = MetadataHelper.readMetadata(parcel, attributes);
-      statsTraceContext.clientInboundHeaders();
+      statsTraceContext.clientInboundHeaders(headers);
       listener.headersRead(headers);
     }
 

core/src/main/java/io/grpc/internal/AbstractClientStream.java

@@ -304,7 +304,7 @@ protected final boolean isOutboundClosed() {
      */
     protected void inboundHeadersReceived(Metadata headers) {
       checkState(!statusReported, "Received headers on closed stream");
-      statsTraceCtx.clientInboundHeaders();
+      statsTraceCtx.clientInboundHeaders(headers);
 
       boolean compressedStream = false;
       String streamEncoding = headers.get(CONTENT_ENCODING_KEY);

core/src/main/java/io/grpc/internal/ForwardingClientStreamTracer.java

@@ -49,6 +49,11 @@ public void inboundHeaders() {
     delegate().inboundHeaders();
   }
 
+  @Override
+  public void inboundHeaders(Metadata headers) {
+    delegate().inboundHeaders(headers);
+  }
+
   @Override
   public void inboundTrailers(Metadata trailers) {
     delegate().inboundTrailers(trailers);

core/src/main/java/io/grpc/internal/StatsTraceContext.java

@@ -101,9 +101,9 @@ public void clientOutboundHeaders() {
    *
    * <p>Called from abstract stream implementations.
    */
-  public void clientInboundHeaders() {
+  public void clientInboundHeaders(Metadata headers) {
     for (StreamTracer tracer : tracers) {
-      ((ClientStreamTracer) tracer).inboundHeaders();
+      ((ClientStreamTracer) tracer).inboundHeaders(headers);
     }
   }
 

inprocess/src/main/java/io/grpc/inprocess/InProcessTransport.java

@@ -571,7 +571,7 @@ public void writeHeaders(Metadata headers, boolean flush) {
             return;
           }
 
-          clientStream.statsTraceCtx.clientInboundHeaders();
+          clientStream.statsTraceCtx.clientInboundHeaders(headers);
           syncContext.executeLater(() -> clientStreamListener.headersRead(headers));
         }
         syncContext.drain();

util/src/main/java/io/grpc/util/ForwardingClientStreamTracer.java

@@ -48,6 +48,11 @@ public void inboundHeaders() {
     delegate().inboundHeaders();
   }
 
+  @Override
+  public void inboundHeaders(Metadata headers) {
+    delegate().inboundHeaders(headers);
+  }
+
   @Override
   public void inboundTrailers(Metadata trailers) {
     delegate().inboundTrailers(trailers);