Make ReactiveClientHeadersFactory replace headers instead of adding - backport to 2.7

docs/src/main/asciidoc/rest-client-reactive.adoc

@@ -611,7 +611,9 @@ public class GetTokenReactiveClientHeadersFactory extends ReactiveClientHeadersF
     Service service;
 
     @Override
-    public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
+    public Uni<MultivaluedMap<String, String>> getHeaders(
+            MultivaluedMap<String, String> incomingHeaders,
+            MultivaluedMap<String, String> clientOutgoingHeaders);
         return Uni.createFrom().item(() -> {
             MultivaluedHashMap<String, String> newHeaders = new MultivaluedHashMap<>();
             // perform blocking call

extensions/resteasy-reactive/rest-client-reactive/deployment/src/test/java/io/quarkus/rest/client/reactive/headers/ReactiveClientHeadersFromProviderTest.java

@@ -1,14 +1,22 @@
 package io.quarkus.rest.client.reactive.headers;
 
+import static io.restassured.RestAssured.given;
+import static java.lang.String.format;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.net.URI;
+import java.util.List;
+import java.util.Map;
 
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 import javax.ws.rs.GET;
 import javax.ws.rs.HeaderParam;
+import javax.ws.rs.POST;
 import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.HttpHeaders;
 import javax.ws.rs.core.MultivaluedHashMap;
 import javax.ws.rs.core.MultivaluedMap;
 
@@ -23,10 +31,15 @@
 import io.quarkus.test.common.http.TestHTTPResource;
 import io.smallrye.common.annotation.Blocking;
 import io.smallrye.mutiny.Uni;
+import io.vertx.core.Vertx;
 
 public class ReactiveClientHeadersFromProviderTest {
     private static final String HEADER_NAME = "my-header";
     private static final String HEADER_VALUE = "oifajrofijaeoir5gjaoasfaxcvcz";
+    public static final String COPIED_INCOMING_HEADER = "copied-incoming-header";
+    public static final String INCOMING_HEADER = "incoming-header";
+    public static final String DIRECT_HEADER_PARAM = "direct-header-param";
+    public static final String DIRECT_HEADER_PARAM_VAL = "direct-header-param-val";
 
     @TestHTTPResource
     URI baseUri;
@@ -39,19 +52,42 @@ public class ReactiveClientHeadersFromProviderTest {
                             "application.properties"));
 
     @Test
-    void shouldSetHeaderFromProperties() {
-        ReactiveClientHeadersFromProviderTest.Client client = RestClientBuilder.newBuilder().baseUri(baseUri)
-                .build(ReactiveClientHeadersFromProviderTest.Client.class);
-
-        assertThat(client.getWithHeader()).isEqualTo(HEADER_VALUE);
+    void shouldPropagateHeaders() {
+        // we're calling a resource that sets "incoming-header" header
+        // this header should be dropped by the client and its value should be put into copied-incoming-header
+        String propagatedHeaderValue = "propag8ed header";
+        // @formatter:off
+        var response =
+                given()
+                   .header(INCOMING_HEADER, propagatedHeaderValue)
+                   .body(baseUri.toString())
+                .when()
+                   .post("/call-client")
+                .thenReturn();
+        // @formatter:on
+        assertThat(response.statusCode()).isEqualTo(200);
+        assertThat(response.jsonPath().getString(INCOMING_HEADER)).isNull();
+        assertThat(response.jsonPath().getString(COPIED_INCOMING_HEADER)).isEqualTo(format("[%s]", propagatedHeaderValue));
+        assertThat(response.jsonPath().getString(HEADER_NAME)).isEqualTo(format("[%s]", HEADER_VALUE));
+        assertThat(response.jsonPath().getString(DIRECT_HEADER_PARAM)).isEqualTo(format("[%s]", DIRECT_HEADER_PARAM_VAL));
     }
 
     @Path("/")
     @ApplicationScoped
     public static class Resource {
+
         @GET
-        public String returnHeaderValue(@HeaderParam(HEADER_NAME) String header) {
-            return header;
+        @Produces("application/json")
+        public Map<String, List<String>> returnHeaderValues(@Context HttpHeaders headers) {
+            return headers.getRequestHeaders();
+        }
+
+        @Path("/call-client")
+        @POST
+        public Map<String, List<String>> callClient(String uri) {
+            ReactiveClientHeadersFromProviderTest.Client client = RestClientBuilder.newBuilder().baseUri(URI.create(uri))
+                    .build(ReactiveClientHeadersFromProviderTest.Client.class);
+            return client.getWithHeader(DIRECT_HEADER_PARAM_VAL);
         }
     }
 
@@ -66,20 +102,25 @@ public String getValue() {
     @RegisterClientHeaders(CustomReactiveClientHeadersFactory.class)
     public interface Client {
         @GET
-        String getWithHeader();
+        Map<String, List<String>> getWithHeader(@HeaderParam(DIRECT_HEADER_PARAM) String directHeaderParam);
     }
 
     public static class CustomReactiveClientHeadersFactory extends ReactiveClientHeadersFactory {
 
         @Inject
         Service service;
 
+        @Inject
+        Vertx vertx;
+
         @Override
-        public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
-            return Uni.createFrom().item(() -> {
+        public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders,
+                MultivaluedMap<String, String> clientOutgoingHeaders) {
+            return Uni.createFrom().emitter(e -> {
                 MultivaluedHashMap<String, String> newHeaders = new MultivaluedHashMap<>();
                 newHeaders.add(HEADER_NAME, service.getValue());
-                return newHeaders;
+                newHeaders.add(COPIED_INCOMING_HEADER, incomingHeaders.getFirst(INCOMING_HEADER));
+                vertx.setTimer(100L, id -> e.complete(newHeaders));
             });
         }
     }

extensions/resteasy-reactive/rest-client-reactive/runtime/src/main/java/io/quarkus/rest/client/reactive/ReactiveClientHeadersFactory.java

@@ -10,7 +10,35 @@
  * Reactive ClientHeadersFactory flavor for Quarkus rest-client reactive extension.
  */
 public abstract class ReactiveClientHeadersFactory implements ClientHeadersFactory {
-    public abstract Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders);
+
+    /**
+     * Updates the HTTP headers to send to the remote service. Note that providers
+     * on the outbound processing chain could further update the headers.
+     *
+     * @param incomingHeaders the map of headers from the inbound JAX-RS request. This will be an empty map if the
+     *        associated client interface is not part of a JAX-RS request.
+     * @param clientOutgoingHeaders the read-only map of header parameters specified on the client interface.
+     * @return a Uni with a map of HTTP headers to merge with the clientOutgoingHeaders to be sent to the remote service.
+     *
+     * @see ClientHeadersFactory#update(MultivaluedMap, MultivaluedMap)
+     */
+    public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders,
+            MultivaluedMap<String, String> clientOutgoingHeaders) {
+        return getHeaders(incomingHeaders);
+    }
+
+    /**
+     * @deprecated Will be removed in Quarkus 2.8. Implement and use
+     *             {@link ReactiveClientHeadersFactory#getHeaders(MultivaluedMap, MultivaluedMap)} instead
+     *
+     * @param incomingHeaders the map of headers from the inbound JAX-RS request. This will be an empty map if the
+     *        associated client interface is not part of a JAX-RS request.
+     * @return a Uni with a map of HTTP headers to merge with the clientOutgoingHeaders to be sent to the remote service.
+     */
+    @Deprecated
+    public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
+        throw new IllegalStateException(getClass() + " does not implement either of the getHeaders() methods");
+    }
 
     @Override
     public final MultivaluedMap<String, String> update(MultivaluedMap<String, String> incomingHeaders,

extensions/resteasy-reactive/rest-client-reactive/runtime/src/main/java/io/quarkus/rest/client/reactive/runtime/MicroProfileRestClientRequestFilter.java

@@ -6,21 +6,20 @@
 
 import javax.annotation.Priority;
 import javax.enterprise.context.RequestScoped;
-import javax.ws.rs.client.ClientRequestContext;
-import javax.ws.rs.client.ClientRequestFilter;
 import javax.ws.rs.core.MultivaluedHashMap;
 import javax.ws.rs.core.MultivaluedMap;
 
 import org.eclipse.microprofile.rest.client.ext.ClientHeadersFactory;
 import org.eclipse.microprofile.rest.client.ext.DefaultClientHeadersFactoryImpl;
 import org.jboss.resteasy.reactive.client.spi.ResteasyReactiveClientRequestContext;
+import org.jboss.resteasy.reactive.client.spi.ResteasyReactiveClientRequestFilter;
 
 import io.quarkus.arc.Arc;
 import io.quarkus.rest.client.reactive.HeaderFiller;
 import io.quarkus.rest.client.reactive.ReactiveClientHeadersFactory;
 
 @Priority(Integer.MIN_VALUE)
-public class MicroProfileRestClientRequestFilter implements ClientRequestFilter {
+public class MicroProfileRestClientRequestFilter implements ResteasyReactiveClientRequestFilter {
 
     private static final MultivaluedMap<String, String> EMPTY_MAP = new MultivaluedHashMap<>();
 
@@ -31,7 +30,7 @@ public MicroProfileRestClientRequestFilter(ClientHeadersFactory clientHeadersFac
     }
 
     @Override
-    public void filter(ClientRequestContext requestContext) {
+    public void filter(ResteasyReactiveClientRequestContext requestContext) {
         HeaderFiller headerFiller = (HeaderFiller) requestContext.getProperty(HeaderFiller.class.getName());
 
         // mutable collection of headers
@@ -67,22 +66,22 @@ public void filter(ClientRequestContext requestContext) {
         if (clientHeadersFactory != null) {
             if (clientHeadersFactory instanceof ReactiveClientHeadersFactory) {
                 // reactive
-                ResteasyReactiveClientRequestContext reactiveRequestContext = (ResteasyReactiveClientRequestContext) requestContext;
                 ReactiveClientHeadersFactory reactiveClientHeadersFactory = (ReactiveClientHeadersFactory) clientHeadersFactory;
-                reactiveRequestContext.suspend();
-                MultivaluedMap<String, String> outgoingHeaders = incomingHeaders;
-                reactiveClientHeadersFactory.getHeaders(incomingHeaders).subscribe().with(newHeaders -> {
-                    outgoingHeaders.putAll(newHeaders);
-                    reactiveRequestContext.resume();
-                }, reactiveRequestContext::resume);
+                requestContext.suspend();
+                reactiveClientHeadersFactory.getHeaders(incomingHeaders, headers).subscribe().with(newHeaders -> {
+                    for (Map.Entry<String, List<String>> headerEntry : newHeaders.entrySet()) {
+                        requestContext.getHeaders().put(headerEntry.getKey(), castToListOfObjects(headerEntry.getValue()));
+                    }
+                    requestContext.resume();
+                }, requestContext::resume);
             } else {
                 // blocking
                 incomingHeaders = clientHeadersFactory.update(incomingHeaders, headers);
-            }
-        }
 
-        for (Map.Entry<String, List<String>> headerEntry : incomingHeaders.entrySet()) {
-            requestContext.getHeaders().put(headerEntry.getKey(), castToListOfObjects(headerEntry.getValue()));
+                for (Map.Entry<String, List<String>> headerEntry : incomingHeaders.entrySet()) {
+                    requestContext.getHeaders().put(headerEntry.getKey(), castToListOfObjects(headerEntry.getValue()));
+                }
+            }
         }
     }