Initialize ping with connect flush time so the first ping is sent earlier

Author: SgtSilvio

src/main/java/com/hivemq/client/internal/mqtt/handler/connect/MqttConnectHandler.java

@@ -76,6 +76,7 @@ public class MqttConnectHandler extends MqttTimeoutInboundHandler {
     private final @NotNull MqttDisconnectOnConnAckHandler disconnectOnConnAckHandler;
 
     private boolean connectCalled = false;
+    private long connectFlushTime;
 
     @Inject
     MqttConnectHandler(
@@ -122,6 +123,7 @@ public void handlerAdded(final @NotNull ChannelHandlerContext ctx) {
      * @param ctx the channel handler context.
      */
     private void writeConnect(final @NotNull ChannelHandlerContext ctx) {
+        connectFlushTime = System.nanoTime();
         ctx.writeAndFlush((connect.getRawEnhancedAuthMechanism() == null) ?
                 connect.createStateful(clientConfig.getRawClientIdentifier(), null) : connect).addListener(this);
     }
@@ -163,8 +165,7 @@ public void channelRead(final @NotNull ChannelHandlerContext ctx, final @NotNull
      */
     private void readConnAck(final @NotNull MqttConnAck connAck, final @NotNull Channel channel) {
         if (connAck.getReasonCode().isError()) {
-            MqttDisconnectUtil.close(channel, new Mqtt5ConnAckException(
-                    connAck,
+            MqttDisconnectUtil.close(channel, new Mqtt5ConnAckException(connAck,
                     "CONNECT failed as CONNACK contained an Error Code: " + connAck.getReasonCode() + "."));
 
         } else if (validateClientIdentifier(connAck, channel)) {
@@ -178,7 +179,8 @@ private void readConnAck(final @NotNull MqttConnAck connAck, final @NotNull Chan
 
             final int keepAlive = connectionConfig.getKeepAlive();
             if (keepAlive > 0) {
-                channel.pipeline().addAfter(MqttDecoder.NAME, MqttPingHandler.NAME, new MqttPingHandler(keepAlive));
+                final MqttPingHandler pingHandler = new MqttPingHandler(keepAlive, connectFlushTime, System.nanoTime());
+                channel.pipeline().addAfter(MqttDecoder.NAME, MqttPingHandler.NAME, pingHandler);
             }
 
             clientConfig.getRawState().set(MqttClientState.CONNECTED);

src/main/java/com/hivemq/client/internal/mqtt/handler/ping/MqttPingHandler.java

@@ -57,15 +57,16 @@ public class MqttPingHandler extends MqttConnectionAwareHandler
     private boolean messageRead;
     private @Nullable ScheduledFuture<?> timeoutFuture;
 
-    public MqttPingHandler(final int keepAlive) {
+    public MqttPingHandler(final int keepAlive, final long lastFlushTimeNanos, final long lastReadTimeNanos) {
         keepAliveNanos = TimeUnit.SECONDS.toNanos(keepAlive) - TimeUnit.MILLISECONDS.toNanos(100);
+        this.lastFlushTimeNanos = lastFlushTimeNanos;
+        this.lastReadTimeNanos = lastReadTimeNanos;
     }
 
     @Override
     public void handlerAdded(final @NotNull ChannelHandlerContext ctx) {
         super.handlerAdded(ctx);
-        lastFlushTimeNanos = lastReadTimeNanos = System.nanoTime();
-        schedule(ctx, keepAliveNanos);
+        schedule(ctx, nextDelay(System.nanoTime()));
     }
 
     @Override
@@ -89,6 +90,10 @@ private void schedule(final @NotNull ChannelHandlerContext ctx, final long delay
         timeoutFuture = ctx.executor().schedule(this, delayNanos, TimeUnit.NANOSECONDS);
     }
 
+    private long nextDelay(final long timeNanos) {
+        return keepAliveNanos - (timeNanos - Math.min(lastReadTimeNanos, lastFlushTimeNanos));
+    }
+
     @Override
     public void run() {
         if (ctx == null) {
@@ -107,7 +112,7 @@ public void run() {
         pingReqFlushed = false;
         messageRead = false;
         final long timeNanos = System.nanoTime();
-        final long nextDelayNanos = keepAliveNanos - (timeNanos - Math.min(lastReadTimeNanos, lastFlushTimeNanos));
+        final long nextDelayNanos = nextDelay(timeNanos);
         if (nextDelayNanos > 1_000) {
             pingReqWritten = false;
             schedule(ctx, nextDelayNanos);