anna/IM-server-new
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #389 from MBOIX/fix/issue_375

Browse Source 
Fixed issues #375, #312
This commit is contained in:
Andrea Selva
2018-02-11 18:39:57 +01:00
committed by GitHub
parent 5d02ecdf94 b780b71e5e
commit 2fb475ae95
4 changed files with 17 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/BrokerConstants.java
+2
View File
@@ -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";
broker/src/main/java/io/moquette/server/config/IConfig.java
+2
View File
@@ -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();
broker/src/main/java/io/moquette/server/netty/NettyAcceptor.java
+8 -4
View File
@@ -104,6 +104,7 @@ public class NettyAcceptor implements ServerAcceptor {
private boolean nettyTcpNodelay;
private boolean nettySoKeepalive;
private int nettyChannelTimeoutSeconds;
private int maxBytesInMessage;
private Class<? extends ServerSocketChannel> 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());
distribution/src/main/resources/moquette.conf
+5 -1
View File
@@ -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