Pause/Resume Reading from Socket

Motivation:
The ability to pause reading from the socket to avoid being overwhelmed by QoS 0 messages during traffic spikes.
When reading is paused, incoming data accumulates in the TCP receive buffer. As the buffer fills up, the TCP stack advertises a reduced window size (eventually a zero window) to the broker, applying backpressure at the TCP level.
At that point, outgoing data may accumulate in the broker’s socket send buffer until its limits are reached. Depending on the broker’s implementation and configuration, QoS 0 messages that cannot be written may be buffered or dropped.

Changes:
Implement new pause/resume methods that change the setAutoRead(bool) connection's option.

Author: dometec

src/main/java/io/vertx/mqtt/MqttClient.java

@@ -321,6 +321,29 @@ static MqttClient create(Vertx vertx) {
   @Fluent
   MqttClient ping();
 
+  /**
+   * Pause the reading channel, so no new byte are read from the server.
+   * Available onliy after connection is established.
+   * <p>
+   * This simply delegates to {@link io.vertx.core.net.NetSocket#pause()}.
+   */
+  void pause();
+
+  /**
+   * Returns true if the reading channel is paused.
+   * Available onliy after connection is established.
+   * @return true if the client publishing is paused.
+   */
+  boolean isPaused();
+
+  /**
+   * Resume the reading channel. see {@link #pause()}
+   * Available onliy after connection is established.
+   * <p>
+   * This simply delegates to {@link io.vertx.core.net.NetSocket#resume()}.
+   */
+  void resume();
+
   /**
    * @return the client identifier
    */

src/main/java/io/vertx/mqtt/impl/MqttClientImpl.java

@@ -18,6 +18,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelConfig;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.DecoderResult;
@@ -76,6 +77,7 @@ private enum Status { CLOSED, CONNECTING, CONNECTED, CLOSING }
   private final VertxInternal vertx;
   private final MqttClientOptions options;
   private NetSocketInternal connection;
+  private ChannelConfig connOption;
   private ContextInternal ctx;
 
   // handler to call when a publish is complete
@@ -245,13 +247,15 @@ private Future<MqttConnAckMessage> doConnect(int port, String host, String serve
           initChannel(soi);
           synchronized (MqttClientImpl.this) {
             this.connection = soi;
+            this.connOption = soi.channelHandlerContext().channel().config();
           }
 
           soi.messageHandler(msg -> this.handleMessage(soi.channelHandlerContext(), msg));
           soi.closeHandler(v2 -> {
             client.close();
             synchronized (MqttClientImpl.this) {
               this.connection = null;
+              this.connOption = null;
               this.status = Status.CLOSED;
               this.connectPromise = null;
               this.disconnectPromise = null;
@@ -721,6 +725,20 @@ public MqttClient ping() {
     return this;
   }
 
+  @Override
+  public synchronized void pause() {
+    connOption.setAutoRead(false);
+  }
+
+  public synchronized boolean isPaused() {
+    return connOption.isAutoRead();
+  }
+
+  @Override
+  public synchronized void resume() {
+    connOption.setAutoRead(true);
+  }
+
   @Override
   public synchronized String clientId() {
     return this.options.getClientId();
@@ -881,6 +899,7 @@ private void handleClosed() {
       this.disconnectPromise = null;
       this.status = Status.CLOSED;
       this.connection = null;
+      this.connOption = null;
       this.ctx = null;
       this.client = null;
       this.pings = new ArrayDeque<>();
@@ -1274,6 +1293,7 @@ private void handleConnack(MqttConnAckMessage msg) {
         this.disconnectPromise = null;
         this.status = Status.CLOSED;
         this.connection = null;
+        this.connOption = null;
         this.client = null;
       }
       connection.closeHandler(null);

src/test/java/io/vertx/mqtt/test/client/MqttClientPauseTest.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2016 Red Hat Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.vertx.mqtt.test.client;
+
+import io.netty.handler.codec.mqtt.MqttQoS;
+import io.vertx.core.Vertx;
+import io.vertx.core.buffer.Buffer;
+import io.vertx.ext.unit.Async;
+import io.vertx.ext.unit.TestContext;
+import io.vertx.ext.unit.junit.VertxUnitRunner;
+import io.vertx.mqtt.MqttClient;
+import io.vertx.mqtt.MqttClientOptions;
+import io.vertx.mqtt.MqttServer;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * MQTT client pause tests.
+ */
+@RunWith(VertxUnitRunner.class)
+public class MqttClientPauseTest {
+
+  private Vertx vertx;
+  private MqttServer server;
+
+  @Before
+  public void before(TestContext ctx) {
+    vertx = Vertx.vertx();
+    server = MqttServer.create(vertx);
+    server.endpointHandler(endpoint -> {
+      endpoint.accept(false);
+      endpoint.publishHandler(msg -> {
+        // Echo back
+        endpoint.publish(msg.topicName(), msg.payload(), MqttQoS.AT_MOST_ONCE, false, false);
+      });
+    });
+    Async async = ctx.async();
+    server.listen(ctx.asyncAssertSuccess(server -> async.complete()));
+    async.awaitSuccess(10000);
+  }
+
+  @After
+  public void after(TestContext ctx) {
+    server.close(ctx.asyncAssertSuccess(v -> {
+      vertx.close(ctx.asyncAssertSuccess());
+    }));
+  }
+
+  @Test
+  public void testPauseResume(TestContext ctx) {
+    Async async = ctx.async();
+    AtomicInteger received = new AtomicInteger();
+    MqttClient client = MqttClient.create(vertx);
+
+    client.connect(MqttClientOptions.DEFAULT_PORT, "localhost", ctx.asyncAssertSuccess(ack -> {
+
+      client.publishHandler(msg -> {
+        received.incrementAndGet();
+      });
+
+      // Pause the client immediately
+      client.pause();
+
+      // Publish a message from client to trigger server echo
+      // Since client is paused for READING, it should still be able to WRITE.
+      client.publish("test/topic", Buffer.buffer("hello"), MqttQoS.AT_MOST_ONCE, false, false);
+
+      // Wait a bit to ensure nothing is received, check that nothing is receives, and resume reading
+      vertx.setTimer(1_000, id -> {
+
+        ctx.assertEquals(0, received.get(), "Should not receive message while paused");
+
+        // Wait for message
+        long timerId = vertx.setTimer(2_000, id2 -> {
+          ctx.fail("Did not receive message after resume");
+        });
+
+        client.publishHandler(msg -> {
+          vertx.cancelTimer(timerId);
+          received.incrementAndGet();
+          async.complete();
+        });
+
+        client.resume();
+      });
+    }));
+
+    async.await();
+    client.disconnect();
+  }
+}