diff --git a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java index 12b2b677d4..a0423804b0 100644 --- a/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/coap/client/CoapClientIntegrationTest.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import lombok.Getter; import lombok.extern.slf4j.Slf4j; +import org.awaitility.Awaitility; import org.eclipse.californium.core.CoapHandler; import org.eclipse.californium.core.CoapObserveRelation; import org.eclipse.californium.core.CoapResponse; @@ -27,13 +28,23 @@ import org.eclipse.californium.core.coap.MediaTypeRegistry; import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.User; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.query.AliasEntityId; import org.thingsboard.server.common.data.query.EntityKey; import org.thingsboard.server.common.data.query.EntityKeyType; import org.thingsboard.server.common.data.query.SingleEntityFilter; +import org.thingsboard.server.common.data.security.Authority; +import org.thingsboard.server.common.data.security.DeviceCredentials; import org.thingsboard.server.common.msg.session.FeatureType; +import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.service.DefaultTransportService; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.coap.AbstractCoapIntegrationTest; import org.thingsboard.server.transport.coap.CoapTestCallback; @@ -56,6 +67,10 @@ import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.common.data.query.EntityKeyType.CLIENT_ATTRIBUTE; import static org.thingsboard.server.common.data.query.EntityKeyType.SHARED_ATTRIBUTE; @@ -67,6 +82,11 @@ public class CoapClientIntegrationTest extends AbstractCoapIntegrationTest { private static final List EXPECTED_KEYS = Arrays.asList("key1", "key2", "key3", "key4", "key5"); private static final String DEVICE_RESPONSE = "{\"value1\":\"A\",\"value2\":\"B\"}"; + @Autowired + DefaultTransportService defaultTransportService; + @MockitoSpyBean + TransportRateLimitService rateLimitService; + @Before public void beforeTest() throws Exception { CoapTestConfigProperties configProperties = CoapTestConfigProperties.builder() @@ -304,4 +324,44 @@ public class CoapClientIntegrationTest extends AbstractCoapIntegrationTest { private List getEntityKeys(EntityKeyType scope) { return CoapClientIntegrationTest.EXPECTED_KEYS.stream().map(key -> new EntityKey(scope, key)).collect(Collectors.toList()); } + + @Test + public void testCancelAttributeSubscriptionAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default"); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + client = new CoapTestClient(accessToken, FeatureType.ATTRIBUTES); + CoapTestCallback callbackCoap = new CoapTestCallback(); + + client.getObserveRelation(callbackCoap); + String awaitAlias = "await subscribe to attributes updates"; + await(awaitAlias) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> CoAP.ResponseCode.CONTENT.equals(callbackCoap.getResponseCode()) && + callbackCoap.getObserve() != null && 0 == callbackCoap.getObserve()); + + loginSysAdmin(); + Mockito.clearInvocations(rateLimitService); + deleteTenant(savedTenant.getId()); + + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); + } } diff --git a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java index 0430e10bfe..b6b86e9821 100644 --- a/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java +++ b/application/src/test/java/org/thingsboard/server/transport/mqtt/mqttv3/client/MqttClientConnectionTest.java @@ -15,14 +15,45 @@ */ package org.thingsboard.server.transport.mqtt.mqttv3.client; +import com.google.gson.JsonObject; +import io.netty.buffer.Unpooled; +import io.netty.handler.codec.mqtt.MqttQoS; +import org.awaitility.Awaitility; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.mockito.Mockito; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.bean.override.mockito.MockitoSpyBean; +import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.User; +import org.thingsboard.server.common.data.security.Authority; +import org.thingsboard.server.common.data.security.DeviceCredentials; +import org.thingsboard.server.common.transport.limits.TransportRateLimitService; +import org.thingsboard.server.common.transport.service.DefaultTransportService; +import org.thingsboard.server.dao.device.DeviceService; import org.thingsboard.server.dao.service.DaoSqlTest; import org.thingsboard.server.transport.mqtt.MqttTestConfigProperties; +import org.thingsboard.server.transport.mqtt.mqttv3.MqttTestClient; + +import java.util.concurrent.TimeUnit; + +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; @DaoSqlTest public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { + @Autowired + DefaultTransportService defaultTransportService; + @Autowired + DeviceService deviceService; + @MockitoSpyBean + TransportRateLimitService rateLimitService; + @Before public void beforeTest() throws Exception { MqttTestConfigProperties configProperties = MqttTestConfigProperties.builder() @@ -45,4 +76,95 @@ public class MqttClientConnectionTest extends AbstractMqttClientConnectionTest { public void testClientWithWrongClientIdAndEmptyUsernamePassword() throws Exception { processClientWithWrongClientIdAndEmptyUsernamePasswordTest(); } + + @Test + public void testClientShouldBeDisconnectedAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default", false); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + Assert.assertTrue(client.isConnected()); + + loginSysAdmin(); + + Mockito.clearInvocations(rateLimitService); + deleteTenant(savedTenant.getId()); + + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> !client.isConnected()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); + } + + @Test + public void testGatewayShouldBeDisconnectedAfterTenantDeletion() throws Exception { + loginSysAdmin(); + Tenant tenant = new Tenant(); + tenant.setTitle("Mqtt Client Connection Test Tenant"); + Tenant savedTenant = saveTenant(tenant); + + User tenantAdmin = new User(); + tenantAdmin.setAuthority(Authority.TENANT_ADMIN); + tenantAdmin.setTenantId(savedTenant.getTenantId()); + tenantAdmin.setEmail("mqttTestClient@thingsboard.org"); + createUserAndLogin(tenantAdmin, TENANT_ADMIN_PASSWORD); + + savedDevice = createDevice(RandomStringUtils.randomAlphabetic(10), "default", true); + DeviceCredentials deviceCredentials = + doGet("/api/device/" + savedDevice.getId().getId().toString() + "/credentials", DeviceCredentials.class); + accessToken = deviceCredentials.getCredentialsId(); + + MqttTestClient client = new MqttTestClient(); + client.connectAndWait(accessToken); + + String deviceName = "gateway_device_" + RandomStringUtils.randomAlphabetic(5); + client.publish("v1/gateway/connect", createGatewayConnectPayload(deviceName).toString().getBytes()); + + Awaitility.await("awaiting till device is created") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> deviceService.findDeviceByTenantIdAndName(savedTenant.getTenantId(), deviceName) != null); + + loginSysAdmin(); + + Mockito.clearInvocations(rateLimitService); + deleteTenant(savedTenant.getId()); + + Awaitility.await("awaiting defaultTransportService.sessions is empty") + .pollInterval(10, TimeUnit.MILLISECONDS) + .atMost(DEFAULT_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .until(() -> defaultTransportService.sessions.isEmpty()); + + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> !client.isConnected()); + + verify(rateLimitService, never()).checkLimits(Mockito.any(), Mockito.any(), Mockito.any(), anyInt(), anyBoolean()); + } + + protected JsonObject createGatewayConnectPayload(String deviceName){ + JsonObject payload = new JsonObject(); + payload.addProperty("device", deviceName); + return payload; + } } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java index bf23ea6766..d668d20578 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java @@ -57,6 +57,11 @@ public abstract class AbstractSyncSessionCallback implements SessionMsgListener } + @Override + public void onTenantDeleted(DeviceId deviceId) { + + } + @Override public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg toDeviceRequest) { logUnsupportedCommandMessage(toDeviceRequest); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java index aa61992a6f..88ee511612 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java @@ -51,11 +51,13 @@ import org.thingsboard.server.common.transport.DeviceDeletedEvent; import org.thingsboard.server.common.transport.DeviceProfileUpdatedEvent; import org.thingsboard.server.common.transport.DeviceUpdatedEvent; import org.thingsboard.server.common.transport.SessionMsgListener; +import org.thingsboard.server.common.transport.TenantDeletedEvent; import org.thingsboard.server.common.transport.TransportDeviceProfileCache; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.SessionInfoCreator; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.transport.coap.CoapSessionMsgType; @@ -142,6 +144,11 @@ public class DefaultCoapClientContext implements CoapClientContext { clients.remove(event.getDeviceId()); } + @EventListener(TenantDeletedEvent.class) + public void onApplicationEvent(TenantDeletedEvent event) { + clients.values().removeIf(state -> event.getTenantId().equals(state.getTenantId())); + } + @Override public boolean registerAttributeObservation(TbCoapClientState clientState, String token, CoapExchange exchange) { return registerFeatureObservation(clientState, token, exchange, FeatureType.ATTRIBUTES); @@ -562,6 +569,12 @@ public class DefaultCoapClientContext implements CoapClientContext { cancelAttributeSubscription(state); } + @Override + public void onTenantDeleted(DeviceId deviceId) { + state.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); + onDeviceDeleted(deviceId); + } + @Override public void onRemoteSessionCloseCommand(UUID sessionId, TransportProtos.SessionCloseNotificationProto sessionCloseNotification) { log.trace("[{}] Received the remote command to close the session: {}", sessionId, sessionCloseNotification.getMessage()); @@ -825,9 +838,11 @@ public class DefaultCoapClientContext implements CoapClientContext { clientsByToken.remove(state.getRpc().getToken()); CoapExchange exchange = state.getRpc().getExchange(); state.setRpc(null); - transportService.process(state.getSession(), - TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), - new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + if (state.shouldNotifyCore()) { + transportService.process(state.getSession(), + TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), + new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + } if (state.getAttrs() == null) { closeAndCleanup(state); } @@ -839,9 +854,11 @@ public class DefaultCoapClientContext implements CoapClientContext { clientsByToken.remove(state.getAttrs().getToken()); CoapExchange exchange = state.getAttrs().getExchange(); state.setAttrs(null); - transportService.process(state.getSession(), - TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), - new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + if (state.shouldNotifyCore()) { + transportService.process(state.getSession(), + TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), + new CoapResponseCodeCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); + } if (state.getRpc() == null) { closeAndCleanup(state); } @@ -849,7 +866,9 @@ public class DefaultCoapClientContext implements CoapClientContext { } private void closeAndCleanup(TbCoapClientState state) { - transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); + if (state.shouldNotifyCore()) { + transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); + } transportService.deregisterSession(state.getSession()); state.setSession(null); state.setConfiguration(null); diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java index ab2ab7c592..1f922b6d54 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/TbCoapClientState.java @@ -24,7 +24,9 @@ import org.thingsboard.server.common.data.device.data.CoapDeviceTransportConfigu import org.thingsboard.server.common.data.device.data.PowerMode; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceProfileId; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.transport.coap.TransportConfigurationContainer; import org.thingsboard.server.transport.coap.adaptors.CoapTransportAdaptor; @@ -33,6 +35,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.Future; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; @@ -72,6 +75,9 @@ public class TbCoapClientState { @Getter @Setter private Future sleepTask; + @Getter + @Setter + private SessionCloseReason sessionCloseReason; private boolean firstEdrxDownlink = true; @@ -149,4 +155,18 @@ public class TbCoapClientState { this.missedAttributeUpdates = null; return result; } + + public boolean shouldNotifyCore() { + return SessionCloseReason.shouldNotifyCore(sessionCloseReason); + } + + public TenantId getTenantId() { + if (credentials != null && credentials.getDeviceInfo() != null) { + return credentials.getDeviceInfo().getTenantId(); + } + if (session != null) { + return TenantId.fromUUID(new UUID(session.getTenantIdMSB(), session.getTenantIdLSB())); + } + return null; + } } diff --git a/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java new file mode 100644 index 0000000000..e28d880a87 --- /dev/null +++ b/common/transport/coap/src/test/java/org/thingsboard/server/transport/coap/CoapSessionListenerTest.java @@ -0,0 +1,82 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.transport.coap; + +import org.eclipse.californium.core.server.resources.CoapExchange; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.thingsboard.server.coapserver.CoapServerContext; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.transport.TransportDeviceProfileCache; +import org.thingsboard.server.common.transport.TransportService; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.transport.coap.client.DefaultCoapClientContext; +import org.thingsboard.server.transport.coap.client.TbCoapClientState; +import org.thingsboard.server.transport.coap.client.TbCoapObservationState; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +public class CoapSessionListenerTest { + + @Mock + CoapServerContext config; + @Mock + CoapTransportContext transportContext; + @Mock + TransportService transportService; + @Mock + TransportDeviceProfileCache profileCache; + @Mock + PartitionService partitionService; + + private DefaultCoapClientContext coapClientContext; + + @BeforeEach + void setUp() { + coapClientContext = new DefaultCoapClientContext(config, transportContext, transportService, profileCache, partitionService); + } + + @Test + void onTenantDeleted_shouldNotInteractWithCore() { + // GIVEN + TbCoapClientState state = mock(TbCoapClientState.class); + + TbCoapObservationState rpcObs = mock(TbCoapObservationState.class); + when(rpcObs.getToken()).thenReturn("rpc-token"); + when(rpcObs.getExchange()).thenReturn(mock(CoapExchange.class)); + + TbCoapObservationState attrsObs = mock(TbCoapObservationState.class); + when(attrsObs.getToken()).thenReturn("attr-token"); + when(attrsObs.getExchange()).thenReturn(mock(CoapExchange.class)); + when(state.getRpc()).thenReturn(rpcObs); + when(state.getAttrs()).thenReturn(attrsObs); + + DefaultCoapClientContext.CoapSessionListener listener = coapClientContext.new CoapSessionListener(state); + + // WHEN + listener.onTenantDeleted(mock(DeviceId.class)); + + // THEN + verifyNoInteractions(transportService, partitionService); + } + +} diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index d0205314bc..a51e37ad9d 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -614,6 +614,10 @@ public class DeviceApiController implements TbTransportService { responseWriter.setResult(new ResponseEntity<>("Device was deleted!", HttpStatus.FORBIDDEN)); } + @Override + public void onTenantDeleted(DeviceId deviceId) { + onDeviceDeleted(deviceId); + } } private static MediaType parseMediaType(String contentType) { diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java index 6f250a1b87..f427ee9c37 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2mSessionMsgListener.java @@ -114,4 +114,10 @@ public class LwM2mSessionMsgListener implements GenericFutureListener supportedClientObjects; + @Getter + @Setter + private SessionCloseReason sessionCloseReason; + public Object clone() throws CloneNotSupportedException { return super.clone(); } @@ -445,6 +451,10 @@ public class LwM2mClient { return this.supportedClientObjects != null ? this.supportedClientObjects.get(objectid) : null; } + public boolean shouldNotifyCore() { + return SessionCloseReason.shouldNotifyCore(sessionCloseReason); + } + private void setSupportedClientObjects(){ if (this.registration.getSupportedObject() != null && this.registration.getSupportedObject().size() > 0) { this.supportedClientObjects = this.registration.getSupportedObject(); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java index c6ba2cd0c0..b7409073cf 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/DefaultLwM2MSessionManager.java @@ -65,7 +65,14 @@ public class DefaultLwM2MSessionManager implements LwM2MSessionManager { @Override public void deregister(TransportProtos.SessionInfoProto sessionInfo) { - transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); + deregister(sessionInfo, true); + } + + @Override + public void deregister(TransportProtos.SessionInfoProto sessionInfo, boolean notifyCore) { + if (notifyCore) { + transportService.process(sessionInfo, SESSION_EVENT_MSG_CLOSED, null); + } transportService.deregisterSession(sessionInfo); } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java index df6ff0c0b3..a077a4276c 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/session/LwM2MSessionManager.java @@ -23,5 +23,6 @@ public interface LwM2MSessionManager { void deregister(TransportProtos.SessionInfoProto sessionInfo); + void deregister(TransportProtos.SessionInfoProto sessionInfo, boolean notifyCore); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java index cb922e5905..72d5bae5bc 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2mUplinkMsgHandler.java @@ -65,6 +65,7 @@ import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.ota.OtaPackageUtil; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; import org.thingsboard.server.queue.util.TbLwM2mTransportComponent; @@ -289,11 +290,13 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl private void doUnReg(Registration registration, LwM2mClient client) { try { - logService.log(client, LOG_LWM2M_INFO + ": Client unRegistration"); + if (client.shouldNotifyCore()) { + logService.log(client, LOG_LWM2M_INFO + ": Client unRegistration"); + } clientContext.unregister(client, registration); SessionInfoProto sessionInfo = client.getSession(); if (sessionInfo != null) { - sessionManager.deregister(sessionInfo); + sessionManager.deregister(sessionInfo, client.shouldNotifyCore()); sessionStore.remove(registration.getEndpoint()); log.info("Client close session: [{}] unReg [{}] name [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType()); } else { @@ -447,6 +450,15 @@ public class DefaultLwM2mUplinkMsgHandler extends LwM2MExecutorAwareService impl clearAndUnregister(clientContext.getClientByDeviceId(deviceId.getId())); } + @Override + public void onTenantDelete(DeviceId deviceId) { + LwM2mClient client = clientContext.getClientByDeviceId(deviceId.getId()); + if (client != null) { + client.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); + clearAndUnregister(client); + } + } + @Override public void onResourceUpdate(TransportProtos.ResourceUpdateMsg resourceUpdateMsgOpt) { String idVer = resourceUpdateMsgOpt.getResourceKey(); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java index 985bdf4b89..99cf62f093 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/LwM2mUplinkMsgHandler.java @@ -59,6 +59,8 @@ public interface LwM2mUplinkMsgHandler { void onDeviceDelete(DeviceId deviceId); + void onTenantDelete(DeviceId deviceId); + void onResourceUpdate(TransportProtos.ResourceUpdateMsg resourceUpdateMsgOpt); void onResourceDelete(TransportProtos.ResourceDeleteMsg resourceDeleteMsgOpt); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index 48fbad3032..27a817843e 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -73,6 +73,7 @@ import org.thingsboard.server.common.transport.auth.SessionInfoCreator; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse; import org.thingsboard.server.common.transport.service.SessionMetaData; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.common.transport.util.SslUtil; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceResponseMsg; @@ -1327,15 +1328,20 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement public void doDisconnect() { if (deviceSessionCtx.isConnected()) { log.debug("[{}] Client disconnected!", sessionId); - transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); + boolean notifyCore = deviceSessionCtx.shouldNotifyCore(); + if (notifyCore) { + transportService.process(deviceSessionCtx.getSessionInfo(), SESSION_EVENT_MSG_CLOSED, null); + } transportService.deregisterSession(deviceSessionCtx.getSessionInfo()); if (gatewaySessionHandler != null) { gatewaySessionHandler.onDevicesDisconnect(); } if (sparkplugSessionHandler != null) { - // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() - sparkplugSessionHandler.sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), - deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime()); + if (notifyCore) { + // add Msg Telemetry node: key STATE type: String value: OFFLINE ts: sparkplugBProto.getTimestamp() + sparkplugSessionHandler.sendSparkplugStateOnTelemetry(deviceSessionCtx.getSessionInfo(), + deviceSessionCtx.getDeviceInfo().getDeviceName(), OFFLINE, new Date().getTime()); + } sparkplugSessionHandler.onDevicesDisconnect(); } deviceSessionCtx.setDisconnected(); @@ -1611,7 +1617,9 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement @Override public void onDeviceDeleted(DeviceId deviceId) { - context.onAuthFailure(address); + if (deviceSessionCtx.shouldNotifyCore()) { + context.onAuthFailure(address); + } ChannelHandlerContext ctx = deviceSessionCtx.getChannel(); closeCtx(ctx, MqttReasonCodes.Disconnect.ADMINISTRATIVE_ACTION); if (gatewaySessionHandler != null) { @@ -1619,6 +1627,12 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement } } + @Override + public void onTenantDeleted(DeviceId deviceId) { + deviceSessionCtx.setSessionCloseReason(SessionCloseReason.TENANT_DELETED); + onDeviceDeleted(deviceId); + } + public void sendErrorRpcResponse(TransportProtos.SessionInfoProto sessionInfo, int requestId, ThingsboardErrorCode result, String errorMsg) { String payload = JacksonUtil.toString(SparkplugRpcResponseBody.builder().result(result.name()).error(errorMsg).build()); TransportProtos.ToDeviceRpcResponseMsg msg = TransportProtos.ToDeviceRpcResponseMsg.newBuilder().setRequestId(requestId).setError(payload).build(); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java index a1162f1990..652b77b8a8 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/AbstractGatewayDeviceSessionContext.java @@ -27,6 +27,7 @@ import org.thingsboard.server.common.transport.SessionMsgListener; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; import org.thingsboard.server.common.transport.auth.TransportDeviceInfo; +import org.thingsboard.server.common.transport.session.SessionCloseReason; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; @@ -145,6 +146,12 @@ public abstract class AbstractGatewayDeviceSessionContext 0; } 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 99f10eb176..52f826b012 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 @@ -828,12 +828,14 @@ public abstract class AbstractGatewaySessionHandler { + TenantId sessionTenantId = getTenantId(md.getSessionInfo()); + if (sessionTenantId.equals(tenantId)) { + DeviceId sessionDeviceId = getDeviceId(md.getSessionInfo()); + transportCallbackExecutor.submit(() -> { + md.getListener().onTenantDeleted(sessionDeviceId); + }); + } + }); + // notify transports holding per-device state (e.g. CoAP's client states) to clean it up in one shot, + // since no per-device delete messages are sent during tenant deletion + eventPublisher.publishEvent(new TenantDeletedEvent(tenantId)); + } public void onProfileUpdate(DeviceProfile deviceProfile) { long deviceProfileIdMSB = deviceProfile.getId().getId().getMostSignificantBits(); @@ -1108,6 +1125,12 @@ public class DefaultTransportService extends TransportActivityManager implements TopicPartitionInfo tpi; try { tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId); + } catch (TenantNotFoundException e) { + log.warn("Ignoring message delivery because Tenant with ID [{}] not found, routingKey [{}], msg [{}].", tenantId, routingKey, msg); + if (callback != null) { + callback.onError(e); + } + return; } catch (Exception e) { log.warn("Failed to send message to core. Tenant with ID [{}], routingKey [{}], msg [{}]. Message delivery aborted.", tenantId, routingKey, msg, e); if (callback != null) { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java index 535baf921d..8a2ac8a88d 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/DeviceAwareSessionContext.java @@ -47,6 +47,9 @@ public abstract class DeviceAwareSessionContext implements SessionContext { @Getter @Setter protected volatile TransportProtos.SessionInfoProto sessionInfo; + @Getter + @Setter + protected volatile SessionCloseReason sessionCloseReason; @Setter private volatile boolean connected; @@ -84,6 +87,10 @@ public abstract class DeviceAwareSessionContext implements SessionContext { this.connected = false; } + public boolean shouldNotifyCore() { + return SessionCloseReason.shouldNotifyCore(sessionCloseReason); + } + public boolean isSparkplug() { DeviceProfileTransportConfiguration transportConfiguration = this.deviceProfile.getProfileData().getTransportConfiguration(); if (transportConfiguration instanceof MqttDeviceProfileTransportConfiguration) { diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java new file mode 100644 index 0000000000..cd729f58ac --- /dev/null +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/session/SessionCloseReason.java @@ -0,0 +1,26 @@ +/** + * Copyright © 2016-2026 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.thingsboard.server.common.transport.session; + +public enum SessionCloseReason { + + TENANT_DELETED; + + public static boolean shouldNotifyCore(SessionCloseReason reason) { + return reason != TENANT_DELETED; + } + +}