Repository: moquette-io/moquette (java)

Issue:
Server can define its own Keep Alive in CONNACK
Allow the Server to specify the value it wishes the Client to use as a keep alive. This allows the Server to set a maximum allowed keepalive and still have the Client honor it.

## MQTT5 specification
```
If the Server sends a Server Keep Alive on the CONNACK packet, the Client MUST use this value instead 
of the Keep Alive value the Client sent on CONNECT [MQTT-3.2.2-21]. If the Server does not send 
the Server Keep Alive, the Server MUST use the Keep Alive value set by the Client on CONNECT [MQTT-3.2.2-22]
```

## Implementation notes
The server could be configured, through its `config/moquette.conf` file, to use a predefined `keep alive` value (in seconds), to override the one that the clients would transmit. This value is sent back to the the client as `Server Keep Alive` property in CONNACK.

The name of the setting to be used in the config file is `server_keep_alive` and could be expressed as a duration.
Examples:
```
server_keep_alive 10s
```

or 

```
server_keep_alive 2m
```

Pull request:
Implemented server keep alive setting and usage in connect
## Release notes
Added configuration setting so that the broker can specify a keep alive other then the one selected by clients.


## What does this PR do?

Updates the configuration harness to expose a duration setting name `server_keep_alive`.
Updated the connAck properties creation to include `server keep alive` if the broker configuration has selected it.
Updates the connect processing to leverage the `server keep alive` selected keep alive instead of the client one.

## Why is it important/What is the impact to the user?

<!-- Mandatory
Explain here the WHY or the IMPACT to the user, or the rationale/motivation for the changes.

Example:
  This PR fixes an issue that was preventing the docker image from using the proxy setting when sending bla bla information.
  and/or
  This PR now allows the user to define the bla bla setting in the docker container.
-->

## Checklist

<!-- Mandatory
Add a checklist of things that are required to be reviewed in order to have the PR approved

List here all the items you have verified BEFORE sending this PR. Please DO NOT remove any item, striking through those that do not apply. (Just in case, strikethrough uses two tildes. ~~Scratch this.~~)
-->

- [x] My code follows the style guidelines of this project
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] I have made corresponding change to the default configuration files (and/or docker env variables)
- [x] I have added tests that prove my fix is effective or that my feature works
- [x] I have updated the Changelog if it's a feature or a fix that has to be reported

## Author's Checklist

<!-- Recommended
Add a checklist of things that are required to be reviewed in order to have the PR approved
-->
- [ ]

## How to test this PR locally

<!-- Recommended
Explain here how this PR will be tested by the reviewer: commands, dependencies, steps, etc.
-->

## Related issues

<!-- Recommended
Link related issues below. Insert the issue link or reference after the word "Closes" if merging this should automatically close it.

- Closes #123
- Relates #123
- Requires #123
- Superseeds #123
-->
- Closes #789 

## Use cases

<!-- Recommended
Explain here the different behaviors that this PR introduces or modifies in this project, user roles, environment configuration, etc.

If you are familiar with Gherkin test scenarios, we recommend its usage: https://cucumber.io/docs/gherkin/reference/
-->


Patch:
diff --git a/ChangeLog.txt b/ChangeLog.txt
index 6c774163..1c036e81 100644
--- a/ChangeLog.txt
+++ b/ChangeLog.txt
@@ -1,4 +1,5 @@
 Version 0.18-SNAPSHOT:
+   [feature] Server keep alive: added configuration setting so that the broker can specify a keep alive other then the one selected by clients. (#789)
    [feature] User properties: covered publish with user properties with tests and fixed publish of will messages. (#877)
    [feature] Topic alias: implemented handling of topic alias received by publishers. (#873)
    [feature] Flow-control: Handled client's receive maximum property to configure the inflight window through the client. (#858)
diff --git a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
index 421ad5fb..dfbc1436 100644
--- a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
+++ b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
@@ -19,6 +19,7 @@ import io.moquette.BrokerConstants;
 import io.moquette.broker.config.IConfig;
 
 import java.util.Locale;
+import java.util.Optional;
 
 class BrokerConfiguration {
 
@@ -30,6 +31,7 @@ class BrokerConfiguration {
     private final int topicAliasMaximum;
     // integer max value means that the property is unset
     private int receiveMaximum;
+    private Optional<Integer> serverKeepAlive = Optional.empty();
 
     BrokerConfiguration(IConfig props) {
         allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
@@ -70,6 +72,10 @@ class BrokerConfiguration {
         receiveMaximum = props.intProp(IConfig.RECEIVE_MAXIMUM, BrokerConstants.RECEIVE_MAXIMUM);
 
         topicAliasMaximum = props.intProp(IConfig.TOPIC_ALIAS_MAXIMUM_PROPERTY_NAME, BrokerConstants.DISABLED_TOPIC_ALIAS);
+
+        if (props.getProperty(IConfig.SERVER_KEEP_ALIVE_PROPERTY_NAME) != null) {
+            serverKeepAlive = Optional.of((int) props.durationProp(IConfig.SERVER_KEEP_ALIVE_PROPERTY_NAME).toMillis() / 1_000);
+        }
     }
 
     // test method
@@ -133,4 +139,8 @@ class BrokerConfiguration {
     public int topicAliasMaximum() {
         return topicAliasMaximum;
     }
+
+    public Optional<Integer> getServerKeepAlive() {
+        return serverKeepAlive;
+    }
 }
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
index 74085c40..a7930c86 100644
--- a/broker/src/main/java/io/moquette/broker/MQTTConnection.java
+++ b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -358,6 +358,10 @@ final class MQTTConnection {
                 connAckPropertiesBuilder.topicAliasMaximum(topicAliasMaximum);
             }
 
+            if (brokerConfig.getServerKeepAlive().isPresent()) {
+                connAckPropertiesBuilder.serverKeepAlive(brokerConfig.getServerKeepAlive().get());
+            }
+
             final MqttProperties ackProperties = connAckPropertiesBuilder.build();
             connAckBuilder.properties(ackProperties);
         }
@@ -464,6 +468,13 @@ final class MQTTConnection {
 
     private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) {
         int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
+
+        // force server keep alive if configured
+        if (brokerConfig.getServerKeepAlive().isPresent()) {
+            int serverKeepAlive = brokerConfig.getServerKeepAlive().get();
+            LOG.info("Forcing server keep alive ({}) over client selection ({})", serverKeepAlive, keepAlive);
+            keepAlive = serverKeepAlive;
+        }
         NettyUtils.keepAlive(channel, keepAlive);
         NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
         NettyUtils.clientID(channel, clientId);
diff --git a/broker/src/main/java/io/moquette/broker/config/FluentConfig.java b/broker/src/main/java/io/moquette/broker/config/FluentConfig.java
index 93d219fb..46f296e9 100644
--- a/broker/src/main/java/io/moquette/broker/config/FluentConfig.java
+++ b/broker/src/main/java/io/moquette/broker/config/FluentConfig.java
@@ -5,6 +5,7 @@ import io.moquette.broker.Server;
 
 import java.io.IOException;
 import java.nio.file.Path;
+import java.time.Duration;
 import java.util.Locale;
 import java.util.Properties;
 import java.util.function.Consumer;
@@ -30,6 +31,7 @@ import static io.moquette.broker.config.IConfig.PERSISTENT_CLIENT_EXPIRATION_PRO
 import static io.moquette.broker.config.IConfig.PERSISTENT_QUEUE_TYPE_PROPERTY_NAME;
 import static io.moquette.broker.config.IConfig.PORT_PROPERTY_NAME;
 import static io.moquette.broker.config.IConfig.RECEIVE_MAXIMUM;
+import static io.moquette.broker.config.IConfig.SERVER_KEEP_ALIVE_PROPERTY_NAME;
 import static io.moquette.broker.config.IConfig.SESSION_QUEUE_SIZE;
 import static io.moquette.broker.config.IConfig.SSL_PORT_PROPERTY_NAME;
 import static io.moquette.broker.config.IConfig.SSL_PROVIDER;
@@ -215,6 +217,12 @@ public class FluentConfig {
         return this;
     }
 
+    public FluentConfig serverKeepAlive(Duration keepAliveSeconds) {
+        int seconds = (int) keepAliveSeconds.toMillis() / 1_000;
+        configAccumulator.put(SERVER_KEEP_ALIVE_PROPERTY_NAME, seconds + "s");
+        return this;
+    }
+
     public class TLSConfig {
 
         private SSLProvider providerType;
diff --git a/broker/src/main/java/io/moquette/broker/config/IConfig.java b/broker/src/main/java/io/moquette/broker/config/IConfig.java
index ca2ef181..f73fdad3 100644
--- a/broker/src/main/java/io/moquette/broker/config/IConfig.java
+++ b/broker/src/main/java/io/moquette/broker/config/IConfig.java
@@ -68,6 +68,7 @@ public abstract class IConfig {
     public static final String MAX_SERVER_GRANTED_QOS_PROPERTY_NAME = "max_server_granted_qos";
     public static final int DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE = 8092;
     public static final String TOPIC_ALIAS_MAXIMUM_PROPERTY_NAME = "topic_alias_maximum";
+    public static final String SERVER_KEEP_ALIVE_PROPERTY_NAME = "server_keep_alive";
 
     public abstract void setProperty(String name, String value);
 
diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf
index b863cced..b13faad6 100644
--- a/distribution/src/main/resources/moquette.conf
+++ b/distribution/src/main/resources/moquette.conf
@@ -234,3 +234,12 @@ password_file config/password_file.conf
 # default: 0 (disabled)
 #*********************************************************************
 # topic_alias_maximum 16
+
+#*********************************************************************
+# Keep alive provided by server, only for MQTT5.
+#
+# server_keep_alive:
+#       Option used to configure a server preferred keep alive that the client must respect.
+# default: empty (disabled)
+#*********************************************************************
+# server_keep_alive 2s


Test patch:
diff --git a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
index a977e769..906a8397 100644
--- a/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
+++ b/broker/src/test/java/io/moquette/integration/mqtt5/ConnectTest.java
@@ -13,6 +13,8 @@ import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5DisconnectReasonCode
 import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish;
 import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5PublishResult;
 import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5WillPublishBuilder;
+import io.moquette.broker.config.FluentConfig;
+import io.moquette.broker.config.IConfig;
 import io.moquette.testclient.Client;
 import io.netty.handler.codec.mqtt.*;
 import org.awaitility.Awaitility;
@@ -70,6 +72,35 @@ class ConnectTest extends AbstractServerIntegrationTest {
         client.disconnect();
     }
 
+    @Test
+    public void givenServerKeepAliveConfiguredThenConnectAckMustRespectIt() throws IOException {
+        stopServer();
+        IConfig config = new FluentConfig()
+            .dataPath(dbPath)
+            .enablePersistence()
+            .port(1883)
+            .disableTelemetry()
+            .persistentQueueType(FluentConfig.PersistentQueueType.SEGMENTED)
+            .serverKeepAlive(Duration.ofSeconds(12))
+            .build();
+        startServer(config);
+
+        Mqtt5BlockingClient client = MqttClient.builder()
+            .useMqttVersion5()
+            .identifier("simple_connect_test")
+            .serverHost("localhost")
+            .serverPort(1883)
+            .buildBlocking();
+        final Mqtt5ConnAck connectAck = client.connect();
+        assertEquals(Mqtt5ConnAckReasonCode.SUCCESS, connectAck.getReasonCode(), "Accept plain connection");
+        assertTrue(connectAck.getServerKeepAlive().isPresent());
+        connectAck.getServerKeepAlive().ifPresent(serverKeepAlive -> {
+            assertEquals(12 ,serverKeepAlive);
+        });
+
+        client.disconnect();
+    }
+
     @Test
     public void sendConnectOnDisconnectedConnection() throws InterruptedException {
         MqttConnAckMessage connAck = lowLevelClient.connectV5();