Commit 7dfed3e14a6e060df246e88d73130ece5490e585

Authored by Sergey Matvienko
Committed by Andrew Shvayka
1 parent efe123d9

removed msg-processor-on-device-connect executor. will use transportCallbackExecutor instead

... ... @@ -68,23 +68,4 @@ public class MqttTransportContext extends TransportContext {
68 68 @Value("${transport.mqtt.msg_queue_size_per_device_limit:100}")
69 69 private int messageQueueSizePerDeviceLimit;
70 70
71   - @Getter
72   - private ExecutorService msqProcessorExecutor;
73   -
74   - @Override
75   - @PostConstruct
76   - public void init() {
77   - super.init();
78   - msqProcessorExecutor = ThingsBoardExecutors.newWorkStealingPool(Runtime.getRuntime().availableProcessors() + 1, "msg-processor-on-device-connect");
79   - }
80   -
81   - @Override
82   - @PreDestroy
83   - public void stop() {
84   - super.stop();
85   - if (msqProcessorExecutor != null) {
86   - msqProcessorExecutor.shutdownNow();
87   - }
88   - }
89   -
90 71 }
... ...
... ... @@ -164,8 +164,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
164 164 }
165 165 }
166 166
167   - private void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
168   - address = (InetSocketAddress) ctx.channel().remoteAddress();
  167 + void processMqttMsg(ChannelHandlerContext ctx, MqttMessage msg) {
  168 + address = getAddress(ctx);
169 169 if (msg.fixedHeader() == null) {
170 170 log.info("[{}:{}] Invalid message received", address.getHostName(), address.getPort());
171 171 processDisconnect(ctx);
... ... @@ -181,6 +181,10 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
181 181 }
182 182 }
183 183
  184 + InetSocketAddress getAddress(ChannelHandlerContext ctx) {
  185 + return (InetSocketAddress) ctx.channel().remoteAddress();
  186 + }
  187 +
184 188 private void processProvisionSessionMsg(ChannelHandlerContext ctx, MqttMessage msg) {
185 189 switch (msg.fixedHeader().messageType()) {
186 190 case PUBLISH:
... ... @@ -236,7 +240,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
236 240 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
237 241 }
238 242
239   - private void processMsgQueue(ChannelHandlerContext ctx) {
  243 + void processMsgQueue(ChannelHandlerContext ctx) {
240 244 if (!deviceSessionCtx.isConnected()) {
241 245 log.trace("[{}][{}] Postpone processing msg due to device is not connected. Msg queue size is {}", sessionId, deviceSessionCtx.getDeviceId(), deviceSessionCtx.getMsgQueue().size());
242 246 return;
... ... @@ -821,7 +825,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
821 825 ctx.writeAndFlush(createMqttConnAckMsg(CONNECTION_ACCEPTED, connectMessage));
822 826 deviceSessionCtx.setConnected(true);
823 827 log.info("[{}] Client connected!", sessionId);
824   - context.getMsqProcessorExecutor().execute(()->processMsgQueue(ctx));
  828 + processMsgQueue(ctx);
825 829 }
826 830
827 831 @Override
... ...