From ee74bbed218e6a106f672c6804b1dc65e4f51ffc Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 3 Aug 2021 14:18:46 +0300 Subject: [PATCH 1/2] fixed race condition during unreq and updating lwm2m client --- .../lwm2m/server/client/LwM2mClientContextImpl.java | 8 ++++++-- .../lwm2m/server/ota/LwM2MClientOtaInfo.java | 1 + .../lwm2m/server/store/TbRedisLwM2MClientStore.java | 13 ++++++++++--- 3 files changed, 17 insertions(+), 5 deletions(-) diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java index 93aaea43ed..e29dcdd1dd 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java @@ -166,7 +166,7 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { this.lwM2mClientsByRegistrationId.put(registration.getId(), client); client.setState(LwM2MClientState.REGISTERED); onUplink(client); - if(!compareAndSetSleepFlag(client, false)){ + if (!compareAndSetSleepFlag(client, false)) { clientStore.put(client); } } finally { @@ -316,7 +316,11 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { public void update(LwM2mClient client) { client.lock(); try { - clientStore.put(client); + if (client.getState().equals(LwM2MClientState.REGISTERED)) { + clientStore.put(client); + } else { + log.error("[{}] Client is in invalid state: {}!", client.getEndpoint(), client.getState()); + } } finally { client.unlock(); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/ota/LwM2MClientOtaInfo.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/ota/LwM2MClientOtaInfo.java index 7261e5bd40..039629516f 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/ota/LwM2MClientOtaInfo.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/ota/LwM2MClientOtaInfo.java @@ -106,6 +106,7 @@ public abstract class LwM2MClientOtaInfo { public abstract OtaPackageType getType(); + @JsonIgnore public String getTargetPackageId() { return getPackageId(targetName, targetVersion); } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java index d735eed26e..2a111a9f81 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/store/TbRedisLwM2MClientStore.java @@ -15,11 +15,13 @@ */ package org.thingsboard.server.transport.lwm2m.server.store; +import lombok.extern.slf4j.Slf4j; import org.nustaq.serialization.FSTConfiguration; import org.springframework.data.redis.connection.RedisClusterConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.Cursor; import org.springframework.data.redis.core.ScanOptions; +import org.thingsboard.server.transport.lwm2m.server.client.LwM2MClientState; import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient; import java.util.ArrayList; @@ -27,6 +29,7 @@ import java.util.HashSet; import java.util.List; import java.util.Set; +@Slf4j public class TbRedisLwM2MClientStore implements TbLwM2MClientStore { private static final String CLIENT_EP = "CLIENT#EP#"; @@ -76,9 +79,13 @@ public class TbRedisLwM2MClientStore implements TbLwM2MClientStore { @Override public void put(LwM2mClient client) { - byte[] clientSerialized = serializer.asByteArray(client); - try (var connection = connectionFactory.getConnection()) { - connection.getSet(getKey(client.getEndpoint()), clientSerialized); + if (client.getState().equals(LwM2MClientState.UNREGISTERED)) { + log.error("[{}] Client is in invalid state: {}!", client.getEndpoint(), client.getState(), new Exception()); + } else { + byte[] clientSerialized = serializer.asByteArray(client); + try (var connection = connectionFactory.getConnection()) { + connection.getSet(getKey(client.getEndpoint()), clientSerialized); + } } } From 1539e4550ddfa11497c077ccd38660dd0f4f8a54 Mon Sep 17 00:00:00 2001 From: YevhenBondarenko Date: Tue, 3 Aug 2021 16:22:33 +0300 Subject: [PATCH 2/2] fixed tests --- .../profile/TbDeviceProfileNodeTest.java | 30 +++++++++---------- 1 file changed, 15 insertions(+), 15 deletions(-) diff --git a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java index c3bbc8bcf6..b2c79102c4 100644 --- a/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java +++ b/rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/profile/TbDeviceProfileNodeTest.java @@ -196,7 +196,7 @@ public class TbDeviceProfileNodeTest { TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); TbMsg theMsg2 = TbMsg.newMsg("ALARM", deviceId, new TbMsgMetaData(), "2"); @@ -207,7 +207,7 @@ public class TbDeviceProfileNodeTest { TbMsgDataType.JSON, mapper.writeValueAsString(data), null, null); node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg2, "Alarm Updated"); + verify(ctx).enqueueForTellNext(theMsg2, "Alarm Updated"); } @@ -289,7 +289,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -376,7 +376,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -445,7 +445,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -554,7 +554,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -678,7 +678,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -781,7 +781,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -897,7 +897,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -999,7 +999,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg2); verify(ctx).tellSuccess(msg2); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -1082,7 +1082,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -1163,7 +1163,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -1237,7 +1237,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -1321,7 +1321,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); } @@ -1407,7 +1407,7 @@ public class TbDeviceProfileNodeTest { node.onMsg(ctx, msg); verify(ctx).tellSuccess(msg); - verify(ctx).tellNext(theMsg, "Alarm Created"); + verify(ctx).enqueueForTellNext(theMsg, "Alarm Created"); verify(ctx, Mockito.never()).tellFailure(Mockito.any(), Mockito.any()); }