diff --git a/broker/src/main/java/io/moquette/BrokerConstants.java b/broker/src/main/java/io/moquette/BrokerConstants.java index 72686f8a..a60f0e42 100644 --- a/broker/src/main/java/io/moquette/BrokerConstants.java +++ b/broker/src/main/java/io/moquette/BrokerConstants.java @@ -57,6 +57,8 @@ public final class BrokerConstants { public static final String NETTY_SO_KEEPALIVE_PROPERTY_NAME = "netty.so_keepalive"; public static final String NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME = "netty.channel_timeout.seconds"; 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 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/server/config/IConfig.java b/broker/src/main/java/io/moquette/server/config/IConfig.java index 1d2e89d5..c5d3440a 100644 --- a/broker/src/main/java/io/moquette/server/config/IConfig.java +++ b/broker/src/main/java/io/moquette/server/config/IConfig.java @@ -42,6 +42,8 @@ public abstract class IConfig { setProperty(BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME, Boolean.TRUE.toString()); setProperty(BrokerConstants.AUTHENTICATOR_CLASS_NAME, ""); setProperty(BrokerConstants.AUTHORIZATOR_CLASS_NAME, ""); + setProperty(BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME, + String.valueOf(BrokerConstants.DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE)); } public abstract IResourceLoader getResourceLoader(); diff --git a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java index 7f67e1b1..4db0de12 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java +++ b/broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java @@ -104,6 +104,7 @@ public class NettyAcceptor implements ServerAcceptor { private boolean nettyTcpNodelay; private boolean nettySoKeepalive; private int nettyChannelTimeoutSeconds; + private int maxBytesInMessage; private Class channelClass; @@ -121,6 +122,9 @@ public class NettyAcceptor implements ServerAcceptor { .parseBoolean(props.getProperty(BrokerConstants.NETTY_SO_KEEPALIVE_PROPERTY_NAME, "true")); nettyChannelTimeoutSeconds = Integer .parseInt(props.getProperty(BrokerConstants.NETTY_CHANNEL_TIMEOUT_SECONDS_PROPERTY_NAME, "10")); + maxBytesInMessage = Integer + .parseInt(props.getProperty(BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME, + String.valueOf(BrokerConstants.DEFAULT_NETTY_MAX_BYTES_IN_MESSAGE))); boolean epoll = Boolean.parseBoolean(props.getProperty(BrokerConstants.NETTY_EPOLL_PROPERTY_NAME, "false")); if (epoll) { @@ -222,7 +226,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addLast("bugsnagCatcher", errorsCather.get()); } pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("decoder", new MqttDecoder()); + pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); pipeline.addLast("encoder", MqttEncoder.INSTANCE); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("messageLogger", new MQTTMessageLogger()); @@ -261,7 +265,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("decoder", new MqttDecoder()); + pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); pipeline.addLast("encoder", MqttEncoder.INSTANCE); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("messageLogger", new MQTTMessageLogger()); @@ -297,7 +301,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); // pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR)); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("decoder", new MqttDecoder()); + pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); pipeline.addLast("encoder", MqttEncoder.INSTANCE); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("messageLogger", new MQTTMessageLogger()); @@ -336,7 +340,7 @@ public class NettyAcceptor implements ServerAcceptor { pipeline.addFirst("idleStateHandler", new IdleStateHandler(nettyChannelTimeoutSeconds, 0, 0)); pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler); pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector)); - pipeline.addLast("decoder", new MqttDecoder()); + pipeline.addLast("decoder", new MqttDecoder(maxBytesInMessage)); pipeline.addLast("encoder", MqttEncoder.INSTANCE); pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector)); pipeline.addLast("messageLogger", new MQTTMessageLogger()); diff --git a/distribution/src/main/resources/moquette.conf b/distribution/src/main/resources/moquette.conf index c257eb83..b072766a 100644 --- a/distribution/src/main/resources/moquette.conf +++ b/distribution/src/main/resources/moquette.conf @@ -106,11 +106,15 @@ password_file config/password_file.conf #********************************************************************* # Netty Configuration # -# Linux systems can use epoll instead of nio. To get a performance +# netty.epoll: Linux systems can use epoll instead of nio. To get a performance # gain and reduced GC. # http://netty.io/wiki/native-transports.html for more information +# netty.mqtt.message_size : by default the max size of message is set at 8092 bytes +# http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/errata01/os/mqtt-v3.1.1-errata01-os-complete.html#_Toc442180836 +# Fore more information about payload size specs. #********************************************************************* # netty.epoll true +# netty.mqtt.message_size 8092 #********************************************************************* # Metrics Configuration