CAMEL-23351: camel-oaipmh: Add custom HTTP Headers

components/camel-oaipmh/src/generated/java/org/apache/camel/oaipmh/component/OAIPMHEndpointConfigurer.java

@@ -38,6 +38,8 @@ public boolean configure(CamelContext camelContext, Object obj, String name, Obj
         case "exchangePattern": target.setExchangePattern(property(camelContext, org.apache.camel.ExchangePattern.class, value)); return true;
         case "from": target.setFrom(property(camelContext, java.lang.String.class, value)); return true;
         case "greedy": target.setGreedy(property(camelContext, boolean.class, value)); return true;
+        case "httpheaders":
+        case "httpHeaders": target.setHttpHeaders(property(camelContext, java.util.Map.class, value)); return true;
         case "identifier": target.setIdentifier(property(camelContext, java.lang.String.class, value)); return true;
         case "ignoresslwarnings":
         case "ignoreSSLWarnings": target.setIgnoreSSLWarnings(property(camelContext, boolean.class, value)); return true;
@@ -94,6 +96,8 @@ public Class<?> getOptionType(String name, boolean ignoreCase) {
         case "exchangePattern": return org.apache.camel.ExchangePattern.class;
         case "from": return java.lang.String.class;
         case "greedy": return boolean.class;
+        case "httpheaders":
+        case "httpHeaders": return java.util.Map.class;
         case "identifier": return java.lang.String.class;
         case "ignoresslwarnings":
         case "ignoreSSLWarnings": return boolean.class;
@@ -151,6 +155,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
         case "exchangePattern": return target.getExchangePattern();
         case "from": return target.getFrom();
         case "greedy": return target.isGreedy();
+        case "httpheaders":
+        case "httpHeaders": return target.getHttpHeaders();
         case "identifier": return target.getIdentifier();
         case "ignoresslwarnings":
         case "ignoreSSLWarnings": return target.isIgnoreSSLWarnings();
@@ -192,6 +198,8 @@ public Object getOptionValue(Object obj, String name, boolean ignoreCase) {
     @Override
     public Object getCollectionValueType(Object target, String name, boolean ignoreCase) {
         switch (ignoreCase ? name.toLowerCase() : name) {
+        case "httpheaders":
+        case "httpHeaders": return java.lang.String.class;
         case "schedulerproperties":
         case "schedulerProperties": return java.lang.Object.class;
         default: return null;

components/camel-oaipmh/src/generated/java/org/apache/camel/oaipmh/component/OAIPMHEndpointUriFactory.java

@@ -23,7 +23,7 @@ public class OAIPMHEndpointUriFactory extends org.apache.camel.support.component
     private static final Set<String> SECRET_PROPERTY_NAMES;
     private static final Map<String, String> MULTI_VALUE_PREFIXES;
     static {
-        Set<String> props = new HashSet<>(30);
+        Set<String> props = new HashSet<>(31);
         props.add("backoffErrorThreshold");
         props.add("backoffIdleThreshold");
         props.add("backoffMultiplier");
@@ -34,6 +34,7 @@ public class OAIPMHEndpointUriFactory extends org.apache.camel.support.component
         props.add("exchangePattern");
         props.add("from");
         props.add("greedy");
+        props.add("httpHeaders");
         props.add("identifier");
         props.add("ignoreSSLWarnings");
         props.add("initialDelay");
@@ -56,7 +57,8 @@ public class OAIPMHEndpointUriFactory extends org.apache.camel.support.component
         props.add("verb");
         PROPERTY_NAMES = Collections.unmodifiableSet(props);
         SECRET_PROPERTY_NAMES = Collections.emptySet();
-        Map<String, String> prefixes = new HashMap<>(1);
+        Map<String, String> prefixes = new HashMap<>(2);
+        prefixes.put("httpHeaders", "httpHeader.");
         prefixes.put("schedulerProperties", "scheduler.");
         MULTI_VALUE_PREFIXES = Collections.unmodifiableMap(prefixes);
     }

components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/component/OAIPMHConsumer.java

@@ -51,6 +51,9 @@ protected void doStart() throws Exception {
         if (getEndpoint().isIgnoreSSLWarnings()) {
             this.harvester.getHttpClient().setIgnoreSSLWarnings(true);
         }
+        if (getEndpoint().getHttpHeaders() != null && !getEndpoint().getHttpHeaders().isEmpty()) {
+            this.harvester.getHttpClient().setHttpHeaders(getEndpoint().getHttpHeaders());
+        }
         super.doStart();
     }
 

components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/component/OAIPMHEndpoint.java

@@ -17,6 +17,7 @@
 package org.apache.camel.oaipmh.component;
 
 import java.net.URI;
+import java.util.HashMap;
 import java.util.Map;
 
 import org.apache.camel.Category;
@@ -75,6 +76,13 @@ public class OAIPMHEndpoint extends ScheduledPollEndpoint implements EndpointSer
               description = "Returns the response of a single request. Otherwise it will make requests until there is no more data to return.")
     private boolean onlyFirst;
 
+    @UriParam(label = "advanced",
+              description = "Custom HTTP headers to send with each request to the OAI-PMH repository, "
+                            + "for example for Authorization or Accept-Language.",
+              javaType = "java.util.Map<java.lang.String, java.lang.String>",
+              prefix = "httpHeader.", multiValue = true)
+    private Map<String, String> httpHeaders;
+
     private Map<String, Object> queryParameters;
 
     public OAIPMHEndpoint(String uri, String remaining, OAIPMHComponent component) {
@@ -103,6 +111,27 @@ protected void doInit() throws Exception {
 
         validateParameters();
 
+        // Collect httpHeader.* lenient params into the httpHeaders map so they become HTTP headers,
+        // not URL query parameters. This handles the URI-string form: ?httpHeader.Authorization=token
+        if (queryParameters != null) {
+            Map<String, String> fromUri = new HashMap<>();
+            queryParameters.entrySet().removeIf(entry -> {
+                if (entry.getKey().startsWith("httpHeader.")) {
+                    fromUri.put(entry.getKey().substring("httpHeader.".length()), String.valueOf(entry.getValue()));
+                    return true;
+                }
+                return false;
+            });
+            if (!fromUri.isEmpty()) {
+                if (httpHeaders == null) {
+                    httpHeaders = fromUri;
+                } else {
+                    fromUri.putAll(httpHeaders);
+                    httpHeaders = fromUri;
+                }
+            }
+        }
+
         // build uri from parameters
         String prefix = "";
         if (!baseUrl.startsWith("http:") && !baseUrl.startsWith("https:")) {
@@ -223,4 +252,12 @@ public void setOnlyFirst(boolean onlyFist) {
         this.onlyFirst = onlyFist;
     }
 
+    public Map<String, String> getHttpHeaders() {
+        return httpHeaders;
+    }
+
+    public void setHttpHeaders(Map<String, String> httpHeaders) {
+        this.httpHeaders = httpHeaders;
+    }
+
 }

components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/component/OAIPMHProducer.java

@@ -18,6 +18,7 @@
 
 import java.net.URI;
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 
 import org.apache.camel.Exchange;
@@ -50,6 +51,9 @@ public void process(Exchange exchange) throws Exception {
                 endpoint.getFrom(),
                 endpoint.getSet(),
                 endpoint.getIdentifier());
+        if (endpoint.getHttpHeaders() != null && !endpoint.getHttpHeaders().isEmpty()) {
+            harvester.getHttpClient().setHttpHeaders(endpoint.getHttpHeaders());
+        }
         overrideHarvesterConfigs(exchange.getIn(), harvester);
         if (endpoint.isIgnoreSSLWarnings()) {
             harvester.getHttpClient().setIgnoreSSLWarnings(true);
@@ -75,6 +79,11 @@ private void overrideHarvesterConfigs(Message msg, Harvester harvester) {
         checkAndSetConfigs(msg, OAIPMHConstants.RESUMPTION_TOKEN, harvester::setResumptionToken, String.class);
         checkAndSetConfigs(msg, OAIPMHConstants.ONLY_FIRST, endpoint::setOnlyFirst, Boolean.class);
         checkAndSetConfigs(msg, OAIPMHConstants.IGNORE_SSL_WARNINGS, endpoint::setIgnoreSSLWarnings, Boolean.class);
+        @SuppressWarnings("unchecked")
+        Map<String, String> httpHeadersOverride = msg.getHeader(OAIPMHConstants.HTTP_HEADERS, Map.class);
+        if (httpHeadersOverride != null && !httpHeadersOverride.isEmpty()) {
+            harvester.getHttpClient().setHttpHeaders(httpHeadersOverride);
+        }
     }
 
     private <T> void checkAndSetConfigs(final Message message, final String key, final Consumer<T> fn, final Class<T> type) {

components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/component/model/OAIPMHConstants.java

@@ -19,6 +19,7 @@
 import org.apache.camel.spi.Metadata;
 
 public final class OAIPMHConstants {
+
     @Metadata(label = "producer", description = "This header is obtained when onlyFirst option is enable. " +
                                                 "Return resumption token of the request when data is still available.",
               javaType = "String")
@@ -34,6 +35,11 @@ public final class OAIPMHConstants {
     public static final String SET = "CamelOaimphSet";
     public static final String IDENTIFIER = "CamelOaimphIdentifier";
 
+    @Metadata(label = "producer",
+              description = "Custom HTTP headers to send with the request, overriding any httpHeader.* endpoint parameters.",
+              javaType = "java.util.Map<java.lang.String, java.lang.String>")
+    public static final String HTTP_HEADERS = "CamelOaimphHttpHeaders";
+
     private OAIPMHConstants() {
     }
 

components/camel-oaipmh/src/main/java/org/apache/camel/oaipmh/utils/OAIPMHHttpClient.java

@@ -24,6 +24,8 @@
 import java.security.KeyManagementException;
 import java.security.KeyStoreException;
 import java.security.NoSuchAlgorithmException;
+import java.util.Collections;
+import java.util.Map;
 
 import org.apache.hc.client5.http.ClientProtocolException;
 import org.apache.hc.client5.http.classic.methods.HttpGet;
@@ -50,6 +52,7 @@ public class OAIPMHHttpClient {
     private static final Logger LOG = LoggerFactory.getLogger(OAIPMHHttpClient.class);
 
     private boolean ignoreSSLWarnings;
+    private Map<String, String> httpHeaders = Collections.emptyMap();
 
     public String doRequest(
             URI baseURI, String verb, String set, String from, String until, String metadataPrefix, String token,
@@ -87,6 +90,7 @@ public String doRequest(
             }
 
             HttpGet httpget = new HttpGet(builder.build());
+            httpHeaders.forEach(httpget::addHeader);
 
             LOG.info("Executing request: {} ", httpget);
 
@@ -144,4 +148,12 @@ public void setIgnoreSSLWarnings(boolean ignoreSSLWarnings) {
         this.ignoreSSLWarnings = ignoreSSLWarnings;
     }
 
+    public Map<String, String> getHttpHeaders() {
+        return httpHeaders;
+    }
+
+    public void setHttpHeaders(Map<String, String> httpHeaders) {
+        this.httpHeaders = httpHeaders != null ? httpHeaders : Collections.emptyMap();
+    }
+
 }

components/camel-oaipmh/src/test/java/org/apache/camel/oaipmh/OAIPMHConsumerHttpHeadersTest.java

@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.oaipmh;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.oaipmh.utils.MockOaipmhServer;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
+import static org.awaitility.Awaitility.await;
+
+public class OAIPMHConsumerHttpHeadersTest extends CamelTestSupport {
+
+    private static MockOaipmhServer mockOaipmhServer;
+
+    @BeforeAll
+    public static void startServer() {
+        mockOaipmhServer = MockOaipmhServer.create();
+        mockOaipmhServer.start();
+    }
+
+    @AfterAll
+    public static void stopServer() {
+        mockOaipmhServer.stop();
+    }
+
+    @Test
+    public void testConsumerSendsCustomHeaders() {
+        await().atMost(15, TimeUnit.SECONDS).untilAsserted(() -> mockOaipmhServer.getServer().verify(
+                getRequestedFor(urlMatching("/oai/request.*"))
+                        .withHeader("X-Custom-Header", equalTo("test-value"))));
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            public void configure() {
+                from("oaipmh://localhost:" + mockOaipmhServer.getHttpPort()
+                     + "/oai/request?initialDelay=100&delay=60000"
+                     + "&httpHeader.X-Custom-Header=test-value")
+                        .to("mock:result");
+            }
+        };
+    }
+}

components/camel-oaipmh/src/test/java/org/apache/camel/oaipmh/OAIPMHProducerHttpHeadersTest.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.oaipmh;
+
+import java.util.Map;
+
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.oaipmh.component.model.OAIPMHConstants;
+import org.apache.camel.oaipmh.utils.MockOaipmhServer;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.getRequestedFor;
+import static com.github.tomakehurst.wiremock.client.WireMock.urlMatching;
+
+public class OAIPMHProducerHttpHeadersTest extends CamelTestSupport {
+
+    private static MockOaipmhServer mockOaipmhServer;
+
+    @BeforeAll
+    public static void startServer() {
+        mockOaipmhServer = MockOaipmhServer.create();
+        mockOaipmhServer.start();
+    }
+
+    @AfterAll
+    public static void stopServer() {
+        mockOaipmhServer.stop();
+    }
+
+    @BeforeEach
+    public void resetRequests() {
+        mockOaipmhServer.getServer().resetRequests();
+    }
+
+    @Test
+    public void testEndpointHttpHeadersSentToServer() throws Exception {
+        template.sendBody("direct:start", "foo");
+
+        mockOaipmhServer.getServer().verify(
+                getRequestedFor(urlMatching("/oai/request.*"))
+                        .withHeader("Authorization", equalTo("test-token"))
+                        .withHeader("X-Custom", equalTo("hello")));
+    }
+
+    @Test
+    public void testExchangeHeaderOverridesEndpointHeaders() throws Exception {
+        template.sendBodyAndHeader("direct:start-override", "foo",
+                OAIPMHConstants.HTTP_HEADERS, Map.of("Authorization", "runtime-token"));
+
+        mockOaipmhServer.getServer().verify(
+                getRequestedFor(urlMatching("/oai/request.*"))
+                        .withHeader("Authorization", equalTo("runtime-token")));
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            public void configure() {
+                from("direct:start")
+                        .to("oaipmh://localhost:" + mockOaipmhServer.getHttpPort()
+                            + "/oai/request?onlyFirst=true"
+                            + "&httpHeader.Authorization=test-token"
+                            + "&httpHeader.X-Custom=hello");
+
+                from("direct:start-override")
+                        .to("oaipmh://localhost:" + mockOaipmhServer.getHttpPort()
+                            + "/oai/request?onlyFirst=true"
+                            + "&httpHeader.Authorization=endpoint-token");
+            }
+        };
+    }
+}

components/camel-oaipmh/src/test/java/org/apache/camel/oaipmh/utils/MockOaipmhServer.java

@@ -129,6 +129,10 @@ public int getHttpsPort() {
         return this.httpsPort;
     }
 
+    public WireMockServer getServer() {
+        return server;
+    }
+
     public static final class OaipmhMockTransformer extends ResponseDefinitionTransformer {
 
         @Override