Merge pull request #5945 from zbeacon/fix/mqtt-doesnt-close-session-for-removed-device

[3.3.3]Fix for PROD-1337 (Device session doesn't close on device removal)

common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/callback/AbstractSyncSessionCallback.java

@@ -21,6 +21,7 @@ import org.eclipse.californium.core.coap.MediaTypeRegistry;
 import org.eclipse.californium.core.coap.Request;
 import org.eclipse.californium.core.coap.Response;
 import org.eclipse.californium.core.server.resources.CoapExchange;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.transport.coap.client.TbCoapClientState;
@@ -52,6 +53,11 @@ public abstract class AbstractSyncSessionCallback implements SessionMsgListener
 
     }
 
+    @Override
+    public void onDeviceDeleted(DeviceId deviceId) {
+
+    }
+
     @Override
     public void onToDeviceRpcRequest(UUID sessionId, TransportProtos.ToDeviceRpcRequestMsg toDeviceRequest) {
         logUnsupportedCommandMessage(toDeviceRequest);

common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java

@@ -603,6 +603,13 @@ public class DeviceApiController implements TbTransportService {
             responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg).toString(), HttpStatus.OK));
         }
 
+        @Override
+        public void onDeviceDeleted(DeviceId deviceId) {
+            UUID sessionId = new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB());
+            log.trace("[{}] Received device deleted notification for device with id: {}",sessionId, deviceId);
+            responseWriter.setResult(new ResponseEntity<>("Device was deleted!", HttpStatus.FORBIDDEN));
+        }
+
     }
 
     private static MediaType parseMediaType(String contentType) {

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

@@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.DeviceProfile;
 import org.thingsboard.server.common.data.DeviceTransportType;
 import org.thingsboard.server.common.data.TransportPayloadType;
 import org.thingsboard.server.common.data.device.profile.MqttTopics;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.data.ota.OtaPackageType;
 import org.thingsboard.server.common.data.rpc.RpcStatus;
@@ -1066,4 +1067,11 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
         deviceSessionCtx.onDeviceUpdate(sessionInfo, device, deviceProfileOpt);
     }
 
+    @Override
+    public void onDeviceDeleted(DeviceId deviceId) {
+        context.onAuthFailure(address);
+        ChannelHandlerContext ctx = deviceSessionCtx.getChannel();
+        ctx.close();
+    }
+
 }

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java

@@ -19,6 +19,7 @@ import io.netty.channel.ChannelFuture;
 import io.netty.handler.codec.mqtt.MqttMessage;
 import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.DeviceProfile;
+import org.thingsboard.server.common.data.id.DeviceId;
 import org.thingsboard.server.common.data.rpc.RpcStatus;
 import org.thingsboard.server.common.transport.SessionMsgListener;
 import org.thingsboard.server.common.transport.TransportService;
@@ -136,6 +137,11 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
         // This feature is not supported in the TB IoT Gateway yet.
     }
 
+    @Override
+    public void onDeviceDeleted(DeviceId deviceId) {
+        parent.onDeviceDeleted(this.getSessionInfo().getDeviceName());
+    }
+
     private boolean isAckExpected(MqttMessage message) {
         return message.fixedHeader().qosLevel().value() > 0;
     }

common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewaySessionHandler.java

@@ -178,6 +178,10 @@ public class GatewaySessionHandler {
         devices.forEach(this::deregisterSession);
     }
 
+    public void onDeviceDeleted(String deviceName) {
+        deregisterSession(deviceName);
+    }
+
     public String getNodeId() {
         return context.getNodeId();
     }

common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/SessionMsgListener.java

@@ -45,6 +45,8 @@ public interface SessionMsgListener {
 
     void onToServerRpcResponse(ToServerRpcResponseMsg toServerResponse);
 
+    void onDeviceDeleted(DeviceId deviceId);
+
     default void onUplinkNotification(UplinkNotificationMsg notificationMsg){};
 
     default void onToTransportUpdateCredentials(ToTransportUpdateCredentialsProto toTransportUpdateCredentials){}
@@ -54,8 +56,6 @@ public interface SessionMsgListener {
     default void onDeviceUpdate(TransportProtos.SessionInfoProto sessionInfo, Device device,
                                 Optional<DeviceProfile> deviceProfileOpt) {}
 
-    default void onDeviceDeleted(DeviceId deviceId) {}
-
     default void onResourceUpdate(TransportProtos.ResourceUpdateMsg resourceUpdateMsgOpt) {}
 
     default void onResourceDelete(TransportProtos.ResourceDeleteMsg resourceUpdateMsgOpt) {}

msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttClientTest.java

@@ -339,6 +339,21 @@ public class MqttClientTest extends AbstractContainerTest {
         restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + device.getId());
     }
 
+    @Test
+    public void deviceDeletedClosingSession() throws Exception {
+        restClient.login("tenant@thingsboard.org", "tenant");
+        String deviceForDeletingTestName = "Device for deleting notification test";
+        Device device = createDevice(deviceForDeletingTestName);
+        DeviceCredentials deviceCredentials = restClient.getDeviceCredentialsByDeviceId(device.getId()).get();
+
+        MqttMessageListener listener = new MqttMessageListener();
+        MqttClient mqttClient = getMqttClient(deviceCredentials, listener);
+
+        restClient.deleteDevice(device.getId());
+        TimeUnit.SECONDS.sleep(3);
+        Assert.assertFalse(mqttClient.isConnected());
+    }
+
     private RuleChainId createRootRuleChainForRpcResponse() throws Exception {
         RuleChain newRuleChain = new RuleChain();
         newRuleChain.setName("testRuleChain");

msa/black-box-tests/src/test/java/org/thingsboard/server/msa/connectivity/MqttGatewayClientTest.java

@@ -374,7 +374,15 @@ public class MqttGatewayClientTest extends AbstractContainerTest {
         Assert.assertEquals(clientResponse.toString(), serverResponse.getBody());
     }
 
-        private void checkAttribute(boolean client, String expectedValue) throws Exception{
+    @Test
+    public void deviceCreationAfterDeleted() throws Exception {
+        restClient.getRestTemplate().delete(HTTPS_URL + "/api/device/" + this.createdDevice.getId());
+        Optional<Device> deletedDevice = restClient.getDeviceById(this.createdDevice.getId());
+        Assert.assertTrue(deletedDevice.isEmpty());
+        this.createdDevice = createDeviceThroughGateway(mqttClient, gatewayDevice);
+    }
+
+    private void checkAttribute(boolean client, String expectedValue) throws Exception {
         JsonObject gatewayAttributesRequest = new JsonObject();
         int messageId = new Random().nextInt(100);
         gatewayAttributesRequest.addProperty("id", messageId);