Commit 607fd7a74f69033f574ba609e04a125a804d38f3

Authored by Sergey Matvienko
Committed by Andrew Shvayka
1 parent 43fc44f0

mqtt transport handler refactored for test

@@ -123,9 +123,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -123,9 +123,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
123 private final SslHandler sslHandler; 123 private final SslHandler sslHandler;
124 private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap; 124 private final ConcurrentMap<MqttTopicMatcher, Integer> mqttQoSMap;
125 125
126 - private final DeviceSessionCtx deviceSessionCtx;  
127 - private volatile InetSocketAddress address;  
128 - private volatile GatewaySessionHandler gatewaySessionHandler; 126 + final DeviceSessionCtx deviceSessionCtx;
  127 + volatile InetSocketAddress address;
  128 + volatile GatewaySessionHandler gatewaySessionHandler;
129 129
130 private final ConcurrentHashMap<String, String> otaPackSessions; 130 private final ConcurrentHashMap<String, String> otaPackSessions;
131 private final ConcurrentHashMap<String, Integer> chunkSizes; 131 private final ConcurrentHashMap<String, Integer> chunkSizes;
@@ -227,7 +227,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -227,7 +227,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
227 } 227 }
228 } 228 }
229 229
230 - private void enqueueRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) { 230 + void enqueueRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
231 final int queueSize = deviceSessionCtx.getMsgQueueSize().incrementAndGet(); 231 final int queueSize = deviceSessionCtx.getMsgQueueSize().incrementAndGet();
232 if (queueSize > context.getMessageQueueSizePerDeviceLimit()) { 232 if (queueSize > context.getMessageQueueSizePerDeviceLimit()) {
233 log.warn("Closing current session because msq queue size for device {} exceed limit {} with msgQueueSize counter {} and actual queue size {}", 233 log.warn("Closing current session because msq queue size for device {} exceed limit {} with msgQueueSize counter {} and actual queue size {}",
@@ -262,7 +262,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -262,7 +262,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
262 } 262 }
263 } 263 }
264 264
265 - private void processRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) { 265 + void processRegularSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
266 switch (msg.fixedHeader().messageType()) { 266 switch (msg.fixedHeader().messageType()) {
267 case PUBLISH: 267 case PUBLISH:
268 processPublish(ctx, (MqttPublishMessage) msg); 268 processPublish(ctx, (MqttPublishMessage) msg);
@@ -628,7 +628,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -628,7 +628,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
628 return new MqttMessage(mqttFixedHeader, mqttMessageIdVariableHeader); 628 return new MqttMessage(mqttFixedHeader, mqttMessageIdVariableHeader);
629 } 629 }
630 630
631 - private void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) { 631 + void processConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {
632 log.info("[{}] Processing connect msg for client: {}!", sessionId, msg.payload().clientIdentifier()); 632 log.info("[{}] Processing connect msg for client: {}!", sessionId, msg.payload().clientIdentifier());
633 String userName = msg.payload().userName(); 633 String userName = msg.payload().userName();
634 String clientId = msg.payload().clientIdentifier(); 634 String clientId = msg.payload().clientIdentifier();
@@ -714,7 +714,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @@ -714,7 +714,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
714 return null; 714 return null;
715 } 715 }
716 716
717 - private void processDisconnect(ChannelHandlerContext ctx) { 717 + void processDisconnect(ChannelHandlerContext ctx) {
718 ctx.close(); 718 ctx.close();
719 log.info("[{}] Client disconnected!", sessionId); 719 log.info("[{}] Client disconnected!", sessionId);
720 doDisconnect(); 720 doDisconnect();