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

fixed NPE

Browse Source
This commit is contained in:
YevhenBondarenko 2021-11-19 17:54:16 +02:00
parent 77b9a8c1af
commit a9151b5100
2 changed files with 83 additions and 83 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/client/LwM2mClientContextImpl.java
View File

@ -278,8 +278,8 @@ public class LwM2mClientContextImpl implements LwM2mClientContext {
public LwM2mClient getClientBySessionInfo(TransportProtos.SessionInfoProto sessionInfo) { public LwM2mClient getClientBySessionInfo(TransportProtos.SessionInfoProto sessionInfo) {
LwM2mClient lwM2mClient = null; LwM2mClient lwM2mClient = null;
UUID sessionId = new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()); UUID sessionId = new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB());
Predicate<LwM2mClient> isClientFilter = c -> Predicate<LwM2mClient> isClientFilter =
sessionId.equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB()))); c -> c.getSession() != null && sessionId.equals((new UUID(c.getSession().getSessionIdMSB(), c.getSession().getSessionIdLSB())));
if (this.lwM2mClientsByEndpoint.size() > 0) { if (this.lwM2mClientsByEndpoint.size() > 0) {
lwM2mClient = this.lwM2mClientsByEndpoint.values().stream().filter(isClientFilter).findAny().orElse(null); lwM2mClient = this.lwM2mClientsByEndpoint.values().stream().filter(isClientFilter).findAny().orElse(null);
} }

162
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java
View File

@ -17,15 +17,12 @@ package org.thingsboard.server.transport.lwm2m.server.rpc;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.eclipse.leshan.core.ResponseCode; import org.eclipse.leshan.core.ResponseCode;
import org.eclipse.leshan.core.request.ReadCompositeRequest;
import org.eclipse.leshan.core.response.ReadCompositeResponse;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportService;
import org.thingsboard.server.common.transport.TransportServiceCallback;
import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.util.TbLwM2mTransportComponent; import org.thingsboard.server.queue.util.TbLwM2mTransportComponent;
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig; import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig;
@ -63,11 +60,9 @@ import org.thingsboard.server.transport.lwm2m.server.rpc.composite.RpcReadRespon
import org.thingsboard.server.transport.lwm2m.server.rpc.composite.RpcWriteCompositeRequest; import org.thingsboard.server.transport.lwm2m.server.rpc.composite.RpcWriteCompositeRequest;
import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler; import org.thingsboard.server.transport.lwm2m.server.uplink.LwM2mUplinkMsgHandler;
import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@Slf4j @Slf4j
@Service @Service
@ -85,91 +80,96 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
@Override @Override
public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) { public void onToDeviceRpcRequest(TransportProtos.ToDeviceRpcRequestMsg rpcRequest, TransportProtos.SessionInfoProto sessionInfo) {
log.debug("Received params: {}", rpcRequest.getParams()); log.debug("Received params: {}", rpcRequest.getParams());
LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName());
if (operationType == null) {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED, "Unsupported operation type: " + rpcRequest.getMethodName());
return;
}
LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo);
if (client.getRegistration() == null) {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
return;
}
UUID rpcId = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB());
if (rpcId.equals(client.getLastSentRpcId())) {
log.debug("[{}]][{}] Rpc has already sent!", client.getEndpoint(), rpcId);
return;
}
try { try {
if (operationType.isHasObjectId()) { LwM2mOperationType operationType = LwM2mOperationType.fromType(rpcRequest.getMethodName());
String objectId = getIdFromParameters(client, rpcRequest); if (operationType == null) {
switch (operationType) { this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.METHOD_NOT_ALLOWED, "Unsupported operation type: " + rpcRequest.getMethodName());
case READ: return;
sendReadRequest(client, rpcRequest, objectId); }
break; LwM2mClient client = clientContext.getClientBySessionInfo(sessionInfo);
case OBSERVE: if (client.getRegistration() == null) {
sendObserveRequest(client, rpcRequest, objectId); this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, "Registration is empty");
break; return;
case DISCOVER: }
sendDiscoverRequest(client, rpcRequest, objectId); UUID rpcId = new UUID(rpcRequest.getRequestIdMSB(), rpcRequest.getRequestIdLSB());
break;
case EXECUTE: if (rpcId.equals(client.getLastSentRpcId())) {
sendExecuteRequest(client, rpcRequest, objectId); log.debug("[{}]][{}] Rpc has already sent!", client.getEndpoint(), rpcId);
break; return;
case WRITE_ATTRIBUTES: }
sendWriteAttributesRequest(client, rpcRequest, objectId); try {
break; if (operationType.isHasObjectId()) {
case OBSERVE_CANCEL: String objectId = getIdFromParameters(client, rpcRequest);
sendCancelObserveRequest(client, rpcRequest, objectId);
break;
case DELETE:
sendDeleteRequest(client, rpcRequest, objectId);
break;
case WRITE_UPDATE:
sendWriteUpdateRequest(client, rpcRequest, objectId);
break;
case WRITE_REPLACE:
sendWriteReplaceRequest(client, rpcRequest, objectId);
break;
default:
throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
}
} else if (operationType.isComposite()) {
if (clientContext.isComposite(client)) {
switch (operationType) { switch (operationType) {
case READ_COMPOSITE: case READ:
sendReadCompositeRequest(client, rpcRequest); sendReadRequest(client, rpcRequest, objectId);
break; break;
case WRITE_COMPOSITE: case OBSERVE:
sendWriteCompositeRequest(client, rpcRequest); sendObserveRequest(client, rpcRequest, objectId);
break;
case DISCOVER:
sendDiscoverRequest(client, rpcRequest, objectId);
break;
case EXECUTE:
sendExecuteRequest(client, rpcRequest, objectId);
break;
case WRITE_ATTRIBUTES:
sendWriteAttributesRequest(client, rpcRequest, objectId);
break;
case OBSERVE_CANCEL:
sendCancelObserveRequest(client, rpcRequest, objectId);
break;
case DELETE:
sendDeleteRequest(client, rpcRequest, objectId);
break;
case WRITE_UPDATE:
sendWriteUpdateRequest(client, rpcRequest, objectId);
break;
case WRITE_REPLACE:
sendWriteReplaceRequest(client, rpcRequest, objectId);
break; break;
default: default:
throw new IllegalArgumentException("Unsupported operation: " + operationType.name()); throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
} }
} else if (operationType.isComposite()) {
if (clientContext.isComposite(client)) {
switch (operationType) {
case READ_COMPOSITE:
sendReadCompositeRequest(client, rpcRequest);
break;
case WRITE_COMPOSITE:
sendWriteCompositeRequest(client, rpcRequest);
break;
default:
throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
}
} else {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(),
ResponseCode.INTERNAL_SERVER_ERROR, "This device does not support Composite Operation");
}
} else { } else {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), switch (operationType) {
ResponseCode.INTERNAL_SERVER_ERROR, "This device does not support Composite Operation"); case OBSERVE_CANCEL_ALL:
} sendCancelAllObserveRequest(client, rpcRequest);
} else { break;
switch (operationType) { case OBSERVE_READ_ALL:
case OBSERVE_CANCEL_ALL: sendObserveAllRequest(client, rpcRequest);
sendCancelAllObserveRequest(client, rpcRequest); break;
break; case DISCOVER_ALL:
case OBSERVE_READ_ALL: sendDiscoverAllRequest(client, rpcRequest);
sendObserveAllRequest(client, rpcRequest); break;
break; case FW_UPDATE:
case DISCOVER_ALL: //TODO: implement and add break statement
sendDiscoverAllRequest(client, rpcRequest); default:
break; throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
case FW_UPDATE: }
//TODO: implement and add break statement
default:
throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
} }
} catch (IllegalArgumentException e) {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage());
} }
} catch (IllegalArgumentException e) { } catch (Exception e) {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage()); log.error("[{}] Failed to send RPC: [{}]", sessionInfo, rpcRequest, e);
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.INTERNAL_SERVER_ERROR, ExceptionUtils.getRootCauseMessage(e));
} }
} }