continue updating subscribe mechanism

Author: JavaSaBr

application/src/test/groovy/javasabr/mqtt/broker/application/ConnectSubscribePublishTest.groovy

@@ -12,7 +12,6 @@ import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish
 import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAckReasonCode
 
 import java.util.concurrent.CompletableFuture
-import java.util.concurrent.atomic.AtomicReference
 
 class ConnectSubscribePublishTest extends IntegrationSpecification {
 

gradle/libs.versions.toml

@@ -1,6 +1,6 @@
 [versions]
 # https://gitlab.com/JavaSaBr/maven-repo/-/packages
-rlib = "10.0.alpha5"
+rlib = "10.0.alpha6"
 # https://mvnrepository.com/artifact/org.projectlombok/lombok
 lombok = "1.18.38"
 # https://mvnrepository.com/artifact/org.jspecify/jspecify

model/src/main/java/javasabr/mqtt/model/reason/code/DisconnectReasonCode.java

@@ -1,162 +1,152 @@
 package javasabr.mqtt.model.reason.code;
 
-import java.util.stream.Stream;
-import javasabr.rlib.common.util.ObjectUtils;
+import javasabr.rlib.common.util.NumberedEnum;
+import javasabr.rlib.common.util.NumberedEnumMap;
+import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
+import lombok.experimental.FieldDefaults;
 
+@Getter
 @RequiredArgsConstructor
-public enum DisconnectReasonCode {
+@Accessors(fluent = true, chain = false)
+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public enum DisconnectReasonCode implements NumberedEnum<DisconnectReasonCode> {
   /**
    * Close the connection normally. Do not send the Will Message. Client or Server.
    */
-  NORMAL_DISCONNECTION((byte) 0x00),
+  NORMAL_DISCONNECTION(0x00),
   /**
    * The Client wishes to disconnect but requires that the Server also publishes its Will Message. Client.
    */
-  DISCONNECT_WITH_WILL_MESSAGE((byte) 0x04),
+  DISCONNECT_WITH_WILL_MESSAGE(0x04),
 
   // ERRORS
 
   /**
    * The Connection is closed but the sender either does not wish to reveal the reason, or none of the other Reason
    * Codes apply. Client or Server.
    */
-  UNSPECIFIED_ERROR((byte) 0x80),
+  UNSPECIFIED_ERROR(0x80),
   /**
    * The received packet does not conform to this specification. Client or Server.
    */
-  MALFORMED_PACKET((byte) 0x81),
+  MALFORMED_PACKET(0x81),
   /**
    * An unexpected or out of order packet was received. Client or Server.
    */
-  PROTOCOL_ERROR((byte) 0x82),
+  PROTOCOL_ERROR(0x82),
   /**
    * The packet received is valid but cannot be processed by this implementation. Client or Server.
    */
-  IMPLEMENTATION_SPECIFIC_ERROR((byte) 0x83),
+  IMPLEMENTATION_SPECIFIC_ERROR(0x83),
   /**
    * The request is not authorized. Server.
    */
-  NOT_AUTHORIZED((byte) 0x87),
+  NOT_AUTHORIZED(0x87),
   /**
    * The Server is busy and cannot continue processing requests from this Client. Server.
    */
-  SERVER_BUSY((byte) 0x89),
+  SERVER_BUSY(0x89),
   /**
    * The Server is shutting down. Server.
    */
-  SERVER_SHUTTING_DOWN((byte) 0x8B),
+  SERVER_SHUTTING_DOWN(0x8B),
   /**
    * The Connection is closed because no packet has been received for 1.5 times the Keepalive time. Server.
    */
-  KEEP_ALIVE_TIMEOUT((byte) 0x8D),
+  KEEP_ALIVE_TIMEOUT(0x8D),
   /**
    * Another Connection using the same ClientID has connected causing this Connection to be closed. Server.
    */
-  SESSION_TAKEN_OVER((byte) 0x8E),
+  SESSION_TAKEN_OVER(0x8E),
   /**
    * The Topic Filter is correctly formed, but is not accepted by this Sever. Server.
    */
-  TOPIC_FILTER_INVALID((byte) 0x8F),
+  TOPIC_FILTER_INVALID(0x8F),
   /**
    * The Topic Name is correctly formed, but is not accepted by this Client or Server. Client or Server.
    */
-  TOPIC_NAME_INVALID((byte) 0x90),
+  TOPIC_NAME_INVALID(0x90),
   /**
    * The Client or Server has received more than Receive Maximum publication for which it has not sent PUBACK or
    * PUBCOMP. Client or Server.
    */
-  RECEIVE_MAXIMUM_EXCEEDED((byte) 0x93),
+  RECEIVE_MAXIMUM_EXCEEDED(0x93),
   /**
    * The Client or Server has received a PUBLISH packet containing a Topic Alias which is greater than the Maximum Topic
    * Alias it sent in the CONNECT or CONNACK packet. Client or Server.
    */
-  TOPIC_ALIAS_INVALID((byte) 0x94),
+  TOPIC_ALIAS_INVALID(0x94),
   /**
    * The packet size is greater than Maximum Packet Size for this Client or Server. Client or Server.
    */
-  PACKET_TOO_LARGE((byte) 0x95),
+  PACKET_TOO_LARGE(0x95),
   /**
    * The received data rate is too high. Client or Server.
    */
-  MESSAGE_RATE_TOO_HIGH((byte) 0x96),
+  MESSAGE_RATE_TOO_HIGH(0x96),
   /**
    * An implementation or administrative imposed limit has been exceeded. Client or Server.
    */
-  QUOTA_EXCEEDED((byte) 0x97),
+  QUOTA_EXCEEDED(0x97),
   /**
    * The Connection is closed due to an administrative action. Client or Server.
    */
-  ADMINISTRATIVE_ACTION((byte) 0x98),
+  ADMINISTRATIVE_ACTION(0x98),
   /**
    * The payload format does not match the one specified by the Payload Format Indicator. Client or Server.
    */
-  PAYLOAD_FORMAT_INVALID((byte) 0x99),
+  PAYLOAD_FORMAT_INVALID(0x99),
   /**
    * The Server has does not support retained messages. Server.
    */
-  RETAIN_NOT_SUPPORTED((byte) 0x9A),
+  RETAIN_NOT_SUPPORTED(0x9A),
   /**
    * The Client specified a QoS greater than the QoS specified in a Maximum QoS in the CONNACK. Server.
    */
-  QOS_NOT_SUPPORTED((byte) 0x9B),
+  QOS_NOT_SUPPORTED(0x9B),
   /**
    * The Client should temporarily change its Server. Server.
    */
-  USE_ANOTHER_SERVER((byte) 0x9C),
+  USE_ANOTHER_SERVER(0x9C),
   /**
    * The Server is moved and the Client should permanently change its server location. Server.
    */
-  SERVER_MOVED((byte) 0x9D),
+  SERVER_MOVED(0x9D),
   /**
    * The Server does not support Shared Subscriptions. Server.
    */
-  SHARED_SUBSCRIPTIONS_NOT_SUPPORTED((byte) 0x9E),
+  SHARED_SUBSCRIPTIONS_NOT_SUPPORTED(0x9E),
   /**
    * This connection is closed because the connection rate is too high. Server.
    */
-  CONNECTION_RATE_EXCEEDED((byte) 0x9F),
+  CONNECTION_RATE_EXCEEDED(0x9F),
   /**
    * The maximum connection time authorized for this connection has been exceeded. Server.
    */
-  MAXIMUM_CONNECT_TIME((byte) 0xA0),
+  MAXIMUM_CONNECT_TIME(0xA0),
   /**
    * The Server does not support Subscription Identifiers; the subscription is not accepted. Server.
    */
-  SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED((byte) 0xA1),
+  SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED(0xA1),
   /**
    * The Server does not support Wildcard Subscriptions; the subscription is not accepted. Server.
    */
-  WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED((byte) 0xA2);
+  WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED(0xA2);
 
-  private static final DisconnectReasonCode[] VALUES;
+  private static final NumberedEnumMap<DisconnectReasonCode> NUMBERED_MAP =
+      new NumberedEnumMap<>(DisconnectReasonCode.class);
 
-  static {
-
-    var maxId = Stream
-        .of(values())
-        .mapToInt(DisconnectReasonCode::getValue)
-        .map(value -> Byte.toUnsignedInt((byte) value))
-        .max()
-        .orElse(0);
-
-    var values = new DisconnectReasonCode[maxId + 1];
-
-    for (var value : values()) {
-      values[Byte.toUnsignedInt(value.value)] = value;
-    }
-
-    VALUES = values;
+  public static DisconnectReasonCode ofCode(int code) {
+    return NUMBERED_MAP.require(code);
   }
 
-  public static DisconnectReasonCode of(int index) {
-    return ObjectUtils.notNull(
-        VALUES[index],
-        index,
-        arg -> new IndexOutOfBoundsException("Doesn't support reason code: " + arg));
-  }
+  int code;
 
-  private @Getter
-  final byte value;
+  @Override
+  public int number() {
+    return code;
+  }
 }

model/src/main/java/javasabr/mqtt/model/reason/code/SubscribeAckReasonCode.java

@@ -1,92 +1,82 @@
 package javasabr.mqtt.model.reason.code;
 
-import java.util.stream.Stream;
-import javasabr.rlib.common.util.ObjectUtils;
+import javasabr.rlib.common.util.NumberedEnum;
+import javasabr.rlib.common.util.NumberedEnumMap;
+import lombok.AccessLevel;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
+import lombok.experimental.FieldDefaults;
 
+@Getter
 @RequiredArgsConstructor
-public enum SubscribeAckReasonCode {
+@Accessors(fluent = true, chain = false)
+@FieldDefaults(level = AccessLevel.PRIVATE, makeFinal = true)
+public enum SubscribeAckReasonCode implements NumberedEnum<SubscribeAckReasonCode> {
   /**
    * The subscription is accepted and the maximum QoS sent will be QoS 0. This might be a lower QoS than was requested.
    */
-  GRANTED_QOS_0((byte) 0x00),
+  GRANTED_QOS_0(0x00),
   /**
    * The subscription is accepted and the maximum QoS sent will be QoS 1. This might be a lower QoS than was requested.
    */
-  GRANTED_QOS_1((byte) 0x01),
+  GRANTED_QOS_1(0x01),
   /**
    * The subscription is accepted and any received QoS will be sent to this subscription.
    */
-  GRANTED_QOS_2((byte) 0x02),
+  GRANTED_QOS_2(0x02),
 
   // ERRORS
 
   /**
    * The subscription is not accepted and the Server either does not wish to reveal the reason or none of the other
    * Reason Codes apply.
    */
-  UNSPECIFIED_ERROR((byte) 0x80),
+  UNSPECIFIED_ERROR(0x80),
   /**
    * The SUBSCRIBE is valid but the Server does not accept it.
    */
-  IMPLEMENTATION_SPECIFIC_ERROR((byte) 0x83),
+  IMPLEMENTATION_SPECIFIC_ERROR(0x83),
   /**
    * The Client is not authorized to make this subscription.
    */
-  NOT_AUTHORIZED((byte) 0x87),
+  NOT_AUTHORIZED(0x87),
   /**
    * The Topic Filter is correctly formed but is not allowed for this Client.
    */
-  TOPIC_FILTER_INVALID((byte) 0x8F),
+  TOPIC_FILTER_INVALID(0x8F),
   /**
    * The specified Packet Identifier is already in use.
    */
-  PACKET_IDENTIFIER_IN_USE((byte) 0x91),
+  PACKET_IDENTIFIER_IN_USE(0x91),
   /**
    * An implementation or administrative imposed limit has been exceeded.
    */
-  QUOTA_EXCEEDED((byte) 0x97),
+  QUOTA_EXCEEDED(0x97),
   /**
    * The Server does not support Shared Subscriptions for this Client.
    */
-  SHARED_SUBSCRIPTIONS_NOT_SUPPORTED((byte) 0x9E),
+  SHARED_SUBSCRIPTIONS_NOT_SUPPORTED(0x9E),
   /**
    * The Server does not support Subscription Identifiers; the subscription is not accepted.
    */
-  SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED((byte) 0xA1),
+  SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED(0xA1),
   /**
    * The Server does not support Wildcard Subscriptions; the subscription is not accepted.
    */
-  WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED((byte) 0xA2);
+  WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED(0xA2);
 
-  private static final SubscribeAckReasonCode[] VALUES;
+  private static final NumberedEnumMap<SubscribeAckReasonCode> NUMBERED_MAP =
+      new NumberedEnumMap<>(SubscribeAckReasonCode.class);
 
-  static {
-
-    var maxId = Stream
-        .of(values())
-        .mapToInt(SubscribeAckReasonCode::getValue)
-        .map(value -> Byte.toUnsignedInt((byte) value))
-        .max()
-        .orElse(0);
-
-    var values = new SubscribeAckReasonCode[maxId + 1];
-
-    for (var value : values()) {
-      values[Byte.toUnsignedInt(value.value)] = value;
-    }
-
-    VALUES = values;
+  public static SubscribeAckReasonCode ofCode(int code) {
+    return NUMBERED_MAP.require(code);
   }
 
-  public static SubscribeAckReasonCode of(int index) {
-    return ObjectUtils.notNull(
-        VALUES[index],
-        index,
-        arg -> new IndexOutOfBoundsException("Doesn't support reason code: " + arg));
-  }
+  int code;
 
-  @Getter
-  private final byte value;
+  @Override
+  public int number() {
+    return code;
+  }
 }

model/src/main/java/javasabr/mqtt/model/subscribtion/RequestedSubscription.java

@@ -2,7 +2,6 @@
 
 import javasabr.mqtt.model.QoS;
 import javasabr.mqtt.model.SubscribeRetainHandling;
-import javasabr.mqtt.model.topic.TopicFilter;
 
 public record RequestedSubscription(
     /*

network/src/main/java/javasabr/mqtt/network/MqttClient.java

@@ -30,7 +30,9 @@ interface UnsafeMqttClient extends MqttClient {
 
   MqttClientConnectionConfig connectionConfig();
 
-  void send(MqttOutMessage packet);
+  void send(MqttOutMessage message);
 
-  CompletableFuture<Boolean> sendWithFeedback(MqttOutMessage packet);
+  CompletableFuture<Boolean> sendWithFeedback(MqttOutMessage message);
+
+  void closeWithReason(MqttOutMessage message);
 }

network/src/main/java/javasabr/mqtt/network/impl/AbstractMqttClient.java

@@ -49,15 +49,21 @@ public AbstractMqttClient(MqttConnection connection, MqttClientReleaseHandler re
   }
 
   @Override
-  public void send(MqttOutMessage packet) {
-    log.debug(clientId, packet.name(), packet, "[%s] Send to client packet:[%s] %s"::formatted);
-    connection.send(packet);
+  public void send(MqttOutMessage message) {
+    log.debug(clientId, message.name(), message, "[%s] Send to client packet:[%s] %s"::formatted);
+    connection.send(message);
   }
 
   @Override
-  public CompletableFuture<Boolean> sendWithFeedback(MqttOutMessage packet) {
-    log.debug(clientId, packet.name(), packet, "[%s] Send to client packet:[%s] %s"::formatted);
-    return connection.sendWithFeedback(packet);
+  public CompletableFuture<Boolean> sendWithFeedback(MqttOutMessage message) {
+    log.debug(clientId, message.name(), message, "[%s] Send to client packet:[%s] %s"::formatted);
+    return connection.sendWithFeedback(message);
+  }
+
+  @Override
+  public void closeWithReason(MqttOutMessage message) {
+    sendWithFeedback(message)
+        .thenAccept(_ -> connection.close());
   }
 
   @Override