Added package 'subscribing'.

This package provides all means to subscribe for topics at the MQTT broker.

Signed-off-by: Juergen Fickel <juergen.fickel@bosch.io>

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/subscribing/AllSubscriptionsFailedException.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.mqtt.hivemq.subscribing;
+
+import java.io.Serial;
+import java.util.List;
+import java.util.Objects;
+
+import javax.annotation.Nullable;
+
+import org.eclipse.ditto.base.model.common.ConditionChecker;
+
+/**
+ * This is exception is thrown to indicate that all subscriptions of an MQTT Subscribe message
+ * ({@link GenericMqttSubscribe}) failed.
+ */
+public final class AllSubscriptionsFailedException extends MqttSubscribeException {
+
+    @Serial private static final long serialVersionUID = 4004096047016339725L;
+
+    private final transient List<SubscriptionStatus> failedSubscriptionStatuses;
+
+    /**
+     * Constructs a new {@code AllSubscriptionsFailedException} object.
+     *
+     * @param failedSubscriptionStatuses a List containing the status of each failed subscription.
+     * @throws NullPointerException if {@code failedSubscriptionStatuses} is {@code null}.
+     * @throws IllegalArgumentException if {@code failedSubscriptionStatuses} is empty.
+     */
+    AllSubscriptionsFailedException(final List<SubscriptionStatus> failedSubscriptionStatuses) {
+        super();
+        ConditionChecker.argumentNotEmpty(failedSubscriptionStatuses, "failedSubscriptionStatuses");
+        this.failedSubscriptionStatuses = List.copyOf(failedSubscriptionStatuses);
+    }
+
+    /**
+     * Returns an unmodifiable unsorted List containing the status of each failed subscription.
+     *
+     * @return the failed subscription statuses.
+     */
+    public List<SubscriptionStatus> getFailedSubscriptionStatuses() {
+        return failedSubscriptionStatuses;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final var that = (AllSubscriptionsFailedException) o;
+        return Objects.equals(failedSubscriptionStatuses, that.failedSubscriptionStatuses);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(failedSubscriptionStatuses);
+    }
+
+}

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/subscribing/GenericMqttSubAckStatus.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.mqtt.hivemq.subscribing;
+
+import static org.eclipse.ditto.base.model.common.ConditionChecker.checkNotNull;
+
+import java.text.MessageFormat;
+import java.util.Objects;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+import com.hivemq.client.mqtt.mqtt3.message.subscribe.suback.Mqtt3SubAckReturnCode;
+import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCode;
+
+/**
+ * Generic representation of an MQTT SubAck message status to abstract HiveMQ API for protocol versions 3 and 5.
+ */
+@Immutable
+public final class GenericMqttSubAckStatus {
+
+    private final int code;
+    private final String name;
+    private final boolean error;
+
+    private GenericMqttSubAckStatus(final int code, final String name, final boolean error) {
+        this.code = code;
+        this.name = name;
+        this.error = error;
+    }
+
+    /**
+     * Returns an instance of {@code GenericMqttSubAckStatus} for the specified {@code Mqtt3SubAckReturnCode} argument.
+     *
+     * @param mqtt3SubAckReturnCode provides the properties of the returned instance.
+     * @return the instance.
+     * @throws NullPointerException if {@code mqtt3SubAckReturnCode} is {@code null}.
+     */
+    static GenericMqttSubAckStatus ofMqtt3SubAckReturnCode(final Mqtt3SubAckReturnCode mqtt3SubAckReturnCode) {
+        checkNotNull(mqtt3SubAckReturnCode, "mqtt3SubAckReturnCode");
+        return new GenericMqttSubAckStatus(mqtt3SubAckReturnCode.getCode(),
+                mqtt3SubAckReturnCode.name(),
+                mqtt3SubAckReturnCode.isError());
+    }
+
+    /**
+     * Returns an instance of {@code GenericMqttSubAckStatus} for the specified {@code Mqtt5SubAckReasonCode} argument.
+     *
+     * @param mqtt5SubAckReasonCode provides the properties of the returned instance.
+     * @return the instance.
+     * @throws NullPointerException if {@code mqtt5SubAckReasonCode} is {@code null}.
+     */
+    static GenericMqttSubAckStatus ofMqtt5SubAckReasonCode(final Mqtt5SubAckReasonCode mqtt5SubAckReasonCode) {
+        checkNotNull(mqtt5SubAckReasonCode, "mqtt5SubAckReasonCode");
+        return new GenericMqttSubAckStatus(mqtt5SubAckReasonCode.getCode(),
+                mqtt5SubAckReasonCode.name(),
+                mqtt5SubAckReasonCode.isError());
+    }
+
+    /**
+     * Returns the code of this MQTT SubAck message status.
+     *
+     * @return the byte code of this MQTT SubAck message status.
+     */
+    public int getCode() {
+        return code;
+    }
+
+    /**
+     * Returns the name of this MQTT SubAck message status.
+     *
+     * @return the name of this MQTT SubAck message status.
+     */
+    public String getName() {
+        return name;
+    }
+
+    /**
+     * Indicates whether this status of a MQTT SubAck message is an error.
+     *
+     * @return {@code true} if this MQTT SubAck message status represents an error, {@code false} else.
+     */
+    public boolean isError() {
+        return error;
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final var that = (GenericMqttSubAckStatus) o;
+        return code == that.code && error == that.error && Objects.equals(name, that.name);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(code, name, error);
+    }
+
+    @Override
+    public String toString() {
+        return MessageFormat.format("{0}: {1}({2,number,integer})",
+                isError() ? "Error" : "Success",
+                getName(),
+                getCode());
+    }
+
+}

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/subscribing/GenericMqttSubscribe.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.mqtt.hivemq.subscribing;
+
+import java.util.LinkedHashSet;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Stream;
+
+import javax.annotation.Nullable;
+import javax.annotation.concurrent.Immutable;
+
+import org.eclipse.ditto.base.model.common.ConditionChecker;
+
+import com.hivemq.client.mqtt.mqtt3.message.subscribe.Mqtt3Subscribe;
+import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscribe;
+
+/**
+ * Generic representation of an MQTT Subscribe message of protocol versions 3 and 5.
+ */
+@Immutable
+final class GenericMqttSubscribe {
+
+    private final LinkedHashSet<GenericMqttSubscription> genericMqttSubscriptions;
+
+    private GenericMqttSubscribe(final Set<GenericMqttSubscription> genericMqttSubscriptions) {
+        this.genericMqttSubscriptions = new LinkedHashSet<>(genericMqttSubscriptions);
+    }
+
+    /**
+     * Returns an instance of {@code GenericMqttSubscribe} for the specified MQTT subscriptions.
+     *
+     * @param genericMqttSubscriptions the subscriptions of the returned Subscribe message. The set must contain at
+     * least one subscription.
+     * @return the instance.
+     * @throws NullPointerException if any argument is {@code null}.
+     * @throws IllegalArgumentException if {@code genericMqttSubscriptions} is empty.
+     */
+    static GenericMqttSubscribe of(final Set<GenericMqttSubscription> genericMqttSubscriptions) {
+        return new GenericMqttSubscribe(ConditionChecker.argumentNotEmpty(genericMqttSubscriptions,
+                "genericMqttSubscriptions"));
+    }
+
+    /**
+     * Returns a stream of the subscriptions of this Subscribe message.
+     *
+     * @return a stream of the subscriptions of this Subscribe message.
+     */
+    Stream<GenericMqttSubscription> genericMqttSubscriptions() {
+        return genericMqttSubscriptions.stream();
+    }
+
+    /**
+     * Returns this Subscribe message as {@link Mqtt3Subscribe}.
+     *
+     * @return this Subscribe message as {@link Mqtt3Subscribe}.
+     */
+    Mqtt3Subscribe getAsMqtt3Subscribe() {
+        return Mqtt3Subscribe.builder()
+                .addSubscriptions(genericMqttSubscriptions().map(GenericMqttSubscription::getAsMqtt3Subscription))
+                .build();
+    }
+
+    /**
+     * Returns this Subscribe message as {@link Mqtt5Subscribe}.
+     *
+     * @return this Subscribe message as {@link Mqtt5Subscribe}.
+     */
+    Mqtt5Subscribe getAsMqtt5Subscribe() {
+        return Mqtt5Subscribe.builder()
+                .addSubscriptions(genericMqttSubscriptions().map(GenericMqttSubscription::getAsMqtt5Subscription))
+                .build();
+    }
+
+    @Override
+    public boolean equals(@Nullable final Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        final var that = (GenericMqttSubscribe) o;
+        return Objects.equals(genericMqttSubscriptions, that.genericMqttSubscriptions);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(genericMqttSubscriptions);
+    }
+
+}

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/mqtt/hivemq/subscribing/GenericMqttSubscribeFactory.java

@@ -0,0 +1,97 @@
+/*
+ * Copyright (c) 2022 Contributors to the Eclipse Foundation
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.eclipse.ditto.connectivity.service.messaging.mqtt.hivemq.subscribing;
+
+import java.text.MessageFormat;
+import java.util.Collection;
+import java.util.LinkedHashSet;
+import java.util.Optional;
+import java.util.Set;
+
+import javax.annotation.concurrent.Immutable;
+
+import org.eclipse.ditto.base.model.common.ConditionChecker;
+import org.eclipse.ditto.connectivity.model.Source;
+
+import com.hivemq.client.mqtt.datatypes.MqttQos;
+import com.hivemq.client.mqtt.datatypes.MqttTopicFilter;
+
+/**
+ * This factory creates an optional {@link GenericMqttSubscribe} for each address of a Connection {@link Source}.
+ * The Optional is present if the Source has at least one address, else the Optional is empty.
+ */
+@Immutable
+final class GenericMqttSubscribeFactory {
+
+    private GenericMqttSubscribeFactory() {
+        throw new AssertionError();
+    }
+
+    /**
+     * Returns the optional {@code GenericMqttSubscribe} for the specified {@code Source} argument.
+     *
+     * @param connectionSource the Connection Source to get the generic MQTT Subscribe message for.
+     * @return an Optional containing the generic MQTT Subscribe message or an empty optional if
+     * {@code connectionSource} did not contain addresses.
+     * @throws NullPointerException if {@code connectionSource} is {@code null}.
+     * @throws InvalidMqttTopicFilterStringException if any address of {@code connectionSource} is not a valid
+     * {@link MqttTopicFilter}.
+     */
+    static Optional<GenericMqttSubscribe> getGenericSourceSubscribeMessage(final Source connectionSource)
+            throws InvalidMqttTopicFilterStringException {
+
+        ConditionChecker.checkNotNull(connectionSource, "connectionSource");
+        final Optional<GenericMqttSubscribe> result;
+        final var connectionSourceAddresses = connectionSource.getAddresses();
+        if (connectionSourceAddresses.isEmpty()) {
+            result = Optional.empty();
+        } else {
+            result = Optional.of(GenericMqttSubscribe.of(tryToGetGenericMqttSubscriptions(connectionSourceAddresses,
+                    getMqttQos(connectionSource))));
+        }
+        return result;
+    }
+
+    private static MqttQos getMqttQos(final Source source) {
+        return source.getQos().map(MqttQos::fromCode).orElse(MqttQos.EXACTLY_ONCE);
+    }
+
+    private static Set<GenericMqttSubscription> tryToGetGenericMqttSubscriptions(
+            final Collection<String> sourceAddresses,
+            final MqttQos mqttQos
+    ) throws InvalidMqttTopicFilterStringException {
+
+        final var result = new LinkedHashSet<GenericMqttSubscription>(sourceAddresses.size());
+        for (final var sourceAddress : sourceAddresses) {
+            result.add(GenericMqttSubscription.newInstance(tryToGetMqttTopicFilter(sourceAddress), mqttQos));
+        }
+        return result;
+    }
+
+    private static MqttTopicFilter tryToGetMqttTopicFilter(final String sourceAddress)
+            throws InvalidMqttTopicFilterStringException {
+
+        try {
+            return MqttTopicFilter.of(sourceAddress);
+        } catch (final IllegalArgumentException e) {
+            throw new InvalidMqttTopicFilterStringException(
+                    MessageFormat.format("Failed to instantiate {0} for <{1}>: {2}",
+                            MqttTopicFilter.class.getSimpleName(),
+                            sourceAddress,
+                            e.getMessage()),
+                    e
+            );
+        }
+    }
+
+}