dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Lwm2m: backEnd: add ExecutorService to LwM2MTransportService

Browse Source
This commit is contained in:
nickAS21 2020-12-25 21:35:13 +02:00
parent 5eeb12ea13
commit 9df5be790a
6 changed files with 126 additions and 122 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportHandler.java
View File

@ -107,6 +107,8 @@ public class LwM2MTransportHandler{
public static final String PUT_TYPE_OPER_WRITE_ATTRIBUTES = "wright-attributes"; public static final String PUT_TYPE_OPER_WRITE_ATTRIBUTES = "wright-attributes";
public static final String EVENT_AWAKE = "AWAKE"; public static final String EVENT_AWAKE = "AWAKE";
public static final String SERVICE_CHANNEL = "SERVICE";
public static final String RESPONSE_CHANNEL = "RESP";
@Autowired @Autowired
@Qualifier("LeshanServerCert") @Qualifier("LeshanServerCert")

43
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportRequest.java
View File

@ -74,26 +74,27 @@ import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandle
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_WRITE_REPLACE; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_WRITE_REPLACE;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.PUT_TYPE_OPER_WRITE_ATTRIBUTES; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.PUT_TYPE_OPER_WRITE_ATTRIBUTES;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.PUT_TYPE_OPER_WRITE_UPDATE; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.PUT_TYPE_OPER_WRITE_UPDATE;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.RESPONSE_CHANNEL;
@Slf4j @Slf4j
@Service("LwM2MTransportRequest") @Service("LwM2MTransportRequest")
@ConditionalOnExpression("('${service.type:null}'=='tb-transport' && '${transport.lwm2m.enabled:false}'=='true' ) || ('${service.type:null}'=='monolith' && '${transport.lwm2m.enabled}'=='true')") @ConditionalOnExpression("('${service.type:null}'=='tb-transport' && '${transport.lwm2m.enabled:false}'=='true' ) || ('${service.type:null}'=='monolith' && '${transport.lwm2m.enabled}'=='true')")
public class LwM2MTransportRequest { public class LwM2MTransportRequest {
private final ExecutorService executorService; private ExecutorService executorResponse;
private static final String RESPONSE_CHANNEL = "THINGSBOARD_RESP"; private ExecutorService executorResponseError;
private final LwM2mValueConverterImpl converter; private LwM2mValueConverterImpl converter;
@Autowired @Autowired
LwM2MTransportService service; LwM2MTransportService service;
public LwM2MTransportRequest() {
this.converter = new LwM2mValueConverterImpl();
executorService = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel response", RESPONSE_CHANNEL)));
}
@PostConstruct @PostConstruct
public void init() { public void init() {
this.converter = new LwM2mValueConverterImpl();
executorResponse = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel response", RESPONSE_CHANNEL)));
executorResponseError = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel response Error", RESPONSE_CHANNEL)));
} }
public Collection<Registration> doGetRegistrations(LeshanServer lwServer) { public Collection<Registration> doGetRegistrations(LeshanServer lwServer) {
@ -306,27 +307,21 @@ public class LwM2MTransportRequest {
} }
private void handleResponse(Registration registration, final String path, LwM2mResponse response, DownlinkRequest request, LwM2MClient lwM2MClient, boolean isDelayedUpdate) { private void handleResponse(Registration registration, final String path, LwM2mResponse response, DownlinkRequest request, LwM2MClient lwM2MClient, boolean isDelayedUpdate) {
executorService.submit(new Runnable() { executorResponse.submit(() -> {
@Override try {
public void run() { sendResponse(registration, path, response, request, lwM2MClient, isDelayedUpdate);
try { } catch (Exception e) {
sendResponse(registration, path, response, request, lwM2MClient, isDelayedUpdate); log.error("[{}] endpoint [{}] path [{}] error Unable to after send response.", registration.getEndpoint(), path, e);
} catch (RuntimeException t) {
log.error("[{}] endpoint [{}] path [{}] error Unable to after send response.", registration.getEndpoint(), path, t.toString());
}
} }
}); });
} }
private void handleResponseError(Registration registration, final String path, LwM2MClient lwM2MClient, boolean isDelayedUpdate) { private void handleResponseError(Registration registration, final String path, LwM2MClient lwM2MClient, boolean isDelayedUpdate) {
executorService.submit(new Runnable() { executorResponseError.submit(() -> {
@Override try {
public void run() { if (isDelayedUpdate) lwM2MClient.onSuccessOrErrorDelayedRequests(path);
try { } catch (RuntimeException t) {
if (isDelayedUpdate) lwM2MClient.onSuccessOrErrorDelayedRequests(path); log.error("[{}] endpoint [{}] path [{}] error Unable to after send response.", registration.getEndpoint(), path, t.toString());
} catch (RuntimeException t) {
log.error("[{}] endpoint [{}] path [{}] error Unable to after send response.", registration.getEndpoint(), path, t.toString());
}
} }
}); });
} }

157
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/LwM2MTransportService.java
View File

@ -5,7 +5,7 @@
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
* You may obtain a copy of the License at * You may obtain a copy of the License at
* *
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, * distributed under the License is distributed on an "AS IS" BASIS,
@ -32,6 +32,7 @@ import org.eclipse.leshan.core.observation.Observation;
import org.eclipse.leshan.core.request.ContentFormat; import org.eclipse.leshan.core.request.ContentFormat;
import org.eclipse.leshan.core.request.WriteRequest; import org.eclipse.leshan.core.request.WriteRequest;
import org.eclipse.leshan.core.response.ReadResponse; import org.eclipse.leshan.core.response.ReadResponse;
import org.eclipse.leshan.core.util.NamedThreadFactory;
import org.eclipse.leshan.server.californium.LeshanServer; import org.eclipse.leshan.server.californium.LeshanServer;
import org.eclipse.leshan.server.registration.Registration; import org.eclipse.leshan.server.registration.Registration;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
@ -71,6 +72,8 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Predicate; import java.util.function.Predicate;
@ -89,6 +92,7 @@ import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandle
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.LOG_LW2M_TELEMETRY; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.LOG_LW2M_TELEMETRY;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_EXECUTE; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_EXECUTE;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_WRITE_REPLACE; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.POST_TYPE_OPER_WRITE_REPLACE;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.SERVICE_CHANNEL;
import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.getAckCallback; import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandler.getAckCallback;
@Slf4j @Slf4j
@ -96,6 +100,10 @@ import static org.thingsboard.server.transport.lwm2m.server.LwM2MTransportHandle
@ConditionalOnExpression("('${service.type:null}'=='tb-transport' && '${transport.lwm2m.enabled:false}'=='true' ) || ('${service.type:null}'=='monolith' && '${transport.lwm2m.enabled}'=='true')") @ConditionalOnExpression("('${service.type:null}'=='tb-transport' && '${transport.lwm2m.enabled:false}'=='true' ) || ('${service.type:null}'=='monolith' && '${transport.lwm2m.enabled}'=='true')")
public class LwM2MTransportService { public class LwM2MTransportService {
private ExecutorService executorRegistered;
private ExecutorService executorUpdateRegistered;
private ExecutorService executorUnRegistered;
@Autowired @Autowired
private TransportService transportService; private TransportService transportService;
@ -111,6 +119,12 @@ public class LwM2MTransportService {
@PostConstruct @PostConstruct
public void init() { public void init() {
context.getScheduler().scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) context.getCtxServer().getSessionReportTimeout()), context.getCtxServer().getSessionReportTimeout(), TimeUnit.MILLISECONDS); context.getScheduler().scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) context.getCtxServer().getSessionReportTimeout()), context.getCtxServer().getSessionReportTimeout(), TimeUnit.MILLISECONDS);
executorRegistered = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel registered", SERVICE_CHANNEL)));
executorUpdateRegistered = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel update registered", SERVICE_CHANNEL)));
executorUnRegistered = Executors.newCachedThreadPool(
new NamedThreadFactory(String.format("LwM2M %s channel un registered", SERVICE_CHANNEL)));
} }
/** /**
@ -129,28 +143,35 @@ public class LwM2MTransportService {
* @param previousObsersations - may be null * @param previousObsersations - may be null
*/ */
public void onRegistered(LeshanServer lwServer, Registration registration, Collection<Observation> previousObsersations) { public void onRegistered(LeshanServer lwServer, Registration registration, Collection<Observation> previousObsersations) {
LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.updateInSessionsLwM2MClient(lwServer, registration); executorRegistered.submit(() -> {
if (lwM2MClient != null) { try {
lwM2MClient.setLwM2MTransportService(this); log.info("[{}] Client: onRegistered name ", registration.getEndpoint());
lwM2MClient.setLwM2MTransportService(this); LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.updateInSessionsLwM2MClient(lwServer, registration);
lwM2MClient.setSessionUuid(UUID.randomUUID()); if (lwM2MClient != null) {
this.setLwM2MClient(lwServer, registration, lwM2MClient); lwM2MClient.setLwM2MTransportService(this);
SessionInfoProto sessionInfo = this.getValidateSessionInfo(registration.getId()); lwM2MClient.setLwM2MTransportService(this);
if (sessionInfo != null) { lwM2MClient.setSessionUuid(UUID.randomUUID());
lwM2MClient.setDeviceUuid(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB())); this.setLwM2MClient(lwServer, registration, lwM2MClient);
lwM2MClient.setProfileUuid(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB())); SessionInfoProto sessionInfo = this.getValidateSessionInfo(registration.getId());
lwM2MClient.setDeviceName(sessionInfo.getDeviceName()); if (sessionInfo != null) {
lwM2MClient.setDeviceProfileName(sessionInfo.getDeviceType()); lwM2MClient.setDeviceUuid(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()));
transportService.registerAsyncSession(sessionInfo, new LwM2MSessionMsgListener(this, sessionInfo)); lwM2MClient.setProfileUuid(new UUID(sessionInfo.getDeviceProfileIdMSB(), sessionInfo.getDeviceProfileIdLSB()));
transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), null); lwM2MClient.setDeviceName(sessionInfo.getDeviceName());
transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null); lwM2MClient.setDeviceProfileName(sessionInfo.getDeviceType());
this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client registration", registration.getId()); transportService.registerAsyncSession(sessionInfo, new LwM2MSessionMsgListener(this, sessionInfo));
} else { transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), null);
log.error("Client: [{}] onRegistered [{}] name [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null); transportService.process(sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg.newBuilder().build(), null);
this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client registration", registration.getId());
} else {
log.error("Client: [{}] onRegistered [{}] name [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null);
}
} else {
log.error("Client: [{}] onRegistered [{}] name [{}] lwM2MClient ", registration.getId(), registration.getEndpoint(), null);
}
} catch (Throwable t) {
log.error("[{}] endpoint [{}] error Unable registration.", registration.getEndpoint(), t);
} }
} else { });
log.error("Client: [{}] onRegistered [{}] name [{}] lwM2MClient ", registration.getId(), registration.getEndpoint(), null);
}
} }
/** /**
@ -158,12 +179,18 @@ public class LwM2MTransportService {
* @param registration - Registration LwM2M Client * @param registration - Registration LwM2M Client
*/ */
public void updatedReg(LeshanServer lwServer, Registration registration) { public void updatedReg(LeshanServer lwServer, Registration registration) {
SessionInfoProto sessionInfo = this.getValidateSessionInfo(registration.getId()); executorUpdateRegistered.submit(() -> {
if (sessionInfo != null) { try {
log.info("Client: [{}] updatedReg [{}] name [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType()); SessionInfoProto sessionInfo = this.getValidateSessionInfo(registration.getId());
} else { if (sessionInfo != null) {
log.error("Client: [{}] updatedReg [{}] name [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null); log.info("Client: [{}] updatedReg [{}] name [{}] profile ", registration.getId(), registration.getEndpoint(), sessionInfo.getDeviceType());
} } else {
log.error("Client: [{}] updatedReg [{}] name [{}] sessionInfo ", registration.getId(), registration.getEndpoint(), null);
}
} catch (Throwable t) {
log.error("[{}] endpoint [{}] error Unable update registration.", registration.getEndpoint(), t);
}
});
} }
/** /**
@ -172,8 +199,14 @@ public class LwM2MTransportService {
* !!! Warn: if have not finishing unReg, then this operation will be finished on next Client`s connect * !!! Warn: if have not finishing unReg, then this operation will be finished on next Client`s connect
*/ */
public void unReg(Registration registration, Collection<Observation> observations) { public void unReg(Registration registration, Collection<Observation> observations) {
this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client unRegistration", registration.getId()); executorUpdateRegistered.submit(() -> {
this.closeClientSession(registration); try {
this.sentLogsToThingsboard(LOG_LW2M_INFO + ": Client unRegistration", registration.getId());
this.closeClientSession(registration);
} catch (Throwable t) {
log.error("[{}] endpoint [{}] error Unable un registration.", registration.getEndpoint(), t);
}
});
} }
private void closeClientSession(Registration registration) { private void closeClientSession(Registration registration) {
@ -848,14 +881,12 @@ public class LwM2MTransportService {
* @param deviceProfile - * @param deviceProfile -
*/ */
public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) { public void onDeviceProfileUpdate(TransportProtos.SessionInfoProto sessionInfo, DeviceProfile deviceProfile) {
String registrationId = lwM2mInMemorySecurityStore.getSessions().entrySet() Set<String> registrationIds = lwM2mInMemorySecurityStore.getSessions().entrySet()
.stream() .stream()
.filter(e -> e.getValue().getDeviceUuid().equals(new UUID(sessionInfo.getDeviceIdMSB(), sessionInfo.getDeviceIdLSB()))) .filter(e -> e.getValue().getProfileUuid().equals(deviceProfile.getUuidId()))
.findFirst() .map(Map.Entry::getKey).sorted().collect(Collectors.toSet());
.map(Map.Entry::getKey) if (registrationIds.size() > 0) {
.orElse(""); this.onDeviceUpdateChangeProfile(registrationIds, deviceProfile);
if (!registrationId.isEmpty()) {
this.onDeviceUpdateChangeProfile(registrationId, deviceProfile);
} }
} }
@ -883,19 +914,23 @@ public class LwM2MTransportService {
LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getSessions().get(registrationId); LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getSessions().get(registrationId);
lwM2MClient.setDeviceName(device.getName()); lwM2MClient.setDeviceName(device.getName());
if (!lwM2MClient.getProfileUuid().equals(device.getDeviceProfileId().getId())) { if (!lwM2MClient.getProfileUuid().equals(device.getDeviceProfileId().getId())) {
deviceProfileOpt.ifPresent(deviceProfile -> this.onDeviceUpdateChangeProfile(registrationId, deviceProfile)); Set<String> registrationIds = new HashSet<>();
registrationIds.add(registrationId);
deviceProfileOpt.ifPresent(deviceProfile -> this.onDeviceUpdateChangeProfile(registrationIds, deviceProfile));
} }
lwM2MClient.setProfileUuid(device.getDeviceProfileId().getId());
} }
/** /**
* #1 Read new, old Value (Attribute, Telemetry, Observe, KeyName) * #1 Read new, old Value (Attribute, Telemetry, Observe, KeyName)
* #2 Update in lwM2MClient: ...Profile * #2 Update in lwM2MClient: ...Profile if changes from update device
* #3 Equivalence test: old <> new Value (Attribute, Telemetry, Observe, KeyName) * #3 Equivalence test: old <> new Value (Attribute, Telemetry, Observe, KeyName)
* #3.1 Attribute isChange (add&del) * #3.1 Attribute isChange (add&del)
* #3.2 Telemetry isChange (add&del) * #3.2 Telemetry isChange (add&del)
* #3.3 KeyName isChange (add) * #3.3 KeyName isChange (add)
* #4 update * #4 update
* #4.1 add If #3 isChange, then analyze and update Value in Transport form Client and sent Value ti thingsboard * #4.1 add If #3 isChange, then analyze and update Value in Transport form Client and sent Value to thingsboard
* #4.2 del * #4.2 del
* -- if add attributes includes del telemetry - result del for observe * -- if add attributes includes del telemetry - result del for observe
* #5 * #5
@ -905,15 +940,14 @@ public class LwM2MTransportService {
* #5.3 Observe.del * #5.3 Observe.del
* -- different between newObserve and oldObserve: sent Request cancel observe to client * -- different between newObserve and oldObserve: sent Request cancel observe to client
* *
* @param registrationId - * @param registrationIds -
* @param deviceProfile - * @param deviceProfile -
*/ */
public void onDeviceUpdateChangeProfile(String registrationId, DeviceProfile deviceProfile) { public void onDeviceUpdateChangeProfile(Set<String> registrationIds, DeviceProfile deviceProfile) {
LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(registrationId);
AttrTelemetryObserveValue attrTelemetryObserveValueOld = lwM2mInMemorySecurityStore.getProfiles().get(lwM2MClient.getProfileUuid()); AttrTelemetryObserveValue attrTelemetryObserveValueOld = lwM2mInMemorySecurityStore.getProfiles().get(deviceProfile.getUuidId());
if (lwM2mInMemorySecurityStore.addUpdateProfileParameters(deviceProfile)) { if (lwM2mInMemorySecurityStore.addUpdateProfileParameters(deviceProfile)) {
LeshanServer lwServer = lwM2MClient.getLwServer();
Registration registration = lwM2mInMemorySecurityStore.getByRegistration(registrationId);
// #1 // #1
JsonArray attributeOld = attrTelemetryObserveValueOld.getPostAttributeProfile(); JsonArray attributeOld = attrTelemetryObserveValueOld.getPostAttributeProfile();
Set attributeSetOld = new Gson().fromJson(attributeOld, Set.class); Set attributeSetOld = new Gson().fromJson(attributeOld, Set.class);
@ -929,9 +963,6 @@ public class LwM2MTransportService {
Set telemetrySetNew = new Gson().fromJson(telemetryNew, Set.class); Set telemetrySetNew = new Gson().fromJson(telemetryNew, Set.class);
JsonArray observeNew = attrTelemetryObserveValueNew.getPostObserveProfile(); JsonArray observeNew = attrTelemetryObserveValueNew.getPostObserveProfile();
JsonObject keyNameNew = attrTelemetryObserveValueNew.getPostKeyNameProfile(); JsonObject keyNameNew = attrTelemetryObserveValueNew.getPostKeyNameProfile();
// #2
lwM2MClient.setDeviceProfileName(deviceProfile.getName());
lwM2MClient.setProfileUuid(deviceProfile.getUuidId());
// #3 // #3
ResultsAnalyzerParameters sentAttrToThingsboard = new ResultsAnalyzerParameters(); ResultsAnalyzerParameters sentAttrToThingsboard = new ResultsAnalyzerParameters();
@ -957,9 +988,15 @@ public class LwM2MTransportService {
// #4.1 add // #4.1 add
if (sentAttrToThingsboard.getPathPostParametersAdd().size() > 0) { if (sentAttrToThingsboard.getPathPostParametersAdd().size() > 0) {
// update value in Resources // update value in Resources
this.updateResourceValueObserve(lwServer, registration, lwM2MClient, sentAttrToThingsboard.getPathPostParametersAdd(), GET_TYPE_OPER_READ); registrationIds.forEach(registrationId -> {
// sent attr/telemetry to tingsboard for new path LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(registrationId);
this.updateAttrTelemetry(registration, false, sentAttrToThingsboard.getPathPostParametersAdd()); LeshanServer lwServer = lwM2MClient.getLwServer();
Registration registration = lwM2mInMemorySecurityStore.getByRegistration(registrationId);
log.warn("[{}] # 4.1", registration.getEndpoint());
this.updateResourceValueObserve(lwServer, registration, lwM2MClient, sentAttrToThingsboard.getPathPostParametersAdd(), GET_TYPE_OPER_READ);
// sent attr/telemetry to tingsboard for new path
this.updateAttrTelemetry(registration, false, sentAttrToThingsboard.getPathPostParametersAdd());
});
} }
// #4.2 del // #4.2 del
if (sentAttrToThingsboard.getPathPostParametersDel().size() > 0) { if (sentAttrToThingsboard.getPathPostParametersDel().size() > 0) {
@ -980,10 +1017,16 @@ public class LwM2MTransportService {
// does not include oldObserve // does not include oldObserve
ResultsAnalyzerParameters postObserveAnalyzer = this.getAnalyzerParameters(sentObserveToClientOld.getPathPostParametersAdd(), sentObserveToClientNew.getPathPostParametersAdd()); ResultsAnalyzerParameters postObserveAnalyzer = this.getAnalyzerParameters(sentObserveToClientOld.getPathPostParametersAdd(), sentObserveToClientNew.getPathPostParametersAdd());
// sent Request observe to Client // sent Request observe to Client
this.updateResourceValueObserve(lwServer, registration, lwM2MClient, postObserveAnalyzer.getPathPostParametersAdd(), GET_TYPE_OPER_OBSERVE); registrationIds.forEach(registrationId -> {
// 5.3 del LwM2MClient lwM2MClient = lwM2mInMemorySecurityStore.getLwM2MClient(registrationId);
// sent Request cancel observe to Client LeshanServer lwServer = lwM2MClient.getLwServer();
this.cancelObserveIsValue(lwServer, registration, postObserveAnalyzer.getPathPostParametersDel()); Registration registration = lwM2mInMemorySecurityStore.getByRegistration(registrationId);
log.warn("[{}] # 5.1", registration.getEndpoint());
this.updateResourceValueObserve(lwServer, registration, lwM2MClient, postObserveAnalyzer.getPathPostParametersAdd(), GET_TYPE_OPER_OBSERVE);
// 5.3 del
// sent Request cancel observe to Client
this.cancelObserveIsValue(lwServer, registration, postObserveAnalyzer.getPathPostParametersDel());
});
} }
} }
} }

38
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/ResultIds.java
View File

@ -1,38 +0,0 @@
/**
* Copyright © 2016-2020 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.lwm2m.server;
import lombok.Builder;
import lombok.Data;
@Data
public class ResultIds {
@Builder.Default
int objectId = -1;
@Builder.Default
int instanceId = -1;
@Builder.Default
int resourceId = -1;
public ResultIds (String path) {
String[] paths = path.split("/");
if (paths != null && paths.length > 1) {
this.objectId = (paths.length > 1) ? Integer.parseInt(paths[1]) : this.objectId;
this.instanceId = (paths.length > 2) ? Integer.parseInt(paths[2]) : this.instanceId;
this.resourceId = (paths.length > 3) ? Integer.parseInt(paths[3]) : this.resourceId;
}
}
}

6
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/secure/LwM2mInMemorySecurityStore.java
View File

@ -173,16 +173,16 @@ public class LwM2mInMemorySecurityStore extends InMemorySecurityStore {
private SecurityInfo add(String identity) { private SecurityInfo add(String identity) {
ReadResultSecurityStore store = lwM2MGetSecurityInfo.getSecurityInfo(identity, TypeServer.CLIENT); ReadResultSecurityStore store = lwM2MGetSecurityInfo.getSecurityInfo(identity, TypeServer.CLIENT);
UUID profileUuid = (addUpdateProfileParameters(store.getDeviceProfile())) ? store.getDeviceProfile().getUuidId() : null; UUID profileUuid = (store.getDeviceProfile() != null && addUpdateProfileParameters(store.getDeviceProfile())) ? store.getDeviceProfile().getUuidId() : null;
if (store.getSecurityInfo() != null) { if (store.getSecurityInfo() != null) {
if (store.getSecurityMode() < DEFAULT_MODE.code) { if (store.getSecurityMode() < DEFAULT_MODE.code) {
String endpoint = store.getSecurityInfo().getEndpoint(); String endpoint = store.getSecurityInfo().getEndpoint();
sessions.put(endpoint, new LwM2MClient(endpoint, store.getSecurityInfo().getIdentity(), store.getSecurityInfo(), store.getMsg(), null, null, profileUuid)); sessions.put(endpoint, new LwM2MClient(endpoint, store.getSecurityInfo().getIdentity(), store.getSecurityInfo(), store.getMsg(), null, null, profileUuid));
} }
} else { } else {
if (store.getSecurityMode() == NO_SEC.code) if (store.getSecurityMode() == NO_SEC.code && profileUuid != null)
sessions.put(identity, new LwM2MClient(identity, null, null, store.getMsg(), null, null, profileUuid)); sessions.put(identity, new LwM2MClient(identity, null, null, store.getMsg(), null, null, profileUuid));
else log.error("Registration failed: FORBIDDEN, endpointId: [{}]", identity); else log.error("Registration failed: FORBIDDEN/profileUuid [{}] , endpointId: [{}]", profileUuid, identity);
} }
return store.getSecurityInfo(); return store.getSecurityInfo();
} }

2
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportContext.java
View File

@ -43,8 +43,10 @@ public abstract class TransportContext {
@Autowired @Autowired
private TransportService transportService; private TransportService transportService;
@Autowired @Autowired
private TbServiceInfoProvider serviceInfoProvider; private TbServiceInfoProvider serviceInfoProvider;
@Autowired @Autowired
private SchedulerComponent scheduler; private SchedulerComponent scheduler;