Fixed issues #375, #312

This commit is contained in:
Mick
2018-02-09 09:55:29 +01:00
parent 5d02ecdf94
commit b780b71e5e
4 changed files with 17 additions and 5 deletions
@@ -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();
@@ -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());