Minor change, avoid to remove and add to the Netty's pipeline the MoquetteIdleTimoutHandler
This commit is contained in:
andsel
+5
-3
@@ -16,10 +16,12 @@
|
||||
package org.eclipse.moquette.server.netty;
|
||||
|
||||
import io.netty.channel.ChannelDuplexHandler;
|
||||
import io.netty.channel.ChannelHandler.Sharable;
|
||||
import io.netty.channel.ChannelHandlerContext;
|
||||
import io.netty.handler.timeout.IdleState;
|
||||
import io.netty.handler.timeout.IdleStateEvent;
|
||||
|
||||
@Sharable
|
||||
class MoquetteIdleTimoutHandler extends ChannelDuplexHandler {
|
||||
@Override
|
||||
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
|
||||
@@ -29,9 +31,9 @@ class MoquetteIdleTimoutHandler extends ChannelDuplexHandler {
|
||||
//fire a channelInactive to trigger publish of Will
|
||||
ctx.fireChannelInactive();
|
||||
ctx.close();
|
||||
} /*else if (e.getState() == IdleState.WRITER_IDLE) {
|
||||
ctx.writeAndFlush(new PingMessage());
|
||||
}*/
|
||||
}
|
||||
} else {
|
||||
super.userEventTriggered(ctx, evt);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -152,6 +152,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
|
||||
private void initializePlainTCPTransport(IMessaging messaging, IConfig props) throws IOException {
|
||||
final NettyMQTTHandler handler = new NettyMQTTHandler();
|
||||
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
|
||||
handler.setMessaging(messaging);
|
||||
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
|
||||
int port = Integer.parseInt(props.getProperty(Constants.PORT_PROPERTY_NAME));
|
||||
@@ -159,7 +160,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
@Override
|
||||
void init(ChannelPipeline pipeline) {
|
||||
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
|
||||
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());
|
||||
@@ -180,6 +181,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
int port = Integer.parseInt(webSocketPortProp);
|
||||
|
||||
final NettyMQTTHandler handler = new NettyMQTTHandler();
|
||||
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
|
||||
handler.setMessaging(messaging);
|
||||
|
||||
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
|
||||
@@ -190,11 +192,10 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
pipeline.addLast("httpDecoder", new HttpRequestDecoder());
|
||||
pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
|
||||
pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler("/mqtt", "mqtt, mqttv3.1, mqttv3.1.1"));
|
||||
//pipeline.addLast("webSocketHandler", new WebSocketServerProtocolHandler(null, "mqtt"));
|
||||
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
|
||||
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
|
||||
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
|
||||
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
|
||||
pipeline.addLast("decoder", new MQTTDecoder());
|
||||
pipeline.addLast("encoder", new MQTTEncoder());
|
||||
@@ -216,6 +217,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
LOG.info("Starting SSL on port {}", sslPort);
|
||||
|
||||
final NettyMQTTHandler handler = new NettyMQTTHandler();
|
||||
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
|
||||
handler.setMessaging(messaging);
|
||||
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
|
||||
initFactory(host, sslPort, new PipelineInitializer() {
|
||||
@@ -223,7 +225,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
void init(ChannelPipeline pipeline) throws Exception {
|
||||
pipeline.addLast("ssl", sslHandlerFactory.create());
|
||||
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
|
||||
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());
|
||||
@@ -243,6 +245,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
}
|
||||
int sslPort = Integer.parseInt(sslPortProp);
|
||||
final NettyMQTTHandler handler = new NettyMQTTHandler();
|
||||
final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
|
||||
handler.setMessaging(messaging);
|
||||
String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
|
||||
initFactory(host, sslPort, new PipelineInitializer() {
|
||||
@@ -256,7 +259,7 @@ public class NettyAcceptor implements ServerAcceptor {
|
||||
pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
|
||||
pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
|
||||
pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
|
||||
pipeline.addAfter("idleStateHandler", "idleEventHandler", timeoutHandler);
|
||||
pipeline.addFirst("bytemetrics", new BytesMetricsHandler(m_bytesMetricsCollector));
|
||||
pipeline.addLast("decoder", new MQTTDecoder());
|
||||
pipeline.addLast("encoder", new MQTTEncoder());
|
||||
|
||||
@@ -55,11 +55,7 @@ public class NettyChannel implements ServerChannel {
|
||||
if (m_channel.pipeline().names().contains("idleStateHandler")) {
|
||||
m_channel.pipeline().remove("idleStateHandler");
|
||||
}
|
||||
if (m_channel.pipeline().names().contains("idleEventHandler")) {
|
||||
m_channel.pipeline().remove("idleEventHandler");
|
||||
}
|
||||
m_channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
|
||||
m_channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
|
||||
}
|
||||
|
||||
public void close(boolean immediately) {
|
||||
|
||||
@@ -101,7 +101,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
public void init(IConfig configProps) {
|
||||
subscriptions = new SubscriptionsStore();
|
||||
m_executor = Executors.newFixedThreadPool(1);
|
||||
m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
|
||||
m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 128, m_executor); //1024 * 32
|
||||
/*Disruptor<ValueEvent> m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
|
||||
ProducerType.MULTI, new BusySpinWaitStrategy());*/
|
||||
m_disruptor.handleEventsWith(this);
|
||||
@@ -115,13 +115,11 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
|
||||
}
|
||||
|
||||
private void disruptorPublish(MessagingEvent msgEvent) {
|
||||
LOG.debug("disruptorPublish publishing event {}", msgEvent);
|
||||
LOG.debug("disruptorPublish publishing event {}, remaining capacity {}", msgEvent, m_ringBuffer.remainingCapacity());
|
||||
long sequence = m_ringBuffer.next();
|
||||
ValueEvent event = m_ringBuffer.get(sequence);
|
||||
|
||||
event.setEvent(msgEvent);
|
||||
|
||||
m_ringBuffer.publish(sequence);
|
||||
m_ringBuffer.publish(sequence);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user