Rest Client Reactive: Add reactive flavor for ClientHeadersFactory

Added reactive flavor of `ClientHeadersFactory` that allows doing blocking operations. For example:

[source, java]
----
package org.acme.rest.client;

import org.eclipse.microprofile.rest.client.ext.ClientHeadersFactory;

import javax.enterprise.context.ApplicationScoped;
import javax.ws.rs.core.MultivaluedHashMap;
import javax.ws.rs.core.MultivaluedMap;
import java.util.UUID;

@ApplicationScoped
public class GetTokenReactiveClientHeadersFactory extends ReactiveClientHeadersFactory {

    @Inject
    Service service;

    @OverRide
    public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
        return Uni.createFrom().item(() -> {
            MultivaluedHashMap<String, String> newHeaders = new MultivaluedHashMap<>();
            // perform blocking call
            newHeaders.add(HEADER_NAME, service.getToken());
            return newHeaders;
        });
    }
}
----

Fixes quarkusio#20001

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

@@ -515,7 +515,7 @@ More details about this can be found in https://smallrye.io/smallrye-mutiny/#_un
 
 There are a few ways in which you can specify custom headers for your REST calls:
 
-- by registering a `ClientHeadersFactory` with the `@RegisterClientHeaders` annotation
+- by registering a `ClientHeadersFactory` or a `ReactiveClientHeadersFactory` with the `@RegisterClientHeaders` annotation
 - by specifying the value of the header with `@ClientHeaderParam`
 - by specifying the value of the header by `@HeaderParam`
 
@@ -591,6 +591,37 @@ To specify a value for `${header.value}`, simply put the following in your `appl
 header.value=value of the header
 ----
 
+Also, there is a reactive flavor of `ClientHeadersFactory` that allows doing blocking operations. For example:
+
+[source, java]
+----
+package org.acme.rest.client;
+
+import org.eclipse.microprofile.rest.client.ext.ClientHeadersFactory;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.ws.rs.core.MultivaluedHashMap;
+import javax.ws.rs.core.MultivaluedMap;
+import java.util.UUID;
+
+@ApplicationScoped
+public class GetTokenReactiveClientHeadersFactory extends ReactiveClientHeadersFactory {
+
+    @Inject
+    Service service;
+
+    @Override
+    public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
+        return Uni.createFrom().item(() -> {
+            MultivaluedHashMap<String, String> newHeaders = new MultivaluedHashMap<>();
+            // perform blocking call
+            newHeaders.add(HEADER_NAME, service.getToken());
+            return newHeaders;
+        });
+    }
+}
+----
+
 === Default header factory
 
 The `@RegisterClientHeaders` annotation can also be used without any custom factory specified. In that case the `DefaultClientHeadersFactoryImpl` factory will be used.

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

@@ -0,0 +1,86 @@
+package io.quarkus.rest.client.reactive.headers;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.net.URI;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.Path;
+import javax.ws.rs.core.MultivaluedHashMap;
+import javax.ws.rs.core.MultivaluedMap;
+
+import org.eclipse.microprofile.rest.client.RestClientBuilder;
+import org.eclipse.microprofile.rest.client.annotation.RegisterClientHeaders;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.rest.client.reactive.ReactiveClientHeadersFactory;
+import io.quarkus.test.QuarkusUnitTest;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.smallrye.common.annotation.Blocking;
+import io.smallrye.mutiny.Uni;
+
+public class ReactiveClientHeadersFromProviderTest {
+    private static final String HEADER_NAME = "my-header";
+    private static final String HEADER_VALUE = "oifajrofijaeoir5gjaoasfaxcvcz";
+
+    @TestHTTPResource
+    URI baseUri;
+
+    @RegisterExtension
+    static final QuarkusUnitTest config = new QuarkusUnitTest()
+            .withApplicationRoot((jar) -> jar.addClasses(ReactiveClientHeadersFromProviderTest.Client.class)
+                    .addAsResource(
+                            new StringAsset("my.property-value=" + HEADER_VALUE),
+                            "application.properties"));
+
+    @Test
+    void shouldSetHeaderFromProperties() {
+        ReactiveClientHeadersFromProviderTest.Client client = RestClientBuilder.newBuilder().baseUri(baseUri)
+                .build(ReactiveClientHeadersFromProviderTest.Client.class);
+
+        assertThat(client.getWithHeader()).isEqualTo(HEADER_VALUE);
+    }
+
+    @Path("/")
+    @ApplicationScoped
+    public static class Resource {
+        @GET
+        public String returnHeaderValue(@HeaderParam(HEADER_NAME) String header) {
+            return header;
+        }
+    }
+
+    @ApplicationScoped
+    public static class Service {
+        @Blocking
+        public String getValue() {
+            return HEADER_VALUE;
+        }
+    }
+
+    @RegisterClientHeaders(CustomReactiveClientHeadersFactory.class)
+    public interface Client {
+        @GET
+        String getWithHeader();
+    }
+
+    public static class CustomReactiveClientHeadersFactory extends ReactiveClientHeadersFactory {
+
+        @Inject
+        Service service;
+
+        @Override
+        public Uni<MultivaluedMap<String, String>> getHeaders(MultivaluedMap<String, String> incomingHeaders) {
+            return Uni.createFrom().item(() -> {
+                MultivaluedHashMap<String, String> newHeaders = new MultivaluedHashMap<>();
+                newHeaders.add(HEADER_NAME, service.getValue());
+                return newHeaders;
+            });
+        }
+    }
+}

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

@@ -0,0 +1,20 @@
+package io.quarkus.rest.client.reactive;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import org.eclipse.microprofile.rest.client.ext.ClientHeadersFactory;
+
+import io.smallrye.mutiny.Uni;
+
+/**
+ * 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);
+
+    @Override
+    public final MultivaluedMap<String, String> update(MultivaluedMap<String, String> incomingHeaders,
+            MultivaluedMap<String, String> clientOutgoingHeaders) {
+        throw new RuntimeException("Can't call `update` method in a Reactive context. Use `getHeaders` or implements ClientHeadersFactory.");
+    }
+}

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

@@ -13,9 +13,11 @@
 
 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 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 {
@@ -60,12 +62,20 @@ public void filter(ClientRequestContext requestContext) {
             for (Map.Entry<String, List<String>> headerEntry : headers.entrySet()) {
                 requestContext.getHeaders().put(headerEntry.getKey(), castToListOfObjects(headerEntry.getValue()));
             }
-        }
-
-        if (clientHeadersFactory != null) {
+        } else if (clientHeadersFactory instanceof ReactiveClientHeadersFactory
+                && requestContext instanceof ResteasyReactiveClientRequestContext) {
+            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);
+        } else if (clientHeadersFactory != null) {
             incomingHeaders = clientHeadersFactory.update(incomingHeaders, headers);
         }
-
+        
         for (Map.Entry<String, List<String>> headerEntry : incomingHeaders.entrySet()) {
             requestContext.getHeaders().put(headerEntry.getKey(), castToListOfObjects(headerEntry.getValue()));
         }