add java docs, fix publish issue, extend ConnectPublishSubscribeTest

Author: crazyrokr

src/main/java/com/ss/mqtt/broker/config/MqttBrokerConfig.java

@@ -83,8 +83,8 @@ private interface ChannelFactory extends
     }
 
     @Bean
-    @NotNull PublishingService publishingService() {
-        return new SimplePublishingService(subscriptionService());
+    @NotNull PublishingService publishingService(@NotNull SubscriptionService subscriptionService) {
+        return new SimplePublishingService(subscriptionService);
     }
 
     @Bean

src/main/java/com/ss/mqtt/broker/network/MqttClient.java

@@ -115,7 +115,7 @@ private void onUnsubscribe(@NotNull UnsubscribeInPacket subscribe) {
     }
 
     public void onPublish(@NotNull PublishInPacket publish) {
-        var ackReasonCode = connection.getPublishingService().publish(connection.getClient(), publish);
+        var ackReasonCode = connection.getPublishingService().publish(publish);
         connection.send(getPacketOutFactory().newPublishAck(
             connection.getClient(),
             publish.getPacketId(),

src/main/java/com/ss/mqtt/broker/network/packet/out/Publish5OutPacket.java

@@ -173,26 +173,14 @@ protected void writeProperties(@NotNull ByteBuffer buffer) {
 
         // https://docs.oasis-open.org/mqtt/mqtt/v5.0/os/mqtt-v5.0-os.html#_Toc511988586
         writeProperty(buffer, PacketProperty.PAYLOAD_FORMAT_INDICATOR, stringPayload);
-        writeProperty(
-            buffer,
+        writeProperty(buffer,
             PacketProperty.MESSAGE_EXPIRY_INTERVAL,
             0,
             MqttPropertyConstants.MESSAGE_EXPIRY_INTERVAL_DEFAULT
         );
-        if (topciAlias > MqttPropertyConstants.TOPIC_ALIAS_MIN && topciAlias < MqttPropertyConstants.TOPIC_ALIAS_MAX) {
-            writeProperty(
-                buffer,
-                PacketProperty.TOPIC_ALIAS,
-                topciAlias,
-                MqttPropertyConstants.TOPIC_ALIAS_MAXIMUM_DEFAULT
-            );
-        }
-        if (responseTopic != null) {
-            writeProperty(buffer, PacketProperty.RESPONSE_TOPIC, responseTopic);
-        }
-        if (correlationData != null) {
-            writeProperty(buffer, PacketProperty.CORRELATION_DATA, correlationData);
-        }
+        writeProperty(buffer, PacketProperty.TOPIC_ALIAS, topciAlias, MqttPropertyConstants.TOPIC_ALIAS_DEFAULT);
+        writeNotEmptyProperty(buffer, PacketProperty.RESPONSE_TOPIC, responseTopic);
+        writeNotEmptyProperty(buffer, PacketProperty.CORRELATION_DATA, correlationData);
         writeStringPairProperties(buffer, PacketProperty.USER_PROPERTY, userProperties);
     }
 

src/main/java/com/ss/mqtt/broker/service/PublishingService.java

@@ -6,8 +6,17 @@
 import com.ss.rlib.common.util.array.Array;
 import org.jetbrains.annotations.NotNull;
 
+/**
+ * Publishing service
+ */
 public interface PublishingService {
 
-    @NotNull PublishAckReasonCode publish(@NotNull MqttClient mqttClient, @NotNull PublishInPacket publish);
+    /**
+     * Sends publish packet to all subscribers
+     *
+     * @param publish publish packet to send
+     * @return publish ack reason code
+     */
+    @NotNull PublishAckReasonCode publish(@NotNull PublishInPacket publish);
 
 }

src/main/java/com/ss/mqtt/broker/service/Subscriber.java

@@ -4,20 +4,26 @@
 import com.ss.mqtt.broker.model.SubscribeRetainHandling;
 import com.ss.mqtt.broker.model.SubscribeTopicFilter;
 import com.ss.mqtt.broker.network.MqttClient;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import org.jetbrains.annotations.NotNull;
 
-import java.util.Objects;
-
 @Getter
+@EqualsAndHashCode(of = "mqttClient")
 public class Subscriber {
 
-    private final MqttClient mqttClient;
-    private final QoS qos;
-    private final SubscribeRetainHandling retainHandling;
+    private final @NotNull MqttClient mqttClient;
+    private final @NotNull QoS qos;
+    private final @NotNull SubscribeRetainHandling retainHandling;
     private final boolean noLocal;
     private final boolean retainAsPublished;
 
+    /**
+     * Creates subscriber
+     *
+     * @param mqttClient  MQTT client which will become a subscriber
+     * @param topicFilter topic filter that MQTT client subscribes to
+     */
     public Subscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter topicFilter) {
         this.mqttClient = mqttClient;
         this.qos = topicFilter.getQos();
@@ -26,16 +32,4 @@ public Subscriber(@NotNull MqttClient mqttClient, @NotNull SubscribeTopicFilter
         this.noLocal = topicFilter.isNoLocal();
     }
 
-    @Override
-    public boolean equals(Object o) {
-        if (this == o) return true;
-        if (o == null || getClass() != o.getClass()) return false;
-        var that = (Subscriber) o;
-        return Objects.equals(mqttClient, that.mqttClient);
-    }
-
-    @Override
-    public int hashCode() {
-        return Objects.hash(mqttClient);
-    }
 }

src/main/java/com/ss/mqtt/broker/service/SubscriptionService.java

@@ -9,17 +9,40 @@
 
 import java.util.List;
 
+/**
+ * Subscription service
+ */
 public interface SubscriptionService {
 
+    /**
+     * Adds MQTT client to topic filter subscribers
+     *
+     * @param mqttClient MQTT client to be added
+     * @param topicFilters topic filters
+     * @return array of subscribe ack reason codes
+     */
     @NotNull Array<SubscribeAckReasonCode> subscribe(
         @NotNull MqttClient mqttClient,
-        @NotNull Array<SubscribeTopicFilter> topicFilter
+        @NotNull Array<SubscribeTopicFilter> topicFilters
     );
 
+    /**
+     * Removes MQTT client from subscribers by array of topic names
+     *
+     * @param mqttClient MQTT client to be removed
+     * @param topicNames topic names
+     * @return array of unsubscribe ack reason codes
+     */
     @NotNull Array<UnsubscribeAckReasonCode> unsubscribe(
         @NotNull MqttClient mqttClient,
-        @NotNull Array<String> topicFilter
+        @NotNull Array<String> topicNames
     );
 
-    @NotNull Array<MqttClient> getSubscribers(@NotNull String topic);
+    /**
+     * Returns subscribers by topic name
+     *
+     * @param topicName topic name
+     * @return array of topic subscribers
+     */
+    @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName);
 }

src/main/java/com/ss/mqtt/broker/service/Subscriptions.java

@@ -7,26 +7,40 @@
 import com.ss.rlib.common.util.array.Array;
 import org.jetbrains.annotations.NotNull;
 
+/**
+ * Container of subscriptions
+ */
 public interface Subscriptions {
 
     /**
-     * Return full subscribers list
+     * Returns array of subscribers by topic name
+     *
+     * @param topicName topic name on which subscribers should be returned
+     * @return array of MQTT clients
      */
-    @NotNull Array<MqttClient> getSubscribers(@NotNull String topic);
+    @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName);
 
     /**
-     * Return true if subscription is added
+     * Returns result of subscription adding
+     *
+     * @param topicFilter topic filter that MQTT client wants to subscribe to
+     * @param mqttClient MQTT client which wants to subscribe
+     * @return subscribe ack reason code
      */
     @NotNull SubscribeAckReasonCode addSubscription(
         @NotNull SubscribeTopicFilter topicFilter,
         @NotNull MqttClient mqttClient
     );
 
     /**
-     * Return true if subscription is removed
+     * Returns result of subscription removing
+     *
+     * @param topicName topic name that MQTT client wants to unsubscribe from
+     * @param mqttClient MQTT client which wants to unsubscribe
+     * @return unsubscribe ack reason code
      */
     @NotNull UnsubscribeAckReasonCode removeSubscription(
-        @NotNull String topicFilter,
+        @NotNull String topicName,
         @NotNull MqttClient mqttClient
     );
 }

src/main/java/com/ss/mqtt/broker/service/impl/SimplePublishingService.java

@@ -5,11 +5,12 @@
 import com.ss.mqtt.broker.network.packet.in.PublishInPacket;
 import com.ss.mqtt.broker.service.PublishingService;
 import com.ss.mqtt.broker.service.SubscriptionService;
-import com.ss.rlib.common.util.ArrayUtils;
-import com.ss.rlib.common.util.StringUtils;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 
+/**
+ * Simple publishing service
+ */
 @RequiredArgsConstructor
 public class SimplePublishingService implements PublishingService {
 
@@ -31,20 +32,23 @@ public class SimplePublishingService implements PublishingService {
             publish.getTopicAlias(),
             publish.getPayload(),
             publish.isPayloadFormatIndicator(),
-            StringUtils.EMPTY, //publish.getResponseTopic(),
-            ArrayUtils.EMPTY_BYTE_ARRAY,
+            publish.getResponseTopic(),
+            publish.getCorrelationData(),
             publish.getUserProperties()
         ));
+        // TODO this reason code only for QoS 0
         return PublishAckReasonCode.SUCCESS;
     }
 
     @Override
-    public @NotNull PublishAckReasonCode publish(@NotNull MqttClient mqttClient, @NotNull PublishInPacket publish) {
+    public @NotNull PublishAckReasonCode publish(@NotNull PublishInPacket publish) {
         var subscribers = subscriptionService.getSubscribers(publish.getTopicName());
-        // TODO choose correct PublishAckReasonCode
-        return subscribers.stream()
+        if (subscribers.isEmpty()) {
+            return PublishAckReasonCode.NO_MATCHING_SUBSCRIBERS;
+        }
+        var success = subscribers.stream()
             .map(targetMqttClient -> send(targetMqttClient, publish))
-            .findFirst()
-            .orElse(PublishAckReasonCode.NO_MATCHING_SUBSCRIBERS);
+            .allMatch(ackReasonCode -> ackReasonCode.equals(PublishAckReasonCode.SUCCESS));
+        return success ? PublishAckReasonCode.SUCCESS : PublishAckReasonCode.UNSPECIFIED_ERROR;
     }
 }

src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptionService.java

@@ -14,33 +14,36 @@
 
 import java.util.List;
 
+/**
+ * Simple subscription service
+ */
 @RequiredArgsConstructor
 public class SimpleSubscriptionService implements SubscriptionService {
 
-    private final Subscriptions subscriptions;
+    private final @NotNull Subscriptions subscriptions;
 
     @Override
-    public Array<SubscribeAckReasonCode> subscribe(
+    public @NotNull Array<SubscribeAckReasonCode> subscribe(
         @NotNull MqttClient mqttClient,
-        @NotNull Array<SubscribeTopicFilter> topicFilter
+        @NotNull Array<SubscribeTopicFilter> topicFilters
     ) {
-        return topicFilter.stream()
+        return topicFilters.stream()
             .map(subscribeTopicFilter -> subscriptions.addSubscription(subscribeTopicFilter, mqttClient))
             .collect(ArrayCollectors.toArray(SubscribeAckReasonCode.class));
     }
 
     @Override
-    public Array<UnsubscribeAckReasonCode> unsubscribe(
+    public @NotNull Array<UnsubscribeAckReasonCode> unsubscribe(
         @NotNull MqttClient mqttClient,
-        @NotNull Array<String> topicFilter
+        @NotNull Array<String> topicNames
     ) {
-        return topicFilter.stream()
+        return topicNames.stream()
             .map(subscribeTopicFilter -> subscriptions.removeSubscription(subscribeTopicFilter, mqttClient))
             .collect(ArrayCollectors.toArray(UnsubscribeAckReasonCode.class));
     }
 
     @Override
-    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topic) {
-        return subscriptions.getSubscribers(topic);
+    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName) {
+        return subscriptions.getSubscribers(topicName);
     }
 }

src/main/java/com/ss/mqtt/broker/service/impl/SimpleSubscriptions.java

@@ -14,23 +14,20 @@
 import java.util.HashMap;
 import java.util.Map;
 
+/**
+ * Simple container of subscriptions
+ */
 public class SimpleSubscriptions implements Subscriptions {
 
-    private final Map<String, Array<Subscriber>> subscriptions = new HashMap<>();
+    private final @NotNull Map<String, Array<Subscriber>> subscriptions = new HashMap<>();
 
-    /**
-     * Return full subscribers list
-     */
-    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topic) {
-        return subscriptions.get(topic)
+    public @NotNull Array<MqttClient> getSubscribers(@NotNull String topicName) {
+        return subscriptions.get(topicName)
             .stream()
             .map(Subscriber::getMqttClient)
             .collect(ArrayCollectors.toArray(MqttClient.class));
     }
 
-    /**
-     * Return true if subscription is added
-     */
     public @NotNull SubscribeAckReasonCode addSubscription(
         @NotNull SubscribeTopicFilter topicFilter,
         @NotNull MqttClient mqttClient
@@ -44,11 +41,11 @@ public class SimpleSubscriptions implements Subscriptions {
         return topicFilter.getQos().getSubscribeAckReasonCode();
     }
 
-    /**
-     * Return true if subscription is removed
-     */
-    public @NotNull UnsubscribeAckReasonCode removeSubscription(@NotNull String topicFilter, @NotNull MqttClient mqttClient) {
-        var subscribers = subscriptions.getOrDefault(topicFilter, Array.empty());
+    public @NotNull UnsubscribeAckReasonCode removeSubscription(
+        @NotNull String topicName,
+        @NotNull MqttClient mqttClient
+    ) {
+        var subscribers = subscriptions.getOrDefault(topicName, Array.empty());
         if (subscribers.removeIf(subscriber -> mqttClient.equals(subscriber.getMqttClient()))) {
             return UnsubscribeAckReasonCode.SUCCESS;
         }