diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java index 4d3e668d61..bb34b14e19 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java @@ -41,8 +41,10 @@ import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.device.credentials.BasicMqttCredentials; import org.thingsboard.server.common.data.device.credentials.ProvisionDeviceCredentialsData; +import org.thingsboard.server.common.data.device.data.CoapDeviceTransportConfiguration; import org.thingsboard.server.common.data.device.data.Lwm2mDeviceTransportConfiguration; import org.thingsboard.server.common.data.device.data.PowerMode; +import org.thingsboard.server.common.data.device.data.PowerSavingConfiguration; import org.thingsboard.server.common.data.device.profile.ProvisionDeviceProfileCredentials; import org.thingsboard.server.common.data.id.CustomerId; import org.thingsboard.server.common.data.id.DeviceId; @@ -461,16 +463,6 @@ public class DefaultTransportApiService implements TransportApiService { } private DeviceInfoProto getDeviceInfoProto(Device device) throws JsonProcessingException { - PowerMode powerMode = null; - Long edrxCycle = null; - switch (device.getDeviceData().getTransportConfiguration().getType()) { - case LWM2M: - Lwm2mDeviceTransportConfiguration transportConfiguration = (Lwm2mDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); - powerMode = transportConfiguration.getPowerMode(); - edrxCycle = transportConfiguration.getEdrxCycle(); - break; - } - DeviceInfoProto.Builder builder = DeviceInfoProto.newBuilder() .setTenantIdMSB(device.getTenantId().getId().getMostSignificantBits()) .setTenantIdLSB(device.getTenantId().getId().getLeastSignificantBits()) @@ -483,9 +475,28 @@ public class DefaultTransportApiService implements TransportApiService { .setDeviceProfileIdMSB(device.getDeviceProfileId().getId().getMostSignificantBits()) .setDeviceProfileIdLSB(device.getDeviceProfileId().getId().getLeastSignificantBits()) .setAdditionalInfo(mapper.writeValueAsString(device.getAdditionalInfo())); - if (powerMode != null) { - builder.setPowerMode(powerMode.name()); - builder.setEdrxCycle(edrxCycle); + + PowerSavingConfiguration psmConfiguration = null; + switch (device.getDeviceData().getTransportConfiguration().getType()) { + case LWM2M: + psmConfiguration = (Lwm2mDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); + break; + case COAP: + psmConfiguration = (CoapDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); + break; + } + + if (psmConfiguration != null) { + PowerMode powerMode = psmConfiguration.getPowerMode(); + if (powerMode != null) { + builder.setPowerMode(powerMode.name()); + if (powerMode.equals(PowerMode.PSM)) { + builder.setPsmActivityTimer(checkLong(psmConfiguration.getPsmActivityTimer())); + } else if (powerMode.equals(PowerMode.E_DRX)) { + builder.setEdrxCycle(checkLong(psmConfiguration.getEdrxCycle())); + builder.setPagingTransmissionWindow(checkLong(psmConfiguration.getPagingTransmissionWindow())); + } + } } return builder.build(); } @@ -584,4 +595,8 @@ public class DefaultTransportApiService implements TransportApiService { deviceCreationLock.unlock(); } } + + private Long checkLong(Long l) { + return l != null ? l : 0; + } } diff --git a/application/src/main/resources/logback.xml b/application/src/main/resources/logback.xml index caf78e1537..35e81193bf 100644 --- a/application/src/main/resources/logback.xml +++ b/application/src/main/resources/logback.xml @@ -41,6 +41,9 @@ + + + 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 07503af03a..b5495042cf 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 @@ -166,13 +166,12 @@ public class DefaultCoapClientContext implements CoapClientContext { var clientProfile = getProfile(client.getProfileId()); if (clientProfile.isPresent()) { profileSettings = clientProfile.get().getClientSettings(); - powerMode = profileSettings.getPowerMode(); - if (powerMode == null) { - powerMode = PowerMode.DRX; + if (profileSettings != null) { + powerMode = profileSettings.getPowerMode(); } } } - if (PowerMode.DRX.equals(powerMode)) { + if (powerMode == null || PowerMode.DRX.equals(powerMode)) { client.updateLastUplinkTime(); return; } @@ -598,13 +597,12 @@ public class DefaultCoapClientContext implements CoapClientContext { var clientProfile = getProfile(client.getProfileId()); if (clientProfile.isPresent()) { profileSettings = clientProfile.get().getClientSettings(); - powerMode = profileSettings.getPowerMode(); - if (powerMode == null) { - powerMode = PowerMode.DRX; + if (profileSettings != null) { + powerMode = profileSettings.getPowerMode(); } } } - if (PowerMode.DRX.equals(powerMode)) { + if (powerMode == null || PowerMode.DRX.equals(powerMode)) { return true; } client.lock(); @@ -679,10 +677,7 @@ public class DefaultCoapClientContext implements CoapClientContext { TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(), new CoapOkCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); if (state.getAttrs() == null) { - transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); - transportService.deregisterSession(state.getSession()); - state.setSession(null); - //TODO: need to delete the client from context as well. + closeAndCleanup(state); } } } @@ -696,11 +691,18 @@ public class DefaultCoapClientContext implements CoapClientContext { TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().setUnsubscribe(true).build(), new CoapOkCallback(exchange, CoAP.ResponseCode.DELETED, CoAP.ResponseCode.INTERNAL_SERVER_ERROR)); if (state.getRpc() == null) { - transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); - transportService.deregisterSession(state.getSession()); - state.setSession(null); - //TODO: need to delete the client from context as well. + closeAndCleanup(state); } } } + + private void closeAndCleanup(TbCoapClientState state) { + transportService.process(state.getSession(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); + transportService.deregisterSession(state.getSession()); + state.setSession(null); + state.setConfiguration(null); + state.setCredentials(null); + state.setAdaptor(null); + //TODO: add optimistic lock check that the client was already deleted and cleanup "clients" map. + } } 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 9861b4f783..7613dc2ece 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 @@ -84,6 +84,7 @@ public class TbCoapClientState { } public void init(ValidateDeviceCredentialsResponse credentials) { + this.credentials = credentials; this.profileId = credentials.getDeviceInfo().getDeviceProfileId(); this.powerMode = credentials.getDeviceInfo().getPowerMode(); this.edrxCycle = credentials.getDeviceInfo().getEdrxCycle(); diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/config/LwM2MTransportServerConfig.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/config/LwM2MTransportServerConfig.java index 8aeacc72c2..b0cb237e42 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/config/LwM2MTransportServerConfig.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/config/LwM2MTransportServerConfig.java @@ -128,6 +128,13 @@ public class LwM2MTransportServerConfig implements LwM2MSecureServerConfig { @Value("${transport.lwm2m.log_max_length:}") private int logMaxLength; + @Getter + @Value("${transport.lwm2m.psm_activity_timer:10000}") + private long psmActivityTimer; + + @Getter + @Value("${transport.lwm2m.paging_transmission_window:10000}") + private long pagingTransmissionWindow; @PostConstruct public void init() { diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClient.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClient.java index a191849215..04917baadc 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClient.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClient.java @@ -50,6 +50,7 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.stream.Collectors; @@ -93,10 +94,24 @@ public class LwM2mClient implements Serializable { @Getter private PowerMode powerMode; @Getter + private Long psmActivityTimer; + @Getter private Long edrxCycle; @Getter + private Long pagingTransmissionWindow; + @Getter @Setter private Registration registration; + @Getter + @Setter + private boolean asleep; + @Getter + private long lastUplinkTime; + @Getter + @Setter + private transient Future sleepTask; + + private boolean firstEdrxDownlink = true; public Object clone() throws CloneNotSupportedException { return super.clone(); @@ -118,6 +133,8 @@ public class LwM2mClient implements Serializable { this.profileId = new UUID(session.getDeviceProfileIdMSB(), session.getDeviceProfileIdLSB()); this.powerMode = credentials.getDeviceInfo().getPowerMode(); this.edrxCycle = credentials.getDeviceInfo().getEdrxCycle(); + this.psmActivityTimer = credentials.getDeviceInfo().getPsmActivityTimer(); + this.pagingTransmissionWindow = credentials.getDeviceInfo().getPagingTransmissionWindow(); } public void lock() { @@ -139,6 +156,8 @@ public class LwM2mClient implements Serializable { Lwm2mDeviceTransportConfiguration transportConfiguration = (Lwm2mDeviceTransportConfiguration) device.getDeviceData().getTransportConfiguration(); this.powerMode = transportConfiguration.getPowerMode(); this.edrxCycle = transportConfiguration.getEdrxCycle(); + this.psmActivityTimer = transportConfiguration.getPsmActivityTimer(); + this.pagingTransmissionWindow = transportConfiguration.getPagingTransmissionWindow(); } public void onDeviceProfileUpdate(DeviceProfile deviceProfile) { @@ -236,7 +255,12 @@ public class LwM2mClient implements Serializable { } public boolean isResourceMultiInstances(String pathIdVer, LwM2mModelProvider modelProvider) { - return getResourceModel(pathIdVer, modelProvider).multiple; + var resourceModel = getResourceModel(pathIdVer, modelProvider); + if (resourceModel != null && resourceModel.multiple != null) { + return resourceModel.multiple; + } else { + return false; + } } public ObjectModel getObjectModel(String pathIdVer, LwM2mModelProvider modelProvider) { @@ -344,5 +368,17 @@ public class LwM2mClient implements Serializable { this.lock = new ReentrantLock(); } + public long updateLastUplinkTime(){ + this.lastUplinkTime = System.currentTimeMillis(); + this.firstEdrxDownlink = true; + return lastUplinkTime; + } + + public boolean checkFirstDownlink() { + boolean result = firstEdrxDownlink; + firstEdrxDownlink = false; + return result; + } + } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContext.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContext.java index 65dec09fc8..c05148017c 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContext.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContext.java @@ -63,5 +63,14 @@ public interface LwM2mClientContext { boolean isComposite(LwM2mClient client); + void onUplink(LwM2mClient client); + Long getRequestTimeout(LwM2mClient client); + + boolean asleep(LwM2mClient client); + + boolean awake(LwM2mClient client); + + boolean isDownlinkAllowed(LwM2mClient client); + } 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 5b92323634..7537e63462 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 @@ -53,6 +53,8 @@ import java.util.Optional; import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.function.Predicate; import static org.eclipse.leshan.core.SecurityMode.NO_SEC; @@ -128,7 +130,7 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { @Override public Optional register(LwM2mClient client, Registration registration) throws LwM2MClientStateException { - TransportProtos.SessionInfoProto oldSession = null; + TransportProtos.SessionInfoProto oldSession; client.lock(); try { if (LwM2MClientState.UNREGISTERED.equals(client.getState())) { @@ -158,13 +160,64 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { client.setRegistration(registration); this.lwM2mClientsByRegistrationId.put(registration.getId(), client); client.setState(LwM2MClientState.REGISTERED); - clientStore.put(client); + onUplink(client); + if(!compareAndSetSleepFlag(client, false)){ + clientStore.put(client); + } } finally { client.unlock(); } return Optional.ofNullable(oldSession); } + @Override + public boolean asleep(LwM2mClient client) { + boolean changed = compareAndSetSleepFlag(client, true); + if (changed) { + log.debug("[{}] client is sleeping", client.getEndpoint()); + context.getTransportService().log(client.getSession(), "Info : Client is sleeping!"); + } + return changed; + } + + @Override + public boolean awake(LwM2mClient client) { + onUplink(client); + boolean changed = compareAndSetSleepFlag(client, false); + if (changed) { + log.debug("[{}] client is awake", client.getEndpoint()); + context.getTransportService().log(client.getSession(), "Info : Client is awake!"); + sendMsgsAfterSleeping(client); + } + return changed; + } + + private boolean compareAndSetSleepFlag(LwM2mClient client, boolean sleeping) { + if (sleeping == client.isAsleep()) { + log.trace("[{}] Client is already at sleeping: {}, ignoring event: {}", client.getEndpoint(), client.isAsleep(), sleeping); + return false; + } + client.lock(); + try { + if (sleeping == client.isAsleep()) { + log.trace("[{}] Client is already at sleeping: {}, ignoring event: {}", client.getEndpoint(), client.isAsleep(), sleeping); + return false; + } else { + PowerMode powerMode = getPowerMode(client); + if (PowerMode.PSM.equals(powerMode) || PowerMode.E_DRX.equals(powerMode)) { + log.trace("[{}] Switch sleeping from: {} to: {}", client.getEndpoint(), client.isAsleep(), sleeping); + client.setAsleep(sleeping); + update(client); + return true; + } else { + return false; + } + } + } finally { + client.unlock(); + } + } + @Override public void updateRegistration(LwM2mClient client, Registration registration) throws LwM2MClientStateException { client.lock(); @@ -173,8 +226,12 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { throw new LwM2MClientStateException(client.getState(), "Client is in invalid state."); } client.setRegistration(registration); - clientStore.put(client); - sendMsgsAfterSleeping(client); + onUplink(client); + if (compareAndSetSleepFlag(client, false)) { + sendMsgsAfterSleeping(client); + } else { + clientStore.put(client); + } } finally { client.unlock(); } @@ -268,12 +325,7 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { @Override public void sendMsgsAfterSleeping(LwM2mClient lwM2MClient) { if (LwM2MClientState.REGISTERED.equals(lwM2MClient.getState())) { - PowerMode powerMode = lwM2MClient.getPowerMode(); - if (powerMode == null) { - Lwm2mDeviceProfileTransportConfiguration deviceProfile = getProfile(lwM2MClient.getProfileId()); - powerMode = deviceProfile.getClientLwM2mSettings().getPowerMode(); - } - + PowerMode powerMode = getPowerMode(lwM2MClient); if (PowerMode.PSM.equals(powerMode) || PowerMode.E_DRX.equals(powerMode)) { defaultLwM2MUplinkMsgHandler.initAttributes(lwM2MClient); TransportProtos.TransportToDeviceActorMsg persistentRpcRequestMsg = TransportProtos.TransportToDeviceActorMsg @@ -287,6 +339,15 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { } } + private PowerMode getPowerMode(LwM2mClient lwM2MClient) { + PowerMode powerMode = lwM2MClient.getPowerMode(); + if (powerMode == null) { + Lwm2mDeviceProfileTransportConfiguration deviceProfile = getProfile(lwM2MClient.getProfileId()); + powerMode = deviceProfile.getClientLwM2mSettings().getPowerMode(); + } + return powerMode; + } + @Override public Collection getLwM2mClients() { return lwM2mClientsByEndpoint.values(); @@ -347,6 +408,113 @@ public class LwM2mClientContextImpl implements LwM2mClientContext { getProfile(client.getProfileId()).getClientLwM2mSettings().isCompositeOperationsSupport(); } + @Override + public boolean isDownlinkAllowed(LwM2mClient client) { + PowerMode powerMode = client.getPowerMode(); + OtherConfiguration profileSettings = null; + if (powerMode == null) { + var clientProfile = getProfile(client.getProfileId()); + profileSettings = clientProfile.getClientLwM2mSettings(); + powerMode = profileSettings.getPowerMode(); + if (powerMode == null) { + powerMode = PowerMode.DRX; + } + } + if (PowerMode.DRX.equals(powerMode)) { + return true; + } + client.lock(); + long timeSinceLastUplink = System.currentTimeMillis() - client.getLastUplinkTime(); + try { + if (PowerMode.PSM.equals(powerMode)) { + Long psmActivityTimer = client.getPsmActivityTimer(); + if (psmActivityTimer == null && profileSettings != null) { + psmActivityTimer = profileSettings.getPsmActivityTimer(); + + } + if (psmActivityTimer == null || psmActivityTimer == 0L) { + psmActivityTimer = config.getPsmActivityTimer(); + } + return timeSinceLastUplink <= psmActivityTimer; + } else { + Long pagingTransmissionWindow = client.getPagingTransmissionWindow(); + if (pagingTransmissionWindow == null && profileSettings != null) { + pagingTransmissionWindow = profileSettings.getPagingTransmissionWindow(); + + } + if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) { + pagingTransmissionWindow = config.getPagingTransmissionWindow(); + } + boolean allowed = timeSinceLastUplink <= pagingTransmissionWindow; + if (!allowed) { + return client.checkFirstDownlink(); + } else { + return true; + } + } + } finally { + client.unlock(); + } + } + + @Override + public void onUplink(LwM2mClient client) { + PowerMode powerMode = client.getPowerMode(); + OtherConfiguration profileSettings = null; + if (powerMode == null) { + var clientProfile = getProfile(client.getProfileId()); + profileSettings = clientProfile.getClientLwM2mSettings(); + powerMode = profileSettings.getPowerMode(); + if (powerMode == null) { + powerMode = PowerMode.DRX; + } + } + if (PowerMode.DRX.equals(powerMode)) { + client.updateLastUplinkTime(); + return; + } + client.lock(); + try { + long uplinkTime = client.updateLastUplinkTime(); + long timeout; + if (PowerMode.PSM.equals(powerMode)) { + Long psmActivityTimer = client.getPsmActivityTimer(); + if (psmActivityTimer == null && profileSettings != null) { + psmActivityTimer = profileSettings.getPsmActivityTimer(); + + } + if (psmActivityTimer == null || psmActivityTimer == 0L) { + psmActivityTimer = config.getPsmActivityTimer(); + } + + timeout = psmActivityTimer; + } else { + Long pagingTransmissionWindow = client.getPagingTransmissionWindow(); + if (pagingTransmissionWindow == null && profileSettings != null) { + pagingTransmissionWindow = profileSettings.getPagingTransmissionWindow(); + + } + if (pagingTransmissionWindow == null || pagingTransmissionWindow == 0L) { + pagingTransmissionWindow = config.getPagingTransmissionWindow(); + } + timeout = pagingTransmissionWindow; + } + Future sleepTask = client.getSleepTask(); + if (sleepTask != null) { + sleepTask.cancel(false); + } + Future task = context.getScheduler().schedule(() -> { + if (uplinkTime == client.getLastUplinkTime()) { + asleep(client); + } + return null; + }, timeout, TimeUnit.MILLISECONDS); + client.setSleepTask(task); + } finally { + client.unlock(); + } + } + @Override public Long getRequestTimeout(LwM2mClient client) { Long timeout = null; diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/downlink/DefaultLwM2mDownlinkMsgHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/downlink/DefaultLwM2mDownlinkMsgHandler.java index 96c93b177a..82ea0c5fbe 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/downlink/DefaultLwM2mDownlinkMsgHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/downlink/DefaultLwM2mDownlinkMsgHandler.java @@ -39,6 +39,8 @@ import org.eclipse.leshan.core.request.SimpleDownlinkRequest; import org.eclipse.leshan.core.request.WriteAttributesRequest; import org.eclipse.leshan.core.request.WriteCompositeRequest; import org.eclipse.leshan.core.request.WriteRequest; +import org.eclipse.leshan.core.request.exception.ClientSleepingException; +import org.eclipse.leshan.core.request.exception.TimeoutException; import org.eclipse.leshan.core.response.DeleteResponse; import org.eclipse.leshan.core.response.DiscoverResponse; import org.eclipse.leshan.core.response.ExecuteResponse; @@ -59,6 +61,7 @@ import org.thingsboard.server.queue.util.TbLwM2mTransportComponent; import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig; import org.thingsboard.server.transport.lwm2m.server.LwM2mTransportContext; import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClient; +import org.thingsboard.server.transport.lwm2m.server.client.LwM2mClientContext; import org.thingsboard.server.transport.lwm2m.server.common.LwM2MExecutorAwareService; import org.thingsboard.server.transport.lwm2m.server.downlink.composite.TbLwM2MReadCompositeRequest; import org.thingsboard.server.transport.lwm2m.server.log.LwM2MTelemetryLogService; @@ -95,6 +98,7 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im private final LwM2mTransportContext context; private final LwM2MTransportServerConfig config; private final LwM2MTelemetryLogService logService; + private final LwM2mClientContext clientContext; @PostConstruct public void init() { @@ -236,10 +240,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im path.getObjectId(), path.getObjectInstanceId(), path.getResourceId(), request.getValue()); sendSimpleRequest(client, downlink, request.getTimeout(), callback); } catch (Exception e) { - callback.onError(JacksonUtil.toString(request), e); + callback.onError(toString(request), e); } } else { - callback.onValidationError(JacksonUtil.toString(request), "Resource " + request.getVersionedId() + " is not configured in the device profile!"); + callback.onValidationError(toString(request), "Resource " + request.getVersionedId() + " is not configured in the device profile!"); } } @@ -251,7 +255,7 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im //TODO: replace config.getTimeout(); sendWriteCompositeRequest(client, downlink, config.getTimeout(), callback); } catch (Exception e) { - callback.onError(JacksonUtil.toString(rpcWriteCompositeRequest), e); + callback.onError(toString(rpcWriteCompositeRequest), e); } } @@ -281,10 +285,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im WriteRequest downlink = new WriteRequest(WriteRequest.Mode.UPDATE, contentFormat, resultIds.getObjectId(), resultIds.getObjectInstanceId(), resources); sendSimpleRequest(client, downlink, request.getTimeout(), callback); } else { - callback.onValidationError(JacksonUtil.toString(request), "No resources to update!"); + callback.onValidationError(toString(request), "No resources to update!"); } } else { - callback.onValidationError(JacksonUtil.toString(request), "Update of the root level object is not supported yet!"); + callback.onValidationError(toString(request), "Update of the root level object is not supported yet!"); } } @@ -298,6 +302,10 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im } private , T extends LwM2mResponse> void sendRequest(LwM2mClient client, R request, long timeoutInMs, DownlinkRequestCallback callback, Function pathToStringFunction) { + if (!clientContext.isDownlinkAllowed(client)) { + log.trace("[{}] ignore downlink request cause client is sleeping.", client.getEndpoint()); + return; + } Registration registration = client.getRegistration(); try { logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), pathToStringFunction.apply(request))); @@ -307,20 +315,21 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im callback.onSuccess(request, response); } catch (Exception e) { log.error("[{}] failed to process successful response [{}] ", registration.getEndpoint(), response, e); + } finally { + clientContext.awake(client); } }); - }, e -> { - executor.submit(() -> { - callback.onError(JacksonUtil.toString(request), e); - }); - }); + },e -> handleDownlinkError(client, request, callback, e)); } catch (Exception e) { - callback.onError(JacksonUtil.toString(request), e); + handleDownlinkError(client, request, callback, e); } } - private , T extends LwM2mResponse> void sendWriteCompositeRequest(LwM2mClient client, WriteCompositeRequest request, long timeoutInMs, - DownlinkRequestCallback callback) { + private , T extends LwM2mResponse> void sendWriteCompositeRequest(LwM2mClient client, WriteCompositeRequest request, long timeoutInMs, DownlinkRequestCallback callback) { + if (!clientContext.isDownlinkAllowed(client)) { + log.trace("[{}] ignore downlink request cause client is sleeping.", client.getEndpoint()); + return; + } Registration registration = client.getRegistration(); try { logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), request.getPaths())); @@ -334,63 +343,28 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im } } catch (Exception e) { log.error("[{}] failed to process successful response [{}] ", registration.getEndpoint(), response, e); + } finally { + clientContext.awake(client); } }); - }, e -> { - executor.submit(() -> { - callback.onError(JacksonUtil.toString(request), e); - }); - }); + }, e -> handleDownlinkError(client, request, callback, e)); } catch (Exception e) { - callback.onError(JacksonUtil.toString(request), e); + handleDownlinkError(client, request, callback, e); } } -// private , T extends LwM2mResponse> void sendReadRequestComposite(LwM2mClient client, ReadCompositeRequest request, long timeoutInMs, -// DownlinkRequestCallback callback) { -// Registration registration = client.getRegistration(); -// try { -// logService.log(client, String.format("[%s][%s] Sending request: %s to %s", registration.getId(), registration.getSocketAddress(), request.getClass().getSimpleName(), request.getPaths())); -// context.getServer().send(registration, request, timeoutInMs, response -> { -// executor.submit(() -> { -// try { -// /** -// * [{"bn":"/3/0/","n":"0","vs":"Thingsboard Test Device"}, -// * {"n":"1","vs":"Model 500"}, -// * {"n":"2","vs":"TH-500-000-0001"}, -// * {"n":"3","vs":"TestThingsboard@TestMore1024_2.04"}, -// * {"n":"6","v":1},{"n":"7","v":56}, -// * {"n":"8","v":42},{"n":"9","v":16}, -// * {"n":"10","v":127619},{"n":"13","v":1624520988}, -// * {"n":"14","vs":"+03"},{"n":"15","vs":"Europe/Kiev"}, -// * {"n":"16","vs":"U"},{"n":"17","vs":"smart meters"}, -// * {"n":"18","vs":"1.01"},{"n":"19","vs":"1.02"}, -// * {"n":"20","v":3},{"n":"21","v":256000}, -// * {"bn":"/5/0/","n":"1","vs":""}, -// * {"n":"3","v":0},{"n":"5","v":0}, -// * {"n":"6","vs":""},{"n":"7","vs":""}, -// * {"n":"8/0","v":0},{"n":"8/1","v":1}, -// * {"n":"9","v":2}, -// * {"bn":"/1/0/","n":"0","v":123}, -// * {"n":"1","v":300}, -// * {"n":"6","vb":false}, -// * {"n":"22","vs":"U"}, -// * {"n":"7","vs":"U"}] -// */ -// callback.onSuccess(request, response); -// } catch (Exception e) { -// log.error("[{}] failed to process successful response [{}] ", registration.getEndpoint(), response, e); -// } -// }); -// }, e -> { -// executor.submit(() -> { -// callback.onError(JacksonUtil.toString(request), e); -// }); -// }); -// } catch (Exception e) { -// callback.onError(JacksonUtil.toString(request), e); -// } -// } + private , T extends LwM2mResponse> void handleDownlinkError(LwM2mClient client, R request, DownlinkRequestCallback callback, Exception e) { + log.trace("[{}] Received downlink error: {}.", client.getEndpoint(), e); + executor.submit(() -> { + if (e instanceof TimeoutException || e instanceof ClientSleepingException) { + log.trace("[{}] Received {}, client is probably sleeping", client.getEndpoint(), e.getClass().getSimpleName()); + clientContext.asleep(client); + } else { + log.trace("[{}] Received {}", client.getEndpoint(), e.getClass().getSimpleName()); + } + callback.onError(toString(request), e); + }); + } private WriteRequest getWriteRequestSingleResource(ResourceModel.Type type, ContentFormat contentFormat, int objectId, int instanceId, int resourceId, Object value) { switch (type) { @@ -483,10 +457,22 @@ public class DefaultLwM2mDownlinkMsgHandler extends LwM2MExecutorAwareService im String id = fromVersionedIdToObjectId(versionedId); if (id != null && new LwM2mPath(id).isResource() && !client.isResourceMultiInstances(versionedId, modelProvider)) { return client.getDefaultContentFormat(); - } - else { + } else { return ContentFormat.DEFAULT; } } } + + private String toString(R request) { + try { + try { + return JacksonUtil.toString(request); + } catch (Exception e) { + return request.toString(); + } + } catch (Exception e) { + log.warn("Failed to convert request to string", e); + return request != null ? request.getClass().getSimpleName() : ""; + } + } } diff --git a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java index 0342d3dd05..2830e7e437 100644 --- a/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java +++ b/common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java @@ -80,53 +80,53 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { private final Map rpcSubscriptions = new ConcurrentHashMap<>(); @Override - public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg rpcRequst, TransportProtos.SessionInfoProto sessionInfo) { + public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { this.cleanupOldSessions(); - UUID requestUUID = new UUID(rpcRequst.getRequestIdMSB(), rpcRequst.getRequestIdLSB()); - log.debug("Received params: {}", rpcRequst.getParams()); + UUID requestUUID = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB()); + log.debug("Received params: {}", rpcRequest.getParams()); // We use this map to protect from browser issue that the same command is sent twice. // TODO: This is probably not the best place and should be moved to DeviceActor if (!this.rpcSubscriptions.containsKey(requestUUID)) { - LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequst.getMethodName()); + LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName()); if (operationType == null) { - this.sendErrorRpcResponse(sessionInfo, rpcRequst.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED.getName(), "Unsupported operation type: " + rpcRequst.getMethodName()); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED.getName(), "Unsupported operation type: " + rpcRequest.getMethodName()); return; } LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo); if (client.getRegistration() == null) { - this.sendErrorRpcResponse(sessionInfo, rpcRequst.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR.getName(), "Registration is empty"); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR.getName(), "Registration is empty"); return; } try { if (operationType.isHasObjectId()) { - String objectId = getIdFromParameters(client, rpcRequst); + String objectId = getIdFromParameters(client, rpcRequest); switch (operationType) { case READ: - sendReadRequest(client, rpcRequst, objectId); + sendReadRequest(client, rpcRequest, objectId); break; case OBSERVE: - sendObserveRequest(client, rpcRequst, objectId); + sendObserveRequest(client, rpcRequest, objectId); break; case DISCOVER: - sendDiscoverRequest(client, rpcRequst, objectId); + sendDiscoverRequest(client, rpcRequest, objectId); break; case EXECUTE: - sendExecuteRequest(client, rpcRequst, objectId); + sendExecuteRequest(client, rpcRequest, objectId); break; case WRITE_ATTRIBUTES: - sendWriteAttributesRequest(client, rpcRequst, objectId); + sendWriteAttributesRequest(client, rpcRequest, objectId); break; case OBSERVE_CANCEL: - sendCancelObserveRequest(client, rpcRequst, objectId); + sendCancelObserveRequest(client, rpcRequest, objectId); break; case DELETE: - sendDeleteRequest(client, rpcRequst, objectId); + sendDeleteRequest(client, rpcRequest, objectId); break; case WRITE_UPDATE: - sendWriteUpdateRequest(client, rpcRequst, objectId); + sendWriteUpdateRequest(client, rpcRequest, objectId); break; case WRITE_REPLACE: - sendWriteReplaceRequest(client, rpcRequst, objectId); + sendWriteReplaceRequest(client, rpcRequest, objectId); break; default: throw new IllegalArgumentException("Unsupported operation: " + operationType.name()); @@ -135,28 +135,28 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { if (clientContext.isComposite(client)) { switch (operationType) { case READ_COMPOSITE: - sendReadCompositeRequest(client, rpcRequst); + sendReadCompositeRequest(client, rpcRequest); break; case WRITE_COMPOSITE: - sendWriteCompositeRequest(client, rpcRequst); + sendWriteCompositeRequest(client, rpcRequest); break; default: throw new IllegalArgumentException("Unsupported operation: " + operationType.name()); } } else { - this.sendErrorRpcResponse(sessionInfo, rpcRequst.getRequestId(), + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR.getName(), "This device does not support Composite Operation"); } } else { switch (operationType) { case OBSERVE_CANCEL_ALL: - sendCancelAllObserveRequest(client, rpcRequst); + sendCancelAllObserveRequest(client, rpcRequest); break; case OBSERVE_READ_ALL: - sendObserveAllRequest(client, rpcRequst); + sendObserveAllRequest(client, rpcRequest); break; case DISCOVER_ALL: - sendDiscoverAllRequest(client, rpcRequst); + sendDiscoverAllRequest(client, rpcRequest); break; case FW_UPDATE: //TODO: implement and add break statement @@ -165,7 +165,7 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler { } } } catch (IllegalArgumentException e) { - this.sendErrorRpcResponse(sessionInfo, rpcRequst.getRequestId(), ResponseCode.BAD_REQUEST.getName(), e.getMessage()); + this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST.getName(), e.getMessage()); } } } 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 799ba54b4a..ebf1021c41 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 @@ -213,7 +213,8 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl log.info("[{}] Closing old session: {}", registration.getEndpoint(), new UUID(oldSessionInfo.get().getSessionIdMSB(), oldSessionInfo.get().getSessionIdLSB())); sessionManager.deregister(oldSessionInfo.get()); } - logService.log(lwM2MClient, LOG_LWM2M_INFO + ": Client registered with registration id: " + registration.getId()); + logService.log(lwM2MClient, LOG_LWM2M_INFO + ": Client registered with registration id: " + registration.getId() + " version: " + + registration.getLwM2mVersion() + " and modes: " + registration.getQueueMode() + ", " + registration.getBindingMode()); sessionManager.register(lwM2MClient.getSession()); this.initClientTelemetry(lwM2MClient); this.initAttributes(lwM2MClient); @@ -288,9 +289,8 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl @Override public void onSleepingDev(Registration registration) { - log.info("[{}] [{}] Received endpoint Sleeping version event", registration.getId(), registration.getEndpoint()); - logService.log(clientContext.getClientByEndpoint(registration.getEndpoint()), LOG_LWM2M_INFO + ": Client is sleeping!"); - //TODO: associate endpointId with device information. + log.debug("[{}] [{}] Received endpoint sleeping event", registration.getId(), registration.getEndpoint()); + clientContext.asleep(clientContext.getClientByEndpoint(registration.getEndpoint())); } /** @@ -318,6 +318,12 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl } } clientContext.update(lwM2MClient); + if (clientContext.awake(lwM2MClient)) { + // clientContext.awake calls clientContext.update + log.debug("[{}] Device is awake", lwM2MClient.getEndpoint()); + } else { + clientContext.update(lwM2MClient); + } } } @@ -336,6 +342,13 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl } } }); + clientContext.update(lwM2MClient); + if (clientContext.awake(lwM2MClient)) { + // clientContext.awake calls clientContext.update + log.debug("[{}] Device is awake", lwM2MClient.getEndpoint()); + } else { + clientContext.update(lwM2MClient); + } } } @@ -390,10 +403,8 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl */ @Override public void onAwakeDev(Registration registration) { - log.trace("[{}] [{}] Received endpoint Awake version event", registration.getId(), registration.getEndpoint()); - LwM2mClient client = this.clientContext.getClientByEndpoint(registration.getEndpoint()); - logService.log(client, LOG_LWM2M_INFO + ": Client is awake!"); - clientContext.sendMsgsAfterSleeping(client); + log.debug("[{}] [{}] Received endpoint awake event", registration.getId(), registration.getEndpoint()); + clientContext.awake(clientContext.getClientByEndpoint(registration.getEndpoint())); } /** @@ -420,41 +431,51 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl } private void sendReadRequests(LwM2mClient lwM2MClient, Lwm2mDeviceProfileTransportConfiguration profile, Set supportedObjects) { - Set targetIds = new HashSet<>(profile.getObserveAttr().getAttribute()); - targetIds.addAll(profile.getObserveAttr().getTelemetry()); - targetIds = diffSets(profile.getObserveAttr().getObserve(), targetIds); - targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet()); - - CountDownLatch latch = new CountDownLatch(targetIds.size()); - targetIds.forEach(versionedId -> sendReadRequest(lwM2MClient, versionedId, - new TbLwM2MLatchCallback<>(latch, new TbLwM2MReadCallback(this, logService, lwM2MClient, versionedId)))); try { + Set targetIds = new HashSet<>(profile.getObserveAttr().getAttribute()); + targetIds.addAll(profile.getObserveAttr().getTelemetry()); + targetIds = diffSets(profile.getObserveAttr().getObserve(), targetIds); + targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet()); + + CountDownLatch latch = new CountDownLatch(targetIds.size()); + targetIds.forEach(versionedId -> sendReadRequest(lwM2MClient, versionedId, + new TbLwM2MLatchCallback<>(latch, new TbLwM2MReadCallback(this, logService, lwM2MClient, versionedId)))); latch.await(); } catch (InterruptedException e) { log.error("[{}] Failed to await Read requests!", lwM2MClient.getEndpoint()); + } catch (Exception e) { + log.error("[{}] Failed to process read requests!", lwM2MClient.getEndpoint(), e); + logService.log(lwM2MClient, "Failed to process read requests. Possible profile misconfiguration."); } } private void sendObserveRequests(LwM2mClient lwM2MClient, Lwm2mDeviceProfileTransportConfiguration profile, Set supportedObjects) { - Set targetIds = profile.getObserveAttr().getObserve(); - targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet()); - - CountDownLatch latch = new CountDownLatch(targetIds.size()); - targetIds.forEach(targetId -> sendObserveRequest(lwM2MClient, targetId, - new TbLwM2MLatchCallback<>(latch, new TbLwM2MObserveCallback(this, logService, lwM2MClient, targetId)))); try { + Set targetIds = profile.getObserveAttr().getObserve(); + targetIds = targetIds.stream().filter(target -> isSupportedTargetId(supportedObjects, target)).collect(Collectors.toSet()); + + CountDownLatch latch = new CountDownLatch(targetIds.size()); + targetIds.forEach(targetId -> sendObserveRequest(lwM2MClient, targetId, + new TbLwM2MLatchCallback<>(latch, new TbLwM2MObserveCallback(this, logService, lwM2MClient, targetId)))); + latch.await(); } catch (InterruptedException e) { log.error("[{}] Failed to await Observe requests!", lwM2MClient.getEndpoint()); + } catch (Exception e) { + log.error("[{}] Failed to process observe requests!", lwM2MClient.getEndpoint(), e); + logService.log(lwM2MClient, "Failed to process observe requests. Possible profile misconfiguration."); } } private void sendWriteAttributeRequests(LwM2mClient lwM2MClient, Lwm2mDeviceProfileTransportConfiguration profile, Set supportedObjects) { - Map attributesMap = profile.getObserveAttr().getAttributeLwm2m(); - attributesMap = attributesMap.entrySet().stream().filter(target -> isSupportedTargetId(supportedObjects, target.getKey())).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); -// TODO: why do we need to put observe into pending read requests? -// lwM2MClient.getPendingReadRequests().addAll(targetIds); - attributesMap.forEach((targetId, params) -> sendWriteAttributesRequest(lwM2MClient, targetId, params)); + try { + Map attributesMap = profile.getObserveAttr().getAttributeLwm2m(); + attributesMap = attributesMap.entrySet().stream().filter(target -> isSupportedTargetId(supportedObjects, target.getKey())).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + attributesMap.forEach((targetId, params) -> sendWriteAttributesRequest(lwM2MClient, targetId, params)); + } catch (Exception e) { + log.error("[{}] Failed to process write attribute requests!", lwM2MClient.getEndpoint(), e); + logService.log(lwM2MClient, "Failed to process write attribute requests. Possible profile misconfiguration."); + } } private void sendDiscoverRequest(LwM2mClient lwM2MClient, String targetId) { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportServerInitializer.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportServerInitializer.java index 386c2a8dd6..edc26cb4a9 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportServerInitializer.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportServerInitializer.java @@ -46,7 +46,7 @@ public class MqttTransportServerInitializer extends ChannelInitializer