Flush written data in MQTTConnection#sendIfWritableElseDrop (moquette-io#454)

komamitsu · andsel · commit c248096b4e69 · 2019-03-17T17:34:00.000+01:00
* Flush written data in MQTTConnection#sendIfWritableElseDrop

* Add a new configuration `immediateBufferFlush`

* Add BrokerConfigurationTest

* Fix NPE in SessionTest
diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java
@@ -68,6 +68,7 @@ public final class BrokerConstants {
     public static final String NETTY_EPOLL_PROPERTY_NAME = "netty.epoll";
     public static final String NETTY_MAX_BYTES_PROPERTY_NAME = "netty.mqtt.message_size";
     public static final int DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE = 8092;
+    public static final String IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME = "immediate_buffer_flush";
     public static final String METRICS_ENABLE_PROPERTY_NAME = "use_metrics";
     public static final String METRICS_LIBRATO_EMAIL_PROPERTY_NAME = "metrics.librato.email";
     public static final String METRICS_LIBRATO_TOKEN_PROPERTY_NAME = "metrics.librato.token";
diff --git a/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java b/broker/src/main/java/io/moquette/broker/BrokerConfiguration.java
@@ -23,18 +23,21 @@ class BrokerConfiguration {
     private final boolean allowAnonymous;
     private final boolean allowZeroByteClientId;
     private final boolean reauthorizeSubscriptionsOnConnect;
+    private final boolean immediateBufferFlush;
 
     BrokerConfiguration(IConfig props) {
         allowAnonymous = props.boolProp(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, true);
         allowZeroByteClientId = props.boolProp(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, false);
         reauthorizeSubscriptionsOnConnect = props.boolProp(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, false);
+        immediateBufferFlush = props.boolProp(BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME, false);
     }
 
     public BrokerConfiguration(boolean allowAnonymous, boolean allowZeroByteClientId,
-                               boolean reauthorizeSubscriptionsOnConnect) {
+                               boolean reauthorizeSubscriptionsOnConnect, boolean immediateBufferFlush) {
         this.allowAnonymous = allowAnonymous;
         this.allowZeroByteClientId = allowZeroByteClientId;
         this.reauthorizeSubscriptionsOnConnect = reauthorizeSubscriptionsOnConnect;
+        this.immediateBufferFlush = immediateBufferFlush;
     }
 
     public boolean isAllowAnonymous() {
@@ -48,4 +51,8 @@ public boolean isAllowZeroByteClientId() {
     public boolean isReauthorizeSubscriptionsOnConnect() {
         return reauthorizeSubscriptionsOnConnect;
     }
+
+    public boolean isImmediateBufferFlush() {
+        return immediateBufferFlush;
+    }
 }
diff --git a/broker/src/main/java/io/moquette/broker/MQTTConnection.java b/broker/src/main/java/io/moquette/broker/MQTTConnection.java
@@ -19,6 +19,7 @@
 import io.moquette.broker.security.IAuthenticator;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.mqtt.*;
 import io.netty.handler.timeout.IdleStateHandler;
@@ -402,7 +403,14 @@ void sendIfWritableElseDrop(MqttMessage msg) {
             LOG.debug("OUT {} on channel {}", msg.fixedHeader().messageType(), channel);
         }
         if (channel.isWritable()) {
-            channel.write(msg).addListener(FIRE_EXCEPTION_ON_FAILURE);
+            ChannelFuture channelFuture;
+            if (brokerConfig.isImmediateBufferFlush()) {
+                channelFuture = channel.writeAndFlush(msg);
+            }
+            else {
+                channelFuture = channel.write(msg);
+            }
+            channelFuture.addListener(FIRE_EXCEPTION_ON_FAILURE);
         }
     }
 
diff --git a/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java b/broker/src/test/java/io/moquette/broker/BrokerConfigurationTest.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright (c) 2012-2018 The original author or authors
+ * ------------------------------------------------------
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ * The Eclipse Public License is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * The Apache License v2.0 is available at
+ * http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.moquette.broker;
+
+import io.moquette.BrokerConstants;
+import io.moquette.broker.config.MemoryConfig;
+import org.junit.Test;
+
+import java.util.Properties;
+
+import static org.junit.Assert.*;
+
+public class BrokerConfigurationTest {
+
+    @Test
+    public void defaultConfig() {
+        MemoryConfig config = new MemoryConfig(new Properties());
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(config);
+        assertTrue(brokerConfiguration.isAllowAnonymous());
+        assertFalse(brokerConfiguration.isAllowZeroByteClientId());
+        assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
+        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+    }
+
+    @Test
+    public void configureAllowAnonymous() {
+        Properties properties = new Properties();
+        properties.put(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, "false");
+        MemoryConfig config = new MemoryConfig(properties);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(config);
+        assertFalse(brokerConfiguration.isAllowAnonymous());
+        assertFalse(brokerConfiguration.isAllowZeroByteClientId());
+        assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
+        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+    }
+
+    @Test
+    public void configureAllowZeroByteClientId() {
+        Properties properties = new Properties();
+        properties.put(BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME, "true");
+        MemoryConfig config = new MemoryConfig(properties);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(config);
+        assertTrue(brokerConfiguration.isAllowAnonymous());
+        assertTrue(brokerConfiguration.isAllowZeroByteClientId());
+        assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
+        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+    }
+
+    @Test
+    public void configureReauthorizeSubscriptionsOnConnect() {
+        Properties properties = new Properties();
+        properties.put(BrokerConstants.REAUTHORIZE_SUBSCRIPTIONS_ON_CONNECT, "true");
+        MemoryConfig config = new MemoryConfig(properties);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(config);
+        assertTrue(brokerConfiguration.isAllowAnonymous());
+        assertFalse(brokerConfiguration.isAllowZeroByteClientId());
+        assertTrue(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
+        assertFalse(brokerConfiguration.isImmediateBufferFlush());
+    }
+
+    @Test
+    public void configureImmediateBufferFlush() {
+        Properties properties = new Properties();
+        properties.put(BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME, "true");
+        MemoryConfig config = new MemoryConfig(properties);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(config);
+        assertTrue(brokerConfiguration.isAllowAnonymous());
+        assertFalse(brokerConfiguration.isAllowZeroByteClientId());
+        assertFalse(brokerConfiguration.isReauthorizeSubscriptionsOnConnect());
+        assertTrue(brokerConfiguration.isImmediateBufferFlush());
+    }
+}
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java
@@ -49,7 +49,7 @@ public class MQTTConnectionConnectTest {
     private EmbeddedChannel channel;
     private SessionRegistry sessionRegistry;
     private MqttMessageBuilders.ConnectBuilder connMsg;
-    private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
+    private static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
     private IAuthenticator mockAuthenticator;
     private PostOffice postOffice;
     private MemoryQueueRepository queueRepository;
@@ -205,7 +205,7 @@ public void noPasswdAuthentication() {
     @Test
     public void prohibitAnonymousClient() {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID).build();
-        BrokerConfiguration config = new BrokerConfiguration(false, true, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, true, false, false);
 
         sut = createMQTTConnection(config);
         channel = (EmbeddedChannel) sut.channel;
@@ -223,7 +223,7 @@ public void prohibitAnonymousClient_providingUsername() {
         MqttConnectMessage msg = connMsg.clientId(FAKE_CLIENT_ID)
             .username(TEST_USER + "_fake")
             .build();
-        BrokerConfiguration config = new BrokerConfiguration(false, true, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, true, false, false);
 
         createMQTTConnection(config);
 
@@ -237,7 +237,7 @@ public void prohibitAnonymousClient_providingUsername() {
 
     @Test
     public void testZeroByteClientIdNotAllowed() {
-        BrokerConfiguration config = new BrokerConfiguration(false, false, false);
+        BrokerConfiguration config = new BrokerConfiguration(false, false, false, false);
 
         sut = createMQTTConnection(config);
         channel = (EmbeddedChannel) sut.channel;
@@ -288,7 +288,7 @@ public void testBindWithSameClientIDBadCredentialsDoesntDropExistingClient() {
         EmbeddedChannel evilChannel = new EmbeddedChannel();
 
         // Exercise
-        BrokerConfiguration config = new BrokerConfiguration(true, true, false);
+        BrokerConfiguration config = new BrokerConfiguration(true, true, false, false);
         final MQTTConnection evilConnection = createMQTTConnection(config, evilChannel, postOffice);
         evilConnection.processConnect(evilClientConnMsg);
 
diff --git a/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java b/broker/src/test/java/io/moquette/broker/MQTTConnectionPublishTest.java
@@ -51,7 +51,7 @@ public class MQTTConnectionPublishTest {
     public void setUp() {
         connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
 
-        BrokerConfiguration config = new BrokerConfiguration(true, true, false);
+        BrokerConfiguration config = new BrokerConfiguration(true, true, false, false);
 
         createMQTTConnection(config);
     }
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeInternalPublishTest.java
@@ -52,7 +52,7 @@ public class PostOfficeInternalPublishTest {
     private SessionRegistry sessionRegistry;
     private MockAuthenticator mockAuthenticator;
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
-        new BrokerConfiguration(true, true, false);
+        new BrokerConfiguration(true, true, false, false);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
 
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java b/broker/src/test/java/io/moquette/broker/PostOfficePublishTest.java
@@ -62,7 +62,7 @@ public class PostOfficePublishTest {
     private SessionRegistry sessionRegistry;
     private MockAuthenticator mockAuthenticator;
     static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZERO_BYTES_CLID =
-        new BrokerConfiguration(true, true, false);
+        new BrokerConfiguration(true, true, false, false);
     private MemoryRetainedRepository retainedRepository;
     private MemoryQueueRepository queueRepository;
 
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeSubscribeTest.java
@@ -64,7 +64,7 @@ public class PostOfficeSubscribeTest {
     private MqttConnectMessage connectMessage;
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
-    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
+    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
     private MemoryQueueRepository queueRepository;
 
     @Before
diff --git a/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java b/broker/src/test/java/io/moquette/broker/PostOfficeUnsubscribeTest.java
@@ -55,7 +55,7 @@ public class PostOfficeUnsubscribeTest {
     private MqttConnectMessage connectMessage;
     private IAuthenticator mockAuthenticator;
     private SessionRegistry sessionRegistry;
-    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false);
+    public static final BrokerConfiguration CONFIG = new BrokerConfiguration(true, true, false, false);
     private MemoryQueueRepository queueRepository;
 
     @Before
diff --git a/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java b/broker/src/test/java/io/moquette/broker/SessionRegistryTest.java
@@ -46,7 +46,7 @@ public class SessionRegistryTest {
     private SessionRegistry sut;
     private MqttMessageBuilders.ConnectBuilder connMsg;
     private static final BrokerConfiguration ALLOW_ANONYMOUS_AND_ZEROBYTE_CLIENT_ID =
-        new BrokerConfiguration(true, true, false);
+        new BrokerConfiguration(true, true, false, false);
     private MemoryQueueRepository queueRepository;
 
     @Before
diff --git a/broker/src/test/java/io/moquette/broker/SessionTest.java b/broker/src/test/java/io/moquette/broker/SessionTest.java
@@ -20,7 +20,8 @@ public void testPubAckDrainMessagesRemainingInQueue() {
         final Queue<SessionRegistry.EnqueuedMessage> queuedMessages = new ConcurrentLinkedQueue<>();
         final Session client = new Session("Subscriber", true, null, queuedMessages);
         final EmbeddedChannel testChannel = new EmbeddedChannel();
-        MQTTConnection mqttConnection = new MQTTConnection(testChannel, null, null, null, null);
+        BrokerConfiguration brokerConfiguration = new BrokerConfiguration(true, false, false, false);
+        MQTTConnection mqttConnection = new MQTTConnection(testChannel, brokerConfiguration, null, null, null);
         client.markConnected();
         client.bind(mqttConnection);
 

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ public class MQTTConnectionPublishTest {`
`51`	`51`	`public void setUp() {`
`52`	`52`	`connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);`
`53`	`53`
`54`		`- BrokerConfiguration config = new BrokerConfiguration(true, true, false);`
	`54`	`+ BrokerConfiguration config = new BrokerConfiguration(true, true, false, false);`
`55`	`55`
`56`	`56`	`createMQTTConnection(config);`
`57`	`57`	`}`