feat: define default request headers for a client (#2018)

johanandren · web-flow · commit cd3392ef8a13 · 2025-01-30T09:51:02.000+01:00
diff --git a/codegen/src/main/twirl/templates/JavaClient/Client.scala.txt b/codegen/src/main/twirl/templates/JavaClient/Client.scala.txt
@@ -46,6 +46,7 @@ public abstract class @{service.name}Client extends @{service.name}ClientPowerAp
       private final io.grpc.CallOptions options;
       private final Materializer mat;
       private final ExecutionContext ec;
+      private final MetadataImpl defaultMetadata;
 
       private Default@{service.name}Client(GrpcChannel channel, boolean isChannelOwned, ClassicActorSystemProvider sys) {
         this.channel = channel;
@@ -54,30 +55,41 @@ public abstract class @{service.name}Client extends @{service.name}ClientPowerAp
         this.mat = SystemMaterializer.get(sys).materializer();
         this.ec = sys.classicSystem().dispatcher();
         this.options = NettyClientUtils.callOptions(settings);
+        this.defaultMetadata = MetadataImpl.empty();
 
         sys.classicSystem().getWhenTerminated().whenComplete((v, e) -> close());
       }
 
+      private Default@{service.name}Client(GrpcChannel channel, boolean isChannelOwned, GrpcClientSettings settings, io.grpc.CallOptions options, Materializer mat, ExecutionContext ec, MetadataImpl defaultMetadata) {
+        this.channel = channel;
+        this.isChannelOwned = isChannelOwned;
+        this.settings = settings;
+        this.options = options;
+        this.mat = mat;
+        this.ec = ec;
+        this.defaultMetadata = defaultMetadata;
+      }
+
   @for(method <- service.methods) {
     @if(method.methodType == akka.grpc.gen.Unary) {
       private final SingleResponseRequestBuilder<@method.inputTypeUnboxed, @method.outputTypeUnboxed> @{method.name}RequestBuilder(akka.grpc.internal.InternalChannel channel){
-        return new JavaUnaryRequestBuilder<>(@{method.name}Descriptor, channel, options, settings, ec);
+        return new JavaUnaryRequestBuilder<>(@{method.name}Descriptor, channel, options, settings, defaultMetadata, ec);
       }
     } else {
       @if(method.methodType == akka.grpc.gen.ClientStreaming){
         private final SingleResponseRequestBuilder<akka.stream.javadsl.Source<@method.inputTypeUnboxed, akka.NotUsed>, @method.outputTypeUnboxed> @{method.name}RequestBuilder(akka.grpc.internal.InternalChannel channel){
           return new JavaClientStreamingRequestBuilder<>(
-                               @{method.name}Descriptor, channel, options, settings, mat, ec);
+                               @{method.name}Descriptor, channel, options, settings, defaultMetadata, mat, ec);
         }
       } else if(method.methodType == akka.grpc.gen.ServerStreaming){
         private final StreamResponseRequestBuilder<@method.inputTypeUnboxed, @method.outputTypeUnboxed> @{method.name}RequestBuilder(akka.grpc.internal.InternalChannel channel){
           return new JavaServerStreamingRequestBuilder<>(
-                               @{method.name}Descriptor, channel, options, settings, ec);
+                               @{method.name}Descriptor, channel, options, settings, defaultMetadata, ec);
         }
       } else if(method.methodType == akka.grpc.gen.BidiStreaming){
         private final StreamResponseRequestBuilder<akka.stream.javadsl.Source<@method.inputTypeUnboxed, akka.NotUsed>, @method.outputTypeUnboxed> @{method.name}RequestBuilder(akka.grpc.internal.InternalChannel channel){
           return new JavaBidirectionalStreamingRequestBuilder<>(
-                               @{method.name}Descriptor, channel, options, settings, ec);
+                               @{method.name}Descriptor, channel, options, settings, defaultMetadata, ec);
         }
       }
     }
@@ -139,8 +151,30 @@ public abstract class @{service.name}Client extends @{service.name}ClientPowerAp
       public java.util.concurrent.CompletionStage<akka.Done> closed() {
         return channel.closedCS();
       }
+
+      /**
+       * The same client instance decorated to add the given key and value to the metadata of any request issued.
+       */
+      public @{service.name}Client addRequestHeader(String key, String value) {
+        return new Default@{service.name}Client(
+          channel,
+          isChannelOwned,
+          settings,
+          options,
+          mat,
+          ec,
+          defaultMetadata.addEntry(key, value));
+      }
   }
 
+   /**
+    * The same client instance decorated to add the given key and value to the metadata of any request issued.
+    */
+   public @{service.name}Client addRequestHeader(String key, String value) {
+     // default implementation ignoring header for source compatibility
+     return this;
+   }
+
 }
 
 
diff --git a/codegen/src/main/twirl/templates/ScalaClient/Client.scala.txt b/codegen/src/main/twirl/templates/ScalaClient/Client.scala.txt
@@ -18,6 +18,7 @@ import akka.grpc.GrpcClientSettings
 import akka.grpc.scaladsl.AkkaGrpcClient
 
 import akka.grpc.internal.NettyClientUtils
+import akka.grpc.internal.MetadataImpl
 
 import akka.grpc.AkkaGrpcGenerated
 
@@ -39,16 +40,23 @@ import akka.grpc.AkkaGrpcGenerated
 
 // Not sealed so users can extend to write their stubs
 @@AkkaGrpcGenerated
-trait @{service.name}Client extends @{service.name} with @{service.name}ClientPowerApi with AkkaGrpcClient
+trait @{service.name}Client extends @{service.name} with @{service.name}ClientPowerApi with AkkaGrpcClient {
+  /**
+   * The same client instance decorated to add the given key and value to the metadata of any request issued.
+   */
+  override def addRequestHeader(key: String, value: String): @{service.name}Client =
+    // default implementation ignoring header for source compatibility
+    this
+}
 
 @@AkkaGrpcGenerated
 object @{service.name}Client {
   def apply(settings: GrpcClientSettings)(implicit sys: ClassicActorSystemProvider): @{service.name}Client =
-    new Default@{service.name}Client(GrpcChannel(settings), isChannelOwned = true)
+    new Default@{service.name}Client(GrpcChannel(settings), isChannelOwned = true, defaultMetadata = MetadataImpl.empty)
   def apply(channel: GrpcChannel)(implicit sys: ClassicActorSystemProvider): @{service.name}Client =
-    new Default@{service.name}Client(channel, isChannelOwned = false)
+    new Default@{service.name}Client(channel, isChannelOwned = false, defaultMetadata = MetadataImpl.empty)
 
-  private class Default@{service.name}Client(channel: GrpcChannel, isChannelOwned: Boolean)(implicit sys: ClassicActorSystemProvider) extends @{service.name}Client {
+  private class Default@{service.name}Client(channel: GrpcChannel, isChannelOwned: Boolean, defaultMetadata: MetadataImpl)(implicit sys: ClassicActorSystemProvider) extends @{service.name}Client {
     import @{service.name}.MethodDescriptors._
 
     private implicit val ex: ExecutionContext = sys.classicSystem.dispatcher
@@ -58,14 +66,14 @@ object @{service.name}Client {
     @for(method <- service.methods) {
     private def @{method.name}RequestBuilder(channel: akka.grpc.internal.InternalChannel) =
     @if(method.methodType == akka.grpc.gen.Unary) {
-      new ScalaUnaryRequestBuilder(@{method.name}Descriptor, channel, options, settings)
+      new ScalaUnaryRequestBuilder(@{method.name}Descriptor, channel, options, settings, defaultMetadata)
     } else {
       @if(method.methodType == akka.grpc.gen.ServerStreaming) {
-      new ScalaServerStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings)
+      new ScalaServerStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings, defaultMetadata)
       } else if(method.methodType == akka.grpc.gen.ClientStreaming) {
-      new ScalaClientStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings)
+      new ScalaClientStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings, defaultMetadata)
       } else if (method.methodType == akka.grpc.gen.BidiStreaming) {
-      new ScalaBidirectionalStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings)
+      new ScalaBidirectionalStreamingRequestBuilder(@{method.name}Descriptor, channel, options, settings, defaultMetadata)
       }
     }
     }
@@ -90,6 +98,13 @@ object @{service.name}Client {
       @{method.nameSafe}().invoke(in)
     }
 
+    /**
+     * The same client instance decorated to add the given key and value to the metadata of any request issued.
+     */
+    override def addRequestHeader(key: String, value: String): @{service.name}Client =
+      new Default@{service.name}Client(channel, isChannelOwned, defaultMetadata.addEntry(key, value))
+
+
     override def close(): scala.concurrent.Future[akka.Done] =
       if (isChannelOwned) channel.close()
       else throw new GrpcClientCloseException()
diff --git a/plugin-tester-java/src/main/java/example/myapp/helloworld/AuthenticatedGreeterClient.java b/plugin-tester-java/src/main/java/example/myapp/helloworld/AuthenticatedGreeterClient.java
@@ -0,0 +1,61 @@
+/*
+ * Copyright (C) 2024 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package example.myapp.helloworld;
+
+import akka.actor.ActorSystem;
+import akka.grpc.GrpcClientSettings;
+import example.myapp.helloworld.grpc.GreeterServiceClient;
+import example.myapp.helloworld.grpc.HelloReply;
+import example.myapp.helloworld.grpc.HelloRequest;
+import io.grpc.StatusRuntimeException;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public class AuthenticatedGreeterClient {
+
+  public static void main(String[] args) throws Exception {
+
+    String serverHost = "127.0.0.1";
+    int serverPort = 8082;
+
+    ActorSystem system = ActorSystem.create("HelloWorldClient");
+
+    // Configure the client by code:
+    GrpcClientSettings settings = GrpcClientSettings.connectToServiceAt(serverHost, serverPort, system).withTls(false);
+
+    GreeterServiceClient client = null;
+    try {
+      client = GreeterServiceClient.create(settings, system);
+
+      try {
+        client.sayHello(HelloRequest.newBuilder().setName("Alice").build()).toCompletableFuture().get(5, TimeUnit.SECONDS);
+        throw new RuntimeException("Call should have failed");
+      } catch (ExecutionException ex) {
+        system.log().warning(ex, "Call without authentication fails as expected");
+      }
+
+      HelloReply replyWhenAuthenticated = client.sayHello()
+        .addHeader("Token", "XYZ")
+        .invoke(HelloRequest.newBuilder().setName("Alice").build()).toCompletableFuture().get(5, TimeUnit.SECONDS);
+      system.log().info("Call with authentication succeeds: {}", replyWhenAuthenticated);
+
+      GreeterServiceClient clientWithMeta = client.addRequestHeader("Token", "XYZ");
+
+      HelloReply replyWhenInterceptAuthenticated = clientWithMeta.sayHello()
+        .addHeader("Token", "XYZ")
+        .invoke(HelloRequest.newBuilder().setName("Alice").build()).toCompletableFuture().get(5, TimeUnit.SECONDS);
+      system.log().info("Call with authentication succeeds: {}", replyWhenInterceptAuthenticated);
+
+    } catch (StatusRuntimeException e) {
+      System.out.println("Status: " + e.getStatus());
+    } catch (Exception e)  {
+      System.out.println(e);
+    } finally {
+      if (client != null) client.close();
+      system.terminate();
+    }
+  }
+}
diff --git a/plugin-tester-java/src/test/scala/example/myapp/helloworld/JGreeterServiceSpec.scala b/plugin-tester-java/src/test/scala/example/myapp/helloworld/JGreeterServiceSpec.scala
@@ -56,6 +56,12 @@ class JGreeterServiceSpec extends Matchers with AnyWordSpecLike with BeforeAndAf
         HelloReply.newBuilder.setMessage("Hello, Alice").setTimestamp(timestamp).build()
       reply.toCompletableFuture.get should ===(expectedResponse)
     }
+
+    "use default metadata" in {
+      val clientWithHeader = clients.last.addRequestHeader("Authorization", "Bearer test")
+      val reply = clientWithHeader.sayHello(HelloRequest.newBuilder().setName("Alice").build())
+      reply.toCompletableFuture.get should ===(HelloReply.newBuilder.setMessage("Hello, Alice (authenticated)").build())
+    }
   }
 
   "GreeterServicePowerApi" should {
diff --git a/plugin-tester-scala/src/main/scala/example/myapp/helloworld/AuthenticatedGreeterClient.scala b/plugin-tester-scala/src/main/scala/example/myapp/helloworld/AuthenticatedGreeterClient.scala
@@ -27,5 +27,13 @@ object AuthenticatedGreeterClient {
     val replyWhenAuthenticated =
       Await.result(client.sayHello().addHeader("Token", "XYZ").invoke(HelloRequest("Alice")), 10.seconds)
     sys.log.warning(s"Call with authentication succeeds: $replyWhenAuthenticated")
+
+    val interceptedClient: GreeterServiceClient =
+      client.addRequestHeader("Token", "XYZ")
+
+    val replyWhenInterceptAuthenticated =
+      Await.result(interceptedClient.sayHello(HelloRequest("Alice")), 10.seconds)
+    sys.log.warning(s"Call with intercepted authentication succeeds: $replyWhenInterceptAuthenticated")
+
   }
 }
diff --git a/plugin-tester-scala/src/test/scala/example/myapp/helloworld/GreeterServiceSpec.scala b/plugin-tester-scala/src/test/scala/example/myapp/helloworld/GreeterServiceSpec.scala
@@ -69,6 +69,12 @@ class GreeterSpec extends Matchers with AnyWordSpecLike with BeforeAndAfterAll w
       val reply = eagerClient.sayHello(HelloRequest("Alice"))
       reply.futureValue should ===(HelloReply("Hello, Alice", Some(Timestamp.apply(123456, 123))))
     }
+
+    "use default metadata" in {
+      val clientWithHeader = clients.last.addRequestHeader("Authorization", "Bearer test")
+      val reply = clientWithHeader.sayHello(HelloRequest("Alice"))
+      reply.futureValue should ===(HelloReply("Hello, Alice (authenticated)"))
+    }
   }
 
   "GreeterServicePowerApi" should {
diff --git a/runtime/src/main/mima-filters/2.5.1.backwards.excludes/client-request-headers.excludes b/runtime/src/main/mima-filters/2.5.1.backwards.excludes/client-request-headers.excludes
@@ -0,0 +1,5 @@
+# ok because has default impl in generated user impls
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.javadsl.AkkaGrpcClient.addRequestHeader")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.grpc.scaladsl.AkkaGrpcClient.addRequestHeader")
+# new internal constructor in generated classes (and we generate reflection client)
+ProblemFilters.exclude[DirectMissingMethodProblem]("grpc.reflection.v1alpha.reflection.ServerReflectionClient#DefaultServerReflectionClient.this")
diff --git a/runtime/src/main/scala/akka/grpc/internal/MetadataImpl.scala b/runtime/src/main/scala/akka/grpc/internal/MetadataImpl.scala
@@ -19,7 +19,11 @@ import com.google.protobuf.any
 import com.google.rpc.Status
 import scalapb.{ GeneratedMessage, GeneratedMessageCompanion }
 
-@InternalApi private[akka] object MetadataImpl {
+/**
+ * INTERNAL API
+ */
+// Note: empty value used by generated code, cannot be private
+@InternalApi object MetadataImpl {
   val BINARY_SUFFIX: String = io.grpc.Metadata.BINARY_HEADER_SUFFIX
 
   val empty = new MetadataImpl(List.empty)
@@ -56,7 +60,11 @@ import scalapb.{ GeneratedMessage, GeneratedMessageCompanion }
   }
 }
 
-@InternalApi private[akka] final class MetadataImpl(val entries: List[(String, MetadataEntry)]) {
+/**
+ * INTERNAL API
+ */
+// Note: type used by generated code, cannot be private
+@InternalApi final class MetadataImpl(val entries: List[(String, MetadataEntry)]) {
   def addEntry(key: String, value: String): MetadataImpl = {
     if (key.endsWith(MetadataImpl.BINARY_SUFFIX))
       throw new IllegalArgumentException(s"String header names must not end with '${MetadataImpl.BINARY_SUFFIX}'")
diff --git a/runtime/src/main/scala/akka/grpc/javadsl/AkkaGrpcClient.scala b/runtime/src/main/scala/akka/grpc/javadsl/AkkaGrpcClient.scala
@@ -13,6 +13,11 @@ import akka.annotation.DoNotInherit
 @DoNotInherit
 trait AkkaGrpcClient {
 
+  /**
+   * @return The same client decorated to add the given key and value to the metadata of any request issued.
+   */
+  def addRequestHeader(key: String, value: String): AkkaGrpcClient
+
   /**
    * Initiates a shutdown in which preexisting and new calls are cancelled.
    *
diff --git a/runtime/src/main/scala/akka/grpc/scaladsl/AkkaGrpcClient.scala b/runtime/src/main/scala/akka/grpc/scaladsl/AkkaGrpcClient.scala
@@ -29,4 +29,9 @@ trait AkkaGrpcClient {
    * after maxConnectionAttempts.
    */
   def closed: Future[Done]
+
+  /**
+   * The same client instance decorated to add the given key and value to the metadata of any request issued.
+   */
+  def addRequestHeader(key: String, value: String): AkkaGrpcClient
 }

Original file line number	Diff line number	Diff line change
`@@ -56,6 +56,12 @@ class JGreeterServiceSpec extends Matchers with AnyWordSpecLike with BeforeAndAf`
`56`	`56`	`HelloReply.newBuilder.setMessage("Hello, Alice").setTimestamp(timestamp).build()`
`57`	`57`	`reply.toCompletableFuture.get should ===(expectedResponse)`
`58`	`58`	`}`
	`59`	`+`
	`60`	`+ "use default metadata" in {`
	`61`	`+ val clientWithHeader = clients.last.addRequestHeader("Authorization", "Bearer test")`
	`62`	`+ val reply = clientWithHeader.sayHello(HelloRequest.newBuilder().setName("Alice").build())`
	`63`	`+ reply.toCompletableFuture.get should ===(HelloReply.newBuilder.setMessage("Hello, Alice (authenticated)").build())`
	`64`	`+ }`
`59`	`65`	`}`
`60`	`66`
`61`	`67`	`"GreeterServicePowerApi" should {`
Original file line number	Diff line number	Diff line change
`@@ -27,5 +27,13 @@ object AuthenticatedGreeterClient {`
`27`	`27`	`val replyWhenAuthenticated =`
`28`	`28`	`Await.result(client.sayHello().addHeader("Token", "XYZ").invoke(HelloRequest("Alice")), 10.seconds)`
`29`	`29`	`sys.log.warning(s"Call with authentication succeeds: $replyWhenAuthenticated")`
	`30`	`+`
	`31`	`+ val interceptedClient: GreeterServiceClient =`
	`32`	`+ client.addRequestHeader("Token", "XYZ")`
	`33`	`+`
	`34`	`+ val replyWhenInterceptAuthenticated =`
	`35`	`+ Await.result(interceptedClient.sayHello(HelloRequest("Alice")), 10.seconds)`
	`36`	`+ sys.log.warning(s"Call with intercepted authentication succeeds: $replyWhenInterceptAuthenticated")`
	`37`	`+`
`30`	`38`	`}`
`31`	`39`	`}`
Original file line number	Diff line number	Diff line change
`@@ -69,6 +69,12 @@ class GreeterSpec extends Matchers with AnyWordSpecLike with BeforeAndAfterAll w`
`69`	`69`	`val reply = eagerClient.sayHello(HelloRequest("Alice"))`
`70`	`70`	`reply.futureValue should ===(HelloReply("Hello, Alice", Some(Timestamp.apply(123456, 123))))`
`71`	`71`	`}`
	`72`	`+`
	`73`	`+ "use default metadata" in {`
	`74`	`+ val clientWithHeader = clients.last.addRequestHeader("Authorization", "Bearer test")`
	`75`	`+ val reply = clientWithHeader.sayHello(HelloRequest("Alice"))`
	`76`	`+ reply.futureValue should ===(HelloReply("Hello, Alice (authenticated)"))`
	`77`	`+ }`
`72`	`78`	`}`
`73`	`79`
`74`	`80`	`"GreeterServicePowerApi" should {`