Browse Source

Fix reference count issue

pull/5039/head
Andrii Shvaika 5 years ago
parent
commit
9d651a7252
  1. 2
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

2
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@ -237,6 +237,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
} }
deviceSessionCtx.getMsgQueue().add(msg); deviceSessionCtx.getMsgQueue().add(msg);
ReferenceCountUtil.retain(msg);
processMsgQueue(ctx); //Under the normal conditions the msg queue will contain 0 messages. Many messages will be processed on device connect event in separate thread pool processMsgQueue(ctx); //Under the normal conditions the msg queue will contain 0 messages. Many messages will be processed on device connect event in separate thread pool
} }
@ -252,6 +253,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
while ((msg = deviceSessionCtx.getMsgQueue().poll()) != null) { while ((msg = deviceSessionCtx.getMsgQueue().poll()) != null) {
deviceSessionCtx.getMsgQueueSize().decrementAndGet(); deviceSessionCtx.getMsgQueueSize().decrementAndGet();
processRegularSessionMsg(ctx, msg); processRegularSessionMsg(ctx, msg);
ReferenceCountUtil.safeRelease(msg);
} }
} finally { } finally {
deviceSessionCtx.getMsgQueueProcessorLock().unlock(); deviceSessionCtx.getMsgQueueProcessorLock().unlock();

Loading…
Cancel
Save