Paho extension native support for Websocket Connections #1109

extensions/paho/deployment/src/main/java/org/apache/camel/quarkus/component/paho/deployment/PahoProcessor.java

@@ -16,54 +16,39 @@
  */
 package org.apache.camel.quarkus.component.paho.deployment;
 
-import java.util.Arrays;
-import java.util.List;
-
-import javax.inject.Inject;
-
-import io.quarkus.deployment.annotations.BuildProducer;
 import io.quarkus.deployment.annotations.BuildStep;
 import io.quarkus.deployment.builditem.FeatureBuildItem;
-import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.NativeImageResourceBundleBuildItem;
 import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem;
-import org.eclipse.paho.client.mqttv3.internal.SSLNetworkModuleFactory;
-import org.eclipse.paho.client.mqttv3.internal.TCPNetworkModuleFactory;
+import io.quarkus.deployment.builditem.nativeimage.ServiceProviderBuildItem;
 import org.eclipse.paho.client.mqttv3.logging.JSR47Logger;
 import org.eclipse.paho.client.mqttv3.spi.NetworkModuleFactory;
 
 class PahoProcessor {
-
     private static final String FEATURE = "camel-paho";
 
-    private static final List<Class<?>> PAHO_REFLECTIVE_CLASSES = Arrays.asList(
-            JSR47Logger.class,
-            TCPNetworkModuleFactory.class,
-            SSLNetworkModuleFactory.class);
-
-    @Inject
-    BuildProducer<NativeImageResourceBuildItem> resource;
-
-    @Inject
-    BuildProducer<NativeImageResourceBundleBuildItem> resourceBundle;
-
     @BuildStep
     FeatureBuildItem feature() {
         return new FeatureBuildItem(FEATURE);
     }
 
     @BuildStep
-    void registerForReflection(BuildProducer<ReflectiveClassBuildItem> reflectiveClass) {
-        for (Class<?> type : PAHO_REFLECTIVE_CLASSES) {
-            reflectiveClass.produce(
-                    new ReflectiveClassBuildItem(true, true, type));
-        }
+    ReflectiveClassBuildItem registerReflectiveClasses() {
+        return new ReflectiveClassBuildItem(true, true, JSR47Logger.class);
     }
 
     @BuildStep
-    void registerBundleResource() {
-        resource.produce(new NativeImageResourceBuildItem("META-INF/services/" + NetworkModuleFactory.class.getName()));
-        resourceBundle.produce(new NativeImageResourceBundleBuildItem("org.eclipse.paho.client.mqttv3.internal.nls.logcat"));
+    ServiceProviderBuildItem registerServiceProviders() {
+        return new ServiceProviderBuildItem(
+                NetworkModuleFactory.class.getName(),
+                org.eclipse.paho.client.mqttv3.internal.TCPNetworkModuleFactory.class.getName(),
+                org.eclipse.paho.client.mqttv3.internal.SSLNetworkModuleFactory.class.getName(),
+                org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketNetworkModuleFactory.class.getName(),
+                org.eclipse.paho.client.mqttv3.internal.websocket.WebSocketSecureNetworkModuleFactory.class.getName());
     }
 
+    @BuildStep
+    NativeImageResourceBundleBuildItem registerResourceBundle() {
+        return new NativeImageResourceBundleBuildItem("org.eclipse.paho.client.mqttv3.internal.nls.logcat");
+    }
 }

extensions/paho/runtime/pom.xml

@@ -54,6 +54,10 @@
             <groupId>org.apache.camel</groupId>
             <artifactId>camel-paho</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.graalvm.nativeimage</groupId>
+            <artifactId>svm</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

extensions/paho/runtime/src/main/java/org/apache/camel/quarkus/component/paho/graal/SubstituteBase64.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.graal;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+
+import com.oracle.svm.core.annotate.Substitute;
+import com.oracle.svm.core.annotate.TargetClass;
+
+@TargetClass(org.eclipse.paho.client.mqttv3.internal.websocket.Base64.class)
+final class SubstituteBase64 {
+    @Substitute
+    public static String encode(String s) {
+        return Base64.getEncoder().encodeToString(s.getBytes(StandardCharsets.UTF_8));
+    }
+
+    @Substitute
+    public static String encodeBytes(byte[] b) {
+        return Base64.getEncoder().encodeToString(b);
+    }
+}

integration-tests/messaging/src/main/java/org/apache/camel/quarkus/component/messaging/it/JmsResource.java

@@ -82,6 +82,21 @@ public Response producePahoMessage(@PathParam("queueName") String queueName, Str
         return Response.created(new URI("https://camel.apache.org/")).build();
     }
 
+    @Path("/paho-ws/{queueName}")
+    @GET
+    @Produces(MediaType.TEXT_PLAIN)
+    public String consumePahoMessageWs(@PathParam("queueName") String queueName) {
+        return consumerTemplate.receiveBody("paho:" + queueName, 5000, String.class);
+    }
+
+    @Path("/paho-ws/{queueName}")
+    @POST
+    @Consumes(MediaType.TEXT_PLAIN)
+    public Response producePahoMessageWs(@PathParam("queueName") String queueName, String message) throws Exception {
+        producerTemplate.sendBody("paho:" + queueName + "?retained=true&brokerUrl={{broker-url.ws}}", message);
+        return Response.created(new URI("https://camel.apache.org/")).build();
+    }
+
     // *****************************
     //
     // camel-sjms

integration-tests/messaging/src/test/java/org/apache/camel/quarkus/component/messaging/it/ActiveMQTestResource.java

@@ -35,30 +35,32 @@ public class ActiveMQTestResource implements QuarkusTestResourceLifecycleManager
     private static final String ACTIVEMQ_PASSWORD = "simetraehcapa";
     private static final int ACTIVEMQ_PORT = 61616;
 
-    private GenericContainer container;
+    private GenericContainer<?> container;
 
     @Override
     public Map<String, String> start() {
         LOGGER.info(TestcontainersConfiguration.getInstance().toString());
 
         try {
-            container = new GenericContainer(ACTIVEMQ_IMAGE)
+            container = new GenericContainer<>(ACTIVEMQ_IMAGE)
                     .withExposedPorts(ACTIVEMQ_PORT)
                     .withLogConsumer(new Slf4jLogConsumer(LOGGER))
                     .withEnv("BROKER_CONFIG_MAX_DISK_USAGE", "100")
                     .waitingFor(Wait.forListeningPort());
 
             container.start();
 
-            String brokerUrl = String.format("tcp://127.0.0.1:%d", container.getMappedPort(ACTIVEMQ_PORT));
+            String brokerUrlTcp = String.format("tcp://127.0.0.1:%d", container.getMappedPort(ACTIVEMQ_PORT));
+            String brokerUrlWs = String.format("ws://127.0.0.1:%d", container.getMappedPort(ACTIVEMQ_PORT));
 
             return CollectionHelper.mapOf(
-                    "quarkus.artemis.url", brokerUrl,
+                    "quarkus.artemis.url", brokerUrlTcp,
                     "quarkus.artemis.username", ACTIVEMQ_USERNAME,
                     "quarkus.artemis.password", ACTIVEMQ_PASSWORD,
-                    "camel.component.paho.brokerUrl", brokerUrl,
+                    "camel.component.paho.brokerUrl", brokerUrlTcp,
                     "camel.component.paho.username", ACTIVEMQ_USERNAME,
-                    "camel.component.paho.password", ACTIVEMQ_PASSWORD);
+                    "camel.component.paho.password", ACTIVEMQ_PASSWORD,
+                    "broker-url.ws", brokerUrlWs);
 
         } catch (Exception e) {
             throw new RuntimeException(e);

integration-tests/messaging/src/test/java/org/apache/camel/quarkus/component/messaging/it/JmsTest.java

@@ -30,7 +30,7 @@
 class JmsTest {
 
     @ParameterizedTest
-    @ValueSource(strings = { "jms", "paho", "sjms" })
+    @ValueSource(strings = { "jms", "paho", "paho-ws", "sjms" })
     public void testJmsComponent(String component) {
         String message = "Hello Camel Quarkus " + component;