All v3 tests green for HiveMQ client

Author: skobow

mqtt-hivemq/build.gradle.kts

@@ -3,6 +3,9 @@ plugins {
 }
 
 dependencies {
+    api(projects.micronautMqttSsl)
+
+    implementation(libs.bcpkix.jdk15on)
     api(libs.managed.hivemq.client)
 }
 

mqtt-hivemq/src/main/java/io/micronaut/mqtt/hivemq/bind/MqttMessage.java

@@ -0,0 +1,156 @@
+/*
+ * Copyright 2017-2023 original authors
+ *
+ * Licensed 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
+ *
+ * https://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 io.micronaut.mqtt.hivemq.bind;
+
+import java.util.List;
+
+/**
+ * A MQTT message holds the payload and options
+ * for message delivery.
+ *
+ * @author Sven Kobow
+ */
+public class MqttMessage {
+
+    private boolean mutable = true;
+    private byte[] payload;
+    private int qos = 1;
+    private boolean retained = false;
+    private boolean dup = false;
+    private int messageId;
+    private byte[] correlationData;
+    private List<UserProperty> userProperties;
+
+    public MqttMessage() {
+        setPayload(new byte[]{});
+    }
+
+    public MqttMessage(final byte[] payload) {
+        setPayload(payload);
+    }
+
+    /**
+     * Returns whether the message is mutable.
+     * @return mutable
+     */
+    public boolean getMutable() {
+        return mutable;
+    }
+
+    /**
+     * Sets whether the message is mutable.
+     * @param mutable boolean value
+     */
+    public void setMutable(final boolean mutable) {
+        this.mutable = mutable;
+    }
+
+    /**
+     * Returns the payload as byte array.
+     * @return payload The payload byte array
+     */
+    public byte[] getPayload() {
+        return payload;
+    }
+
+    /**
+     * Sets the payload of the message.
+     * @param payload The payload as byte array
+     */
+    public void setPayload(final byte[] payload) {
+        this.payload = payload;
+    }
+
+    /**
+     * Returns the quality of service level for the message.
+     * @return MQTT quality of service level
+     */
+    public int getQos() {
+        return qos;
+    }
+
+    /**
+     * Sets the quality of service level for the message.
+     * @param qos MQTT quality of service level
+     */
+    public void setQos(final int qos) {
+        this.qos = qos;
+    }
+
+    /**
+     * Returns whether the message is a retained message.
+     * @return boolean value
+     */
+    public boolean isRetained() {
+        return retained;
+    }
+
+    /**
+     * Sets whether the message is retained.
+     * @param retained boolean value
+     */
+    public void setRetained(final boolean retained) {
+        this.retained = retained;
+    }
+
+    /**
+     * Returns whether the message is flagged as duplicate.
+     * @return boolean value
+     */
+    public boolean getDup() {
+        return dup;
+    }
+
+    /**
+     * Flags the message as duplicate.
+     * @param dup boolean value
+     */
+    public void setDup(final boolean dup) {
+        this.dup = dup;
+    }
+
+    /**
+     * Returns the message id.
+     * @return message id
+     */
+    public int getId() {
+        return messageId;
+    }
+
+    /**
+     * Sets the message id.
+     * @param messageId message id
+     */
+    public void setId(final int messageId) {
+        this.messageId = messageId;
+    }
+
+    /**
+     * Sets the MQTT user properties.
+     * @param userProperties MQTT user properties
+     */
+    public void setUserProperties(final List<UserProperty> userProperties) {
+        this.userProperties = userProperties;
+    }
+
+    /**
+     * Returns the MQTT user properties.
+     * @return MQTT user properties
+     */
+    public List<UserProperty> getUserProperties() {
+        return this.userProperties;
+    }
+}

mqtt-hivemq/src/main/java/io/micronaut/mqtt/hivemq/bind/UserProperty.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017-2023 original authors
+ *
+ * Licensed 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
+ *
+ * https://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 io.micronaut.mqtt.hivemq.bind;
+
+/**
+ * Represents a MQTT v5 property.
+ *
+ * @author Sven Kobow
+ * @since 3.0.0
+ */
+public class UserProperty {
+    private final String key;
+    private final String value;
+
+    public UserProperty(String key, String value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    /**
+     * The key of the property.
+     * @return the key
+     */
+    public String getKey() {
+        return key;
+    }
+
+    /**
+     * The value of the property.
+     * @return the value
+     */
+    public String getValue() {
+        return value;
+    }
+}

mqtt-hivemq/src/main/java/io/micronaut/mqtt/hivemq/client/MqttClientAdapter.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2017-2023 original authors
+ *
+ * Licensed 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
+ *
+ * https://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 io.micronaut.mqtt.hivemq.client;
+
+import io.micronaut.mqtt.bind.MqttBindingContext;
+import io.micronaut.mqtt.hivemq.bind.MqttMessage;
+
+import java.util.Set;
+import java.util.function.Consumer;
+
+public interface MqttClientAdapter {
+    void subscribe(String[] topics, int[] qos, Consumer<MqttBindingContext<MqttMessage>> callback);
+
+    void unsubscribe(Set<String> topics);
+
+    boolean isConnected();
+
+    Object getClientIdentifier();
+}

mqtt-hivemq/src/main/java/io/micronaut/mqtt/hivemq/client/MqttClientFactory.java

@@ -15,8 +15,58 @@
  */
 package io.micronaut.mqtt.hivemq.client;
 
-import io.micronaut.context.annotation.Factory;
+import io.micronaut.mqtt.hivemq.ssl.CertificateReader;
+import io.micronaut.mqtt.hivemq.ssl.KeyManagerFactoryCreationException;
+import io.micronaut.mqtt.hivemq.ssl.PrivateKeyReader;
+import io.micronaut.mqtt.hivemq.ssl.TrustManagerFactoryCreationException;
+import io.micronaut.mqtt.ssl.MqttCertificateConfiguration;
 
-@Factory
-public class MqttClientFactory {
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.TrustManagerFactory;
+import java.io.IOException;
+import java.security.*;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateException;
+
+public interface MqttClientFactory {
+
+    default KeyManagerFactory getKeyManagerFactory(final MqttCertificateConfiguration certConfiguration) throws KeyManagerFactoryCreationException {
+        try {
+            final KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+            final KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+
+            final Certificate certificate = CertificateReader.readCertificate(certConfiguration.getCertificate());
+
+            final PrivateKey key = PrivateKeyReader.getPrivateKey(certConfiguration.getPrivateKey(), certConfiguration.getPassword());
+
+            keyStore.load(null, null);
+            keyStore.setCertificateEntry("certificate", certificate);
+            keyStore.setKeyEntry("private-key", key, certConfiguration.getPassword(), new Certificate[]{certificate});
+
+            kmf.init(keyStore, certConfiguration.getPassword());
+
+            return kmf;
+        } catch (NoSuchAlgorithmException | KeyStoreException | IOException | CertificateException |
+                 UnrecoverableKeyException e) {
+            throw new KeyManagerFactoryCreationException(e.getMessage(), e);
+        }
+    }
+
+    default TrustManagerFactory getTrustManagerFactory(final MqttCertificateConfiguration certConfiguration) throws TrustManagerFactoryCreationException {
+        try {
+            final TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            final KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
+
+            final Certificate certificate = CertificateReader.readCertificate(certConfiguration.getCertificateAuthority());
+
+            keyStore.load(null);
+            keyStore.setCertificateEntry("ca-certificate", certificate);
+
+            tmf.init(keyStore);
+
+            return tmf;
+        } catch (NoSuchAlgorithmException | KeyStoreException | IOException | CertificateException e) {
+            throw new TrustManagerFactoryCreationException(e.getMessage(), e);
+        }
+    }
 }

mqtt-hivemq/src/main/java/io/micronaut/mqtt/hivemq/client/health/MqttHealthIndicator.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017-2023 original authors
+ *
+ * Licensed 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
+ *
+ * https://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 io.micronaut.mqtt.hivemq.client.health;
+
+import io.micronaut.context.annotation.Requires;
+import io.micronaut.core.async.publisher.Publishers;
+import io.micronaut.core.util.StringUtils;
+import io.micronaut.health.HealthStatus;
+import io.micronaut.management.endpoint.health.HealthEndpoint;
+import io.micronaut.management.health.indicator.HealthIndicator;
+import io.micronaut.management.health.indicator.HealthResult;
+import io.micronaut.mqtt.hivemq.client.MqttClientAdapter;
+import jakarta.inject.Singleton;
+import org.reactivestreams.Publisher;
+
+import java.util.Collections;
+
+@Requires(property = HealthEndpoint.PREFIX + ".mqtt.client.enabled", value = StringUtils.TRUE)
+@Requires(beans = HealthEndpoint.class)
+@Singleton
+public class MqttHealthIndicator implements HealthIndicator {
+    public static final String NAME = "mqtt-client";
+    private final MqttClientAdapter client;
+
+    public MqttHealthIndicator(final MqttClientAdapter client) {
+        this.client = client;
+    }
+
+    @Override
+    public Publisher<HealthResult> getResult() {
+        HealthStatus status = client.isConnected() ? HealthStatus.UP : HealthStatus.DOWN;
+        HealthResult.Builder builder = HealthResult.builder(NAME, status).details(Collections.singletonMap("clientId", client.getClientIdentifier()));
+        return Publishers.just(builder.build());
+    }
+}