diff --git a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java index 88b04c7613..03830c1c4c 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ruleChain/DefaultTbContext.java @@ -1063,18 +1063,16 @@ public class DefaultTbContext implements TbContext { @Override public void checkTenantEntity(EntityId entityId) throws TbNodeException { TenantId actualTenantId = TenantIdLoader.findTenantId(this, entityId); - assertSameTenantId(actualTenantId, entityId); + if (!getTenantId().equals(actualTenantId)) { + throw new TbNodeException("Entity with id: '" + entityId + "' specified in the configuration doesn't belong to the current tenant.", true); + } } @Override - public & HasTenantId, I extends EntityId> void checkTenantEntity(E entity) throws TbNodeException { + public & HasTenantId, I extends EntityId> void checkTenantOrSystemEntity(E entity) throws TbNodeException { TenantId actualTenantId = entity.getTenantId(); - assertSameTenantId(actualTenantId, entity.getId()); - } - - private void assertSameTenantId(TenantId tenantId, EntityId entityId) throws TbNodeException { - if (!getTenantId().equals(tenantId)) { - throw new TbNodeException("Entity with id: '" + entityId + "' specified in the configuration doesn't belong to the current tenant.", true); + if (!getTenantId().equals(actualTenantId) && !actualTenantId.isSysTenantId()) { + throw new TbNodeException("Entity with id: '" + entity.getId() + "' specified in the configuration doesn't belong to the current or system tenant.", true); } } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java index 6543955ed5..9856aa18cb 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java @@ -25,10 +25,10 @@ import org.springframework.boot.test.mock.mockito.SpyBean; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; import org.thingsboard.server.transport.mqtt.gateway.GatewayMetricsService; -import org.thingsboard.server.common.msg.gateway.metrics.GatewayMetadata; import org.thingsboard.server.transport.mqtt.gateway.metrics.GatewayMetricsState; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestCallback; import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; @@ -43,6 +43,7 @@ import java.util.Random; import java.util.Set; import java.util.concurrent.TimeUnit; +import static org.awaitility.Awaitility.await; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.mockito.Mockito.any; @@ -112,6 +113,42 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt processGatewayTelemetryTest(GATEWAY_TELEMETRY_TOPIC, expectedKeys, payload.getBytes(), deviceName1, deviceName2); } + @Test + public void testAckIsReceivedOnFailedPublishMessage() throws Exception { + String devicePayload = "[{\"ts\": 10000, \"values\": " + PAYLOAD_VALUES_STR + "}]"; + String payloadA = "{\"Device A\": " + devicePayload + "}"; + + String deviceBPayload = "[{\"ts\": 10000, \"values\": " + PAYLOAD_VALUES_STR + "}]"; + String payloadB = "{\"Device B\": " + deviceBPayload + "}"; + + testAckIsReceivedOnFailedPublishMessage("Device A", payloadA.getBytes(), "Device B", payloadB.getBytes()); + } + + protected void testAckIsReceivedOnFailedPublishMessage(String deviceName1, byte[] payload1, String deviceName2, byte[] payload2) throws Exception { + updateDefaultTenantProfileConfig(profileConfiguration -> { + profileConfiguration.setMaxDevices(3); + }); + + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(gatewayAccessToken); + client.publishAndWait(GATEWAY_TELEMETRY_TOPIC, payload1); + + // check device is created + await().atMost(TIMEOUT, TimeUnit.SECONDS).untilAsserted(() -> { + assertNotNull(doGet("/api/tenant/devices?deviceName=" + deviceName1, Device.class)); + }); + + client.publishAndWait(GATEWAY_TELEMETRY_TOPIC, payload2); + client.disconnectAndWait(); + + // check device was not created due to limit + doGet("/api/tenant/devices?deviceName=" + deviceName2).andExpect(status().isNotFound()); + + updateDefaultTenantProfileConfig(profileConfiguration -> { + profileConfiguration.setMaxDevices(0); + }); + } + @Test public void testGatewayConnect() throws Exception { String payload = "{\"device\":\"Device A\"}"; diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java index a23506a160..92b35d896c 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesJsonIntegrationTest.java @@ -186,4 +186,15 @@ public abstract class AbstractMqttTimeseriesJsonIntegrationTest extends Abstract assertFalse(callback.isPubAckReceived()); } + @Override + public void testAckIsReceivedOnFailedPublishMessage() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .deviceName("Test Post Telemetry device json payload") + .gatewayName("Test Post Telemetry gateway json payload") + .transportPayloadType(TransportPayloadType.JSON) + .telemetryTopicFilter(POST_DATA_TELEMETRY_TOPIC) + .build(); + processBeforeTest(configProperties); + super.testAckIsReceivedOnFailedPublishMessage(); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java index 1b6c247d96..60e5857c04 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java @@ -459,6 +459,31 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac assertFalse(callback.isPubAckReceived()); } + @Override + public void testAckIsReceivedOnFailedPublishMessage() throws Exception { + MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() + .deviceName("Test Post Telemetry device proto payload") + .gatewayName("Test Post Telemetry gateway proto payload") + .transportPayloadType(TransportPayloadType.PROTOBUF) + .telemetryTopicFilter(POST_DATA_TELEMETRY_TOPIC) + .build(); + processBeforeTest(configProperties); + + TransportApiProtos.GatewayTelemetryMsg.Builder gatewayTelemetryMsgProtoBuilder = TransportApiProtos.GatewayTelemetryMsg.newBuilder(); + List expectedKeys = Arrays.asList("key1", "key2", "key3", "key4", "key5"); + String deviceName1 = "Device A"; + String deviceName2 = "Device B"; + TransportApiProtos.TelemetryMsg deviceATelemetryMsgProto = getDeviceTelemetryMsgProto(deviceName1, expectedKeys, 10000, 20000); + gatewayTelemetryMsgProtoBuilder.addAllMsg(List.of(deviceATelemetryMsgProto)); + TransportApiProtos.GatewayTelemetryMsg payload1 = gatewayTelemetryMsgProtoBuilder.build(); + + TransportApiProtos.TelemetryMsg deviceBTelemetryMsgProto = getDeviceTelemetryMsgProto(deviceName2, expectedKeys, 10000, 20000); + TransportApiProtos.GatewayTelemetryMsg payload2 = TransportApiProtos.GatewayTelemetryMsg.newBuilder() + .addAllMsg(List.of(deviceBTelemetryMsgProto)) + .build(); + super.testAckIsReceivedOnFailedPublishMessage(deviceName1, payload1.toByteArray(), deviceName2, payload2.toByteArray()); + } + private DynamicSchema getDynamicSchema() { DeviceProfileTransportConfiguration transportConfiguration = deviceProfile.getProfileData().getTransportConfiguration(); assertTrue(transportConfiguration instanceof MqttDeviceProfileTransportConfiguration); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java index 91e7dedbf2..fee2f04e24 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewaySessionHandler.java @@ -80,6 +80,8 @@ import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; @@ -405,18 +407,34 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : json.getAsJsonObject().entrySet()) { - if (!deviceEntry.getValue().isJsonArray()) { - log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); - continue; - } + + List> deviceEntries = json.getAsJsonObject().entrySet().stream() + .filter(entry -> { + final boolean isArray = entry.getValue().isJsonArray(); + if (!isArray) { + log.warn("{} device='{}' value={}", CAN_T_PARSE_VALUE, entry.getKey(), entry.getValue()); + } + return isArray; + }) + .toList(); + + if (deviceEntries.isEmpty()) { + log.debug("[{}][{}][{}] Devices telemetry message is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); + throw new IllegalArgumentException("[" + sessionId + "] Devices telemetry message is empty for [" + gateway.getDeviceId() + "]"); + } + + AtomicInteger remaining = new AtomicInteger(deviceEntries.size()); + AtomicBoolean ackSent = new AtomicBoolean(false); + + for (Map.Entry deviceEntry : deviceEntries) { String deviceName = deviceEntry.getKey(); - process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> failedToProcessLog(deviceName, TELEMETRY, t)); + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, TELEMETRY, ackSent, t)); } } - private void processPostTelemetryMsg(T deviceCtx, JsonElement msg, String deviceName, int msgId) { + private void processPostTelemetryMsg(T deviceCtx, JsonElement msg, String deviceName, int msgId, AtomicInteger remaining, AtomicBoolean ackSent) { try { long systemTs = System.currentTimeMillis(); TbPair> gatewayPayloadPair = JsonConverter.convertToGatewayTelemetry(msg.getAsJsonArray(), systemTs); @@ -425,10 +443,10 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(telemetryMsg.getDeviceName()); - process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId), - t -> failedToProcessLog(deviceName, TELEMETRY, t)); + process(deviceName, deviceCtx -> processPostTelemetryMsg(deviceCtx, telemetryMsg.getMsg(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, TELEMETRY, ackSent, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - protected void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg msg, String deviceName, int msgId) { + protected void processPostTelemetryMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostTelemetryMsg msg, String deviceName, int msgId, + AtomicInteger remaining, AtomicBoolean ackSent) { try { TransportProtos.PostTelemetryMsg postTelemetryMsg = ProtoConverter.validatePostTelemetryMsg(msg.toByteArray()); - transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getPubAckCallback(channel, deviceName, msgId, postTelemetryMsg)); + transportService.process(deviceCtx.getSessionInfo(), postTelemetryMsg, getAggregatePubAckCallback(channel, msgId, deviceName, postTelemetryMsg, remaining, ackSent)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert telemetry: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } @@ -475,26 +498,42 @@ public abstract class AbstractGatewaySessionHandler deviceEntry : json.getAsJsonObject().entrySet()) { - if (!deviceEntry.getValue().isJsonObject()) { - log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); - continue; - } + List> deviceEntries = json.getAsJsonObject().entrySet().stream() + .filter(entry -> { + boolean isJsonObject = entry.getValue().isJsonObject(); + if (!isJsonObject) { + log.warn("{} device='{}' value={}", CAN_T_PARSE_VALUE, entry.getKey(), entry.getValue()); + } + return isJsonObject; + }) + .toList(); + + if (deviceEntries.isEmpty()) { + log.debug("[{}][{}][{}] Devices claim message is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); + throw new IllegalArgumentException("[" + sessionId + "] Devices claim message is empty for [" + gateway.getDeviceId() + "]"); + } + + AtomicInteger remaining = new AtomicInteger(deviceEntries.size()); + AtomicBoolean ackSent = new AtomicBoolean(false); + + for (Map.Entry deviceEntry : deviceEntries) { String deviceName = deviceEntry.getKey(); - process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> failedToProcessLog(deviceName, CLAIMING, t)); + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, CLAIMING, ackSent, t)); } } - private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement claimRequest, String deviceName, int msgId) { + private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement claimRequest, String deviceName, int msgId, + AtomicInteger remaining, AtomicBoolean ackSent) { try { DeviceId deviceId = deviceCtx.getDeviceId(); TransportProtos.ClaimDeviceMsg claimDeviceMsg = JsonConverter.convertToClaimDeviceProto(deviceId, claimRequest); - transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg)); + transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getAggregatePubAckCallback(channel, msgId, deviceName, claimDeviceMsg, remaining, ackSent)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, claimRequest, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } @@ -507,49 +546,70 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(claimDeviceMsg.getDeviceName()); - process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId), - t -> failedToProcessLog(deviceName, CLAIMING, t)); + process(deviceName, deviceCtx -> processClaimDeviceMsg(deviceCtx, claimDeviceMsg.getClaimRequest(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, CLAIMING, ackSent, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, TransportApiProtos.ClaimDevice claimRequest, String deviceName, int msgId) { + private void processClaimDeviceMsg(MqttDeviceAwareSessionContext deviceCtx, TransportApiProtos.ClaimDevice claimRequest, String deviceName, int msgId, + AtomicInteger remaining, AtomicBoolean ackSent) { try { DeviceId deviceId = deviceCtx.getDeviceId(); TransportProtos.ClaimDeviceMsg claimDeviceMsg = ProtoConverter.convertToClaimDeviceProto(deviceId, claimRequest.toByteArray()); - transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getPubAckCallback(channel, deviceName, msgId, claimDeviceMsg)); + transportService.process(deviceCtx.getSessionInfo(), claimDeviceMsg, getAggregatePubAckCallback(channel, msgId, deviceName, claimDeviceMsg, remaining, ackSent)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to convert claim message: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, claimRequest, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } private void onDeviceAttributesJson(int msgId, ByteBuf payload) throws AdaptorException { JsonElement json = JsonMqttAdaptor.validateJsonPayload(sessionId, payload); validateJsonObject(json); - for (Map.Entry deviceEntry : json.getAsJsonObject().entrySet()) { - if (!deviceEntry.getValue().isJsonObject()) { - log.warn("{}[{}]", CAN_T_PARSE_VALUE, json); - continue; - } + List> deviceEntries = json.getAsJsonObject().entrySet().stream() + .filter(entry -> { + boolean isJsonObject = entry.getValue().isJsonObject(); + if (!isJsonObject) { + log.warn("{} device='{}' value={}", CAN_T_PARSE_VALUE, entry.getKey(), entry.getValue()); + } + return isJsonObject; + }) + .toList(); + + if (deviceEntries.isEmpty()) { + log.debug("[{}][{}][{}] Devices attribute message is empty", gateway.getTenantId(), gateway.getDeviceId(), sessionId); + throw new IllegalArgumentException("[" + sessionId + "] Devices attribute message is empty for [" + gateway.getDeviceId() + "]"); + } + + AtomicInteger remaining = new AtomicInteger(deviceEntries.size()); + AtomicBoolean ackSent = new AtomicBoolean(false); + + for (Map.Entry deviceEntry : deviceEntries) { String deviceName = deviceEntry.getKey(); - process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId), - t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, deviceEntry.getValue(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, ATTRIBUTE, ackSent, t)); } } - private void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement msg, String deviceName, int msgId) { + private void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, JsonElement msg, String deviceName, int msgId, + AtomicInteger remaining, AtomicBoolean ackSent) { try { TransportProtos.PostAttributeMsg postAttributeMsg = JsonConverter.convertToAttributesProto(msg.getAsJsonObject()); - transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getAggregatePubAckCallback(channel, msgId, deviceName, postAttributeMsg, remaining, ackSent)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, msg, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } @@ -562,23 +622,28 @@ public abstract class AbstractGatewaySessionHandler { String deviceName = checkDeviceName(attributesMsg.getDeviceName()); - process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId), - t -> failedToProcessLog(deviceName, ATTRIBUTE, t)); + process(deviceName, deviceCtx -> processPostAttributesMsg(deviceCtx, attributesMsg.getMsg(), deviceName, msgId, + remaining, ackSent), + t -> processFailure(msgId, deviceName, ATTRIBUTE, ackSent, t)); }); } catch (RuntimeException | InvalidProtocolBufferException e) { throw new AdaptorException(e); } } - protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostAttributeMsg kvListProto, String deviceName, int msgId) { + protected void processPostAttributesMsg(MqttDeviceAwareSessionContext deviceCtx, TransportProtos.PostAttributeMsg kvListProto, String deviceName, int msgId, + AtomicInteger remaining, AtomicBoolean ackSent) { try { TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); - transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getPubAckCallback(channel, deviceName, msgId, postAttributeMsg)); + transportService.process(deviceCtx.getSessionInfo(), postAttributeMsg, getAggregatePubAckCallback(channel, msgId, deviceName, postAttributeMsg, remaining, ackSent)); } catch (Throwable e) { log.warn("[{}][{}][{}] Failed to process device attributes command: [{}]", gateway.getTenantId(), gateway.getDeviceId(), deviceName, kvListProto, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } @@ -647,27 +712,34 @@ public abstract class AbstractGatewaySessionHandler processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId), - t -> failedToProcessLog(deviceName, RPC_RESPONSE, t)); + AtomicInteger remaining = new AtomicInteger(1); + AtomicBoolean ackSent = new AtomicBoolean(false); + process(deviceName, deviceCtx -> processRpcResponseMsg(deviceCtx, requestId, data, deviceName, msgId, remaining, ackSent), + t -> processFailure(msgId, deviceName, RPC_RESPONSE, ackSent, t)); } - private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, Integer requestId, String data, String deviceName, int msgId) { + private void processRpcResponseMsg(MqttDeviceAwareSessionContext deviceCtx, Integer requestId, String data, String deviceName, + int msgId, AtomicInteger remaining, AtomicBoolean ackSent) { TransportProtos.ToDeviceRpcResponseMsg rpcResponseMsg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder() .setRequestId(requestId).setPayload(data).build(); - transportService.process(deviceCtx.getSessionInfo(), rpcResponseMsg, getPubAckCallback(channel, deviceName, msgId, rpcResponseMsg)); + transportService.process(deviceCtx.getSessionInfo(), rpcResponseMsg, + getAggregatePubAckCallback(channel, msgId, deviceName, rpcResponseMsg, remaining, ackSent)); } private void processGetAttributeRequestMessage(MqttPublishMessage mqttMsg, String deviceName, TransportProtos.GetAttributeRequestMsg requestMsg) { int msgId = getMsgId(mqttMsg); - process(deviceName, deviceCtx -> processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId), - t -> { - failedToProcessLog(deviceName, ATTRIBUTES_REQUEST, t); - ack(mqttMsg, MqttReasonCodes.PubAck.IMPLEMENTATION_SPECIFIC_ERROR); - }); + AtomicInteger remaining = new AtomicInteger(1); + AtomicBoolean ackSent = new AtomicBoolean(false); + process(deviceName, deviceCtx -> { + processGetAttributeRequestMessage(deviceCtx, requestMsg, deviceName, msgId, remaining, ackSent); + }, + t -> processFailure(msgId, deviceName, ATTRIBUTES_REQUEST, ackSent, MqttReasonCodes.PubAck.IMPLEMENTATION_SPECIFIC_ERROR, t)); } - private void processGetAttributeRequestMessage(T deviceCtx, TransportProtos.GetAttributeRequestMsg requestMsg, String deviceName, int msgId) { - transportService.process(deviceCtx.getSessionInfo(), requestMsg, getPubAckCallback(channel, deviceName, msgId, requestMsg)); + private void processGetAttributeRequestMessage(T deviceCtx, TransportProtos.GetAttributeRequestMsg requestMsg, + String deviceName, int msgId, AtomicInteger remaining, AtomicBoolean ackSent) { + transportService.process(deviceCtx.getSessionInfo(), requestMsg, + getAggregatePubAckCallback(channel, msgId, deviceName, requestMsg, remaining, ackSent)); } private TransportProtos.GetAttributeRequestMsg toGetAttributeRequestMsg(int requestId, boolean clientScope, Set keys) { @@ -718,9 +790,11 @@ public abstract class AbstractGatewaySessionHandler pubAckCallback = getAggregatePubAckCallback(channel, -1, deviceName, postTelemetryMsg, + new AtomicInteger(1), new AtomicBoolean(false)); + transportService.process(sessionInfo, postTelemetryMsg, pubAckCallback); } - public ConcurrentMap getDevices () { + public ConcurrentMap getDevices() { return this.devices; } - private TransportServiceCallback getPubAckCallback(final ChannelHandlerContext ctx, final String deviceName, final int msgId, final T msg) { + protected TransportServiceCallback getAggregatePubAckCallback( + final ChannelHandlerContext ctx, + final int msgId, + final String deviceName, + final T msg, + final AtomicInteger remaining, + final AtomicBoolean ackSent) { + return new TransportServiceCallback() { @Override public void onSuccess(Void dummy) { log.trace("[{}][{}][{}][{}] Published msg: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, deviceName, msg); - if (msgId > 0) { - ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(deviceSessionCtx, msgId, MqttReasonCodes.PubAck.SUCCESS.byteValue())); - } else { - log.trace("[{}][{}][{}] Wrong msg id: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, msg); - ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg(deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue())); + if (remaining.decrementAndGet() == 0 && ackSent.compareAndSet(false, true)) { + if (msgId > 0) { + ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg( + deviceSessionCtx, msgId, MqttReasonCodes.PubAck.SUCCESS.byteValue())); + } else { + log.trace("[{}][{}][{}] Wrong msg id: [{}]", gateway.getTenantId(), gateway.getDeviceId(), sessionId, msgId); + ctx.writeAndFlush(MqttTransportHandler.createMqttPubAckMsg( + deviceSessionCtx, msgId, MqttReasonCodes.PubAck.UNSPECIFIED_ERROR.byteValue())); + } + } + if (msgId <= 0) { closeDeviceSession(deviceName, MqttReasonCodes.Disconnect.MALFORMED_PACKET); } } @@ -767,11 +856,20 @@ public abstract class AbstractGatewaySessionHandler contextListenableFuture, int msgId, List postTelemetryMsgList, String deviceName) { + if (CollectionUtils.isEmpty(postTelemetryMsgList)) { + log.debug("[{}] Device telemetry list is empty for: [{}]", sessionId, gateway.getDeviceId()); + } + + AtomicInteger remaining = new AtomicInteger(postTelemetryMsgList.size()); + AtomicBoolean ackSent = new AtomicBoolean(false); + process(contextListenableFuture, deviceCtx -> { for (TransportProtos.PostTelemetryMsg telemetryMsg : postTelemetryMsgList) { try { - processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, msgId); + processPostTelemetryMsg(deviceCtx, telemetryMsg, deviceName, msgId, remaining, ackSent); } catch (Throwable e) { log.warn("[{}][{}] Failed to convert telemetry: {}", gateway.getDeviceId(), deviceName, telemetryMsg, e); - ackOrClose(msgId); + ackOrClose(msgId, ackSent); } } }, - t -> log.debug("[{}] Failed to process device telemetry command: {}", sessionId, deviceName, t)); + t -> processFailure(msgId, deviceName, "Failed to process device telemetry command", ackSent, t)); } private void onDeviceAttributesProto(ListenableFuture contextListenableFuture, int msgId, List attributesMsgList, String deviceName) throws AdaptorException { try { if (CollectionUtils.isEmpty(attributesMsgList)) { - log.debug("[{}] Devices attributes keys list is empty for: [{}]", sessionId, gateway.getDeviceId()); + log.debug("[{}] Device attribute list is empty for: [{}]", sessionId, gateway.getDeviceId()); } + + AtomicInteger remaining = new AtomicInteger(attributesMsgList.size()); + AtomicBoolean ackSent = new AtomicBoolean(false); + process(contextListenableFuture, deviceCtx -> { for (TransportApiProtos.AttributesMsg attributesMsg : attributesMsgList) { TransportProtos.PostAttributeMsg kvListProto = attributesMsg.getMsg(); try { TransportProtos.PostAttributeMsg postAttributeMsg = ProtoConverter.validatePostAttributeMsg(kvListProto); - processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId); + processPostAttributesMsg(deviceCtx, postAttributeMsg, deviceName, msgId, remaining, ackSent); } catch (Throwable e) { log.warn("[{}][{}] Failed to process device attributes command: {}", gateway.getDeviceId(), deviceName, kvListProto, e); + ackOrClose(msgId, ackSent); } } }, - t -> log.debug("[{}] Failed to process device attributes command: {}", sessionId, deviceName, t)); + t -> processFailure(msgId, deviceName, "Failed to process device attributes command", ackSent, t)); } catch (RuntimeException e) { throw new AdaptorException(e); } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java index 920f00ed27..e703a7b256 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/TbContext.java @@ -255,7 +255,7 @@ public interface TbContext { void checkTenantEntity(EntityId entityId) throws TbNodeException; - & HasTenantId, I extends EntityId> void checkTenantEntity(E entity) throws TbNodeException; + & HasTenantId, I extends EntityId> void checkTenantOrSystemEntity(E entity) throws TbNodeException; boolean isLocalEntity(EntityId entityId); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/ai/TbAiNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/ai/TbAiNode.java index 054b06fb54..654ed3f448 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/ai/TbAiNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/ai/TbAiNode.java @@ -261,7 +261,7 @@ public final class TbAiNode extends TbAbstractExternalNode implements TbNode { if (!ResourceType.GENERAL.equals(resource.getResourceType())) { throw new TbNodeException("[" + ctx.getTenantId() + "] Resource with ID: [" + tbResourceId + "] has unsupported resource type: " + resource.getResourceType(), true); } - ctx.checkTenantEntity(resource); + ctx.checkTenantOrSystemEntity(resource); } private ListenableFuture> loadResources(TbContext ctx) {