Fix apache#3656 Improve camel-quarkus-paho-mqtt5 test coverage (apach…

…e#3709)

extensions/paho-mqtt5/deployment/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/deployment/PahoMqtt5Processor.java

@@ -16,11 +16,18 @@
  */
 package org.apache.camel.quarkus.component.paho.mqtt5.deployment;
 
+import java.io.RandomAccessFile;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.util.ResourceBundle;
+
+import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBundleBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
+import org.eclipse.paho.mqttv5.client.internal.ResourceBundleCatalog;
 import org.eclipse.paho.mqttv5.client.internal.SSLNetworkModuleFactory;
 import org.eclipse.paho.mqttv5.client.internal.TCPNetworkModuleFactory;
 import org.eclipse.paho.mqttv5.client.logging.JSR47Logger;
@@ -38,8 +45,14 @@ FeatureBuildItem feature() {
     }
 
     @BuildStep
-    ReflectiveClassBuildItem registerReflectiveClasses() {
-        return new ReflectiveClassBuildItem(false, false, JSR47Logger.class);
+    void registerReflectiveClasses(BuildProducer<ReflectiveClassBuildItem> p) {
+        p.produce(new ReflectiveClassBuildItem(false, false, JSR47Logger.class));
+        p.produce(new ReflectiveClassBuildItem(false, false, ResourceBundleCatalog.class));
+        p.produce(new ReflectiveClassBuildItem(false, false, ResourceBundle.class));
+        p.produce(new ReflectiveClassBuildItem(false, false, FileLock.class));
+        p.produce(new ReflectiveClassBuildItem(true, false, FileChannel.class));
+        p.produce(new ReflectiveClassBuildItem(true, false, RandomAccessFile.class));
+        p.produce(new ReflectiveClassBuildItem(true, false, "sun.nio.ch.FileLockImpl"));
     }
 
     @BuildStep
@@ -53,8 +66,9 @@ ServiceProviderBuildItem registerServiceProviders() {
     }
 
     @BuildStep()
-    NativeImageResourceBundleBuildItem hapiMessages() {
-        return new NativeImageResourceBundleBuildItem("org.eclipse.paho.mqttv5.client.internal.nls.logcat");
+    void registerResourceBundle(BuildProducer<NativeImageResourceBundleBuildItem> p) {
+        p.produce(new NativeImageResourceBundleBuildItem("org.eclipse.paho.mqttv5.client.internal.nls.logcat"));
+        p.produce(new NativeImageResourceBundleBuildItem("org.eclipse.paho.mqttv5.common.nls.messages"));
     }
 
 }

integration-tests/paho-mqtt5/pom.xml

@@ -35,6 +35,14 @@
             <groupId>org.apache.camel.quarkus</groupId>
             <artifactId>camel-quarkus-paho-mqtt5</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-direct</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-mock</artifactId>
+        </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
             <artifactId>quarkus-resteasy</artifactId>
@@ -73,6 +81,11 @@
             <artifactId>quarkus-junit4-mock</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.camel.quarkus</groupId>
+            <artifactId>camel-quarkus-integration-test-support</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <profiles>
@@ -112,6 +125,32 @@
             </activation>
             <dependencies>
                 <!-- The following dependencies guarantee that this module is built after them. You can update them by running `mvn process-resources -Pformat -N` from the source tree root directory -->
+                <dependency>
+                    <groupId>org.apache.camel.quarkus</groupId>
+                    <artifactId>camel-quarkus-direct-deployment</artifactId>
+                    <version>${project.version}</version>
+                    <type>pom</type>
+                    <scope>test</scope>
+                    <exclusions>
+                        <exclusion>
+                            <groupId>*</groupId>
+                            <artifactId>*</artifactId>
+                        </exclusion>
+                    </exclusions>
+                </dependency>
+                <dependency>
+                    <groupId>org.apache.camel.quarkus</groupId>
+                    <artifactId>camel-quarkus-mock-deployment</artifactId>
+                    <version>${project.version}</version>
+                    <type>pom</type>
+                    <scope>test</scope>
+                    <exclusions>
+                        <exclusion>
+                            <groupId>*</groupId>
+                            <artifactId>*</artifactId>
+                        </exclusion>
+                    </exclusions>
+                </dependency>
                 <dependency>
                     <groupId>org.apache.camel.quarkus</groupId>
                     <artifactId>camel-quarkus-paho-mqtt5-deployment</artifactId>

integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/Counter.java

@@ -0,0 +1,36 @@
+/*
+ * 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.quarkus.component.paho.mqtt5.it;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import javax.enterprise.context.ApplicationScoped;
+
+@ApplicationScoped
+public class Counter {
+    CountDownLatch counter = new CountDownLatch(1);
+
+    public void countDown() {
+        counter.countDown();
+    }
+
+    public void await(int timeout, TimeUnit unit) throws Exception {
+        counter.await(timeout, unit);
+    }
+}

integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Resource.java

@@ -16,41 +16,67 @@
  */
 package org.apache.camel.quarkus.component.paho.mqtt5.it;
 
+import java.io.InputStream;
 import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.concurrent.TimeUnit;
 
 import javax.enterprise.context.ApplicationScoped;
 import javax.inject.Inject;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
 import javax.ws.rs.PathParam;
 import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 
+import org.apache.camel.CamelContext;
 import org.apache.camel.ConsumerTemplate;
 import org.apache.camel.ProducerTemplate;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.component.paho.mqtt5.PahoMqtt5Constants;
+import org.apache.camel.spi.RouteController;
 import org.eclipse.microprofile.config.ConfigProvider;
 
 @Path("/paho-mqtt5")
 @ApplicationScoped
 public class PahoMqtt5Resource {
 
+    @Inject
+    CamelContext context;
+
+    @Inject
+    Counter counter;
+
     @Inject
     ProducerTemplate producerTemplate;
 
     @Inject
     ConsumerTemplate consumerTemplate;
 
+    private final String keystore = "clientkeystore.jks";
+    private final String password = "quarkus";
+
     @Path("/{protocol}/{queueName}")
     @GET
     @Produces(MediaType.TEXT_PLAIN)
     public String consumePahoMessage(
             @PathParam("protocol") String protocol,
             @PathParam("queueName") String queueName) {
-        return consumerTemplate.receiveBody("paho-mqtt5:" + queueName + "?brokerUrl=" + brokerUrl(protocol), 5000,
+        if ("ssl".equals(protocol)) {
+            setKeyStore(keystore, password);
+        }
+        String result = consumerTemplate.receiveBody("paho-mqtt5:" + queueName + "?brokerUrl=" + brokerUrl(protocol), 5000,
                 String.class);
+        if ("ssl".equals(protocol)) {
+            removeKeyStore(keystore);
+        }
+        return result;
     }
 
     @Path("/{protocol}/{queueName}")
@@ -60,12 +86,102 @@ public Response producePahoMessage(
             @PathParam("protocol") String protocol,
             @PathParam("queueName") String queueName,
             String message) throws Exception {
-        producerTemplate.sendBody("paho-mqtt5:" + queueName + "?retained=true&brokerUrl=" + brokerUrl(protocol), message);
+        if ("ssl".equals(protocol)) {
+            setKeyStore(keystore, password);
+        }
+        try {
+            producerTemplate.sendBody("paho-mqtt5:" + queueName + "?retained=true&brokerUrl=" + brokerUrl(protocol), message);
+        } finally {
+            if ("ssl".equals(protocol)) {
+                removeKeyStore(keystore);
+            }
+        }
+        return Response.created(new URI("https://camel.apache.org/")).build();
+    }
+
+    @Path("/override/{queueName}")
+    @POST
+    @Consumes(MediaType.TEXT_PLAIN)
+    public Response overrideQueueName(
+            @PathParam("queueName") String queueName,
+            String message) throws Exception {
+        producerTemplate.sendBodyAndHeader("paho-mqtt5:test?retained=true&brokerUrl=" + brokerUrl("tcp"), message,
+                PahoMqtt5Constants.CAMEL_PAHO_OVERRIDE_TOPIC, queueName);
+        return Response.created(new URI("https://camel.apache.org/")).build();
+    }
+
+    @Path("/readThenWriteWithFilePersistenceShouldSucceed")
+    @GET
+    @Produces(MediaType.TEXT_PLAIN)
+    public String readThenWriteWithFilePersistenceShouldSucceed(@QueryParam("message") String message) throws Exception {
+        producerTemplate.sendBody(
+                "paho-mqtt5:withFilePersistence?retained=true&persistence=FILE&brokerUrl="
+                        + brokerUrl("tcp"),
+                message);
+        return consumerTemplate.receiveBody(
+                "paho-mqtt5:withFilePersistence?persistence=FILE&brokerUrl=" + brokerUrl("tcp"),
+                5000,
+                String.class);
+    }
+
+    @Path("/routeStatus/{id}")
+    @GET
+    @Produces(MediaType.TEXT_PLAIN)
+    public String routeStatus(@PathParam("id") String routeId,
+            @QueryParam("waitForContainerStarted") @DefaultValue("false") boolean wait) throws Exception {
+        RouteController routeController = context.getRouteController();
+        if (wait) {
+            counter.await(30, TimeUnit.SECONDS);
+        }
+        return routeController.getRouteStatus(routeId).name();
+    }
+
+    @Path("/mock")
+    @POST
+    @Produces(MediaType.TEXT_PLAIN)
+    public String mock(String message) throws Exception {
+        MockEndpoint endpoint = context.getEndpoint("mock:test", MockEndpoint.class);
+        endpoint.expectedBodiesReceived(message);
+
+        endpoint.assertIsSatisfied();
+        return "OK";
+    }
+
+    @Path("/send")
+    @POST
+    @Produces(MediaType.TEXT_PLAIN)
+    public Response send(String message) throws Exception {
+        producerTemplate.sendBody("direct:test", message);
         return Response.created(new URI("https://camel.apache.org/")).build();
     }
 
     private String brokerUrl(String protocol) {
         return ConfigProvider.getConfig().getValue("paho5.broker." + protocol + ".url", String.class);
     }
 
+    private void setKeyStore(String keystore, String password) {
+        InputStream in = Thread.currentThread().getContextClassLoader().getResourceAsStream(keystore);
+
+        try {
+            Files.copy(in, Paths.get(keystore));
+        } catch (Exception e) {
+        }
+
+        System.setProperty("javax.net.ssl.keyStore", keystore);
+        System.setProperty("javax.net.ssl.keyStorePassword", password);
+        System.setProperty("javax.net.ssl.trustStore", keystore);
+        System.setProperty("javax.net.ssl.trustStorePassword", password);
+    }
+
+    private void removeKeyStore(String keystore) {
+        try {
+            Files.delete(Paths.get(keystore));
+        } catch (Exception e) {
+        }
+
+        System.clearProperty("javax.net.ssl.keyStore");
+        System.clearProperty("javax.net.ssl.keyStorePassword");
+        System.clearProperty("javax.net.ssl.trustStore");
+        System.clearProperty("javax.net.ssl.trustStorePassword");
+    }
 }

integration-tests/paho-mqtt5/src/main/java/org/apache/camel/quarkus/component/paho/mqtt5/it/PahoMqtt5Route.java

@@ -0,0 +1,57 @@
+/*
+ * 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.quarkus.component.paho.mqtt5.it;
+
+import javax.enterprise.context.ApplicationScoped;
+import javax.inject.Inject;
+
+import org.apache.camel.Route;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.SupervisingRouteController;
+import org.apache.camel.support.RoutePolicySupport;
+import org.eclipse.microprofile.config.ConfigProvider;
+
+@ApplicationScoped
+public class PahoMqtt5Route extends RouteBuilder {
+    public static final String TESTING_ROUTE_ID = "testingRoute";
+
+    @Inject
+    Counter counter;
+
+    @Override
+    public void configure() throws Exception {
+        SupervisingRouteController supervising = getCamelContext().getRouteController().supervising();
+        supervising.setBackOffDelay(200);
+        supervising.setIncludeRoutes("paho-mqtt5:*");
+
+        from("direct:test").to("paho-mqtt5:queue?lazyStartProducer=true&brokerUrl=" + brokerUrl("tcp"));
+        from("paho-mqtt5:queue?brokerUrl=" + brokerUrl("tcp"))
+                .id(TESTING_ROUTE_ID)
+                .routePolicy(new RoutePolicySupport() {
+                    @Override
+                    public void onStart(Route route) {
+                        counter.countDown();
+                    }
+                })
+                .to("mock:test");
+    }
+
+    private String brokerUrl(String protocol) {
+        return ConfigProvider.getConfig().getValue("paho5.broker." + protocol + ".url", String.class);
+    }
+}

integration-tests/paho-mqtt5/src/main/resources/application.properties

@@ -0,0 +1,17 @@
+## ---------------------------------------------------------------------------
+## 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.
+## ---------------------------------------------------------------------------
+quarkus.native.resources.includes=*.jks