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

Fixed issue #380, now catch and print the full stack trace on error

Browse Source
This commit is contained in:
andsel
2018-05-07 20:26:40 +02:00
parent b1aa0019b2
commit 3c9cc5fa8e
2 changed files with 9 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java
+7 -9
View File
@@ -42,7 +42,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
public void channelRead(ChannelHandlerContext ctx, Object message) { public void channelRead(ChannelHandlerContext ctx, Object message) {
MqttMessage msg = (MqttMessage) message; MqttMessage msg = (MqttMessage) message;
MqttMessageType messageType = msg.fixedHeader().messageType(); MqttMessageType messageType = msg.fixedHeader().messageType();
LOG.debug("Processing MQTT message, type={}", messageType); LOG.debug("Processing MQTT message, type: {}", messageType);
try { try {
switch (messageType) { switch (messageType) {
case CONNECT: case CONNECT:
@@ -83,20 +83,18 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE); ctx.writeAndFlush(pingResp).addListener(CLOSE_ON_FAILURE);
break; break;
default: default:
LOG.error("Unkonwn MessageType:{}", messageType); LOG.error("Unknown MessageType: {}", messageType);
break; break;
} }
} catch (Throwable ex) { } catch (Throwable ex) {
LOG.error("Exception was caught while processing MQTT message, {}", ex.getCause(), ex);
ctx.fireExceptionCaught(ex); ctx.fireExceptionCaught(ex);
ctx.close().addListener(CLOSE_ON_FAILURE);
} finally { } finally {
ReferenceCountUtil.release(msg); ReferenceCountUtil.release(msg);
} }
} }
@Override @Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception { public void channelInactive(ChannelHandlerContext ctx) {
String clientID = NettyUtils.clientID(ctx.channel()); String clientID = NettyUtils.clientID(ctx.channel());
if (clientID != null && !clientID.isEmpty()) { if (clientID != null && !clientID.isEmpty()) {
LOG.info("Notifying connection lost event. MqttClientId = {}", clientID); LOG.info("Notifying connection lost event. MqttClientId = {}", clientID);
@@ -106,14 +104,14 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
} }
@Override @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
LOG.error("An unexpected exception was caught while processing MQTT message. Closing Netty channel. CId={}, " + LOG.error("Unexpected exception while processing MQTT message. Closing Netty channel. CId={}",
"cause={}, errorMessage={}", NettyUtils.clientID(ctx.channel()), cause.getCause(), cause.getMessage()); NettyUtils.clientID(ctx.channel()), cause);
ctx.close().addListener(CLOSE_ON_FAILURE); ctx.close().addListener(CLOSE_ON_FAILURE);
} }
@Override @Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { public void channelWritabilityChanged(ChannelHandlerContext ctx) {
if (ctx.channel().isWritable()) { if (ctx.channel().isWritable()) {
m_processor.notifyChannelWritable(ctx.channel()); m_processor.notifyChannelWritable(ctx.channel());
} }
broker/src/main/java/io/moquette/spi/impl/SessionsRepository.java
+2 -1
View File
@@ -27,6 +27,7 @@ import java.util.Collection;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@@ -48,7 +49,7 @@ public class SessionsRepository {
private final ISessionsStore sessions; private final ISessionsStore sessions;
private ISubscriptionsStore subscriptionsStore; private ISubscriptionsStore subscriptionsStore;
private ScheduledExecutorService scheduler; private ScheduledExecutorService scheduler;
private final Map<String, ClientSession> sessionsCache = new ConcurrentHashMap<>(); private final ConcurrentMap<String, ClientSession> sessionsCache = new ConcurrentHashMap<>();
public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) { public SessionsRepository(ISessionsStore sessionsStore, ScheduledExecutorService scheduler) {
this.sessions = sessionsStore; this.sessions = sessionsStore;