简柏林 / thingskit · Commits

Commit 373f2f9f0e6e0b86d8e3f7ebeb7b9f6c7855c8d8

Authored by YevhenBondarenko
Committed by Andrew Shvayka
1 parent 9d651a72

added additionalInfo to the rpc

Inline Side-by-side
Showing 14 changed files with 28 additions and 2 deletions
@@ -209,6 +209,7 @@ CREATE TABLE IF NOT EXISTS rpc ( @@ -209,6 +209,7 @@ CREATE TABLE IF NOT EXISTS rpc (
209 expiration_time bigint NOT NULL, 209 expiration_time bigint NOT NULL,
210 request varchar(10000000) NOT NULL, 210 request varchar(10000000) NOT NULL,
211 response varchar(10000000), 211 response varchar(10000000),
  212 + additional_info varchar(10000000),
212 status varchar(255) NOT NULL 213 status varchar(255) NOT NULL
213 ); 214 );
214 215
@@ -234,6 +234,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { @@ -234,6 +234,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
234 rpc.setExpirationTime(request.getExpirationTime()); 234 rpc.setExpirationTime(request.getExpirationTime());
235 rpc.setRequest(JacksonUtil.valueToTree(request)); 235 rpc.setRequest(JacksonUtil.valueToTree(request));
236 rpc.setStatus(status); 236 rpc.setStatus(status);
  237 + rpc.setAdditionalInfo(JacksonUtil.valueToTree(request.getAdditionalInfo()));
237 return systemContext.getTbRpcService().save(tenantId, rpc); 238 return systemContext.getTbRpcService().save(tenantId, rpc);
238 } 239 }
239 240
@@ -90,6 +90,7 @@ public abstract class AbstractRpcController extends BaseController { @@ -90,6 +90,7 @@ public abstract class AbstractRpcController extends BaseController {
90 long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout); 90 long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout);
91 UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID(); 91 UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID();
92 boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean(); 92 boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean();
  93 + String additionalInfo = JacksonUtil.toString(rpcRequestBody.get(DataConstants.ADDITIONAL_INFO));
93 accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<>() { 94 accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<>() {
94 @Override 95 @Override
95 public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) { 96 public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
@@ -99,7 +100,8 @@ public abstract class AbstractRpcController extends BaseController { @@ -99,7 +100,8 @@ public abstract class AbstractRpcController extends BaseController {
99 oneWay, 100 oneWay,
100 expTime, 101 expTime,
101 body, 102 body,
102 - persisted 103 + persisted,
  104 + additionalInfo
103 ); 105 );
104 deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse, timeoutStatus, noActiveConnectionStatus), currentUser); 106 deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse, timeoutStatus, noActiveConnectionStatus), currentUser);
105 } 107 }
@@ -168,6 +168,8 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService { @@ -168,6 +168,8 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
168 entityNode.put("method", msg.getBody().getMethod()); 168 entityNode.put("method", msg.getBody().getMethod());
169 entityNode.put("params", msg.getBody().getParams()); 169 entityNode.put("params", msg.getBody().getParams());
170 170
  171 + entityNode.put(DataConstants.ADDITIONAL_INFO, msg.getAdditionalInfo());
  172 +
171 try { 173 try {
172 TbMsg tbMsg = TbMsg.newMsg(DataConstants.RPC_CALL_FROM_SERVER_TO_DEVICE, msg.getDeviceId(), currentUser.getCustomerId(), metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode)); 174 TbMsg tbMsg = TbMsg.newMsg(DataConstants.RPC_CALL_FROM_SERVER_TO_DEVICE, msg.getDeviceId(), currentUser.getCustomerId(), metaData, TbMsgDataType.JSON, json.writeValueAsString(entityNode));
173 clusterService.pushMsgToRuleEngine(msg.getTenantId(), msg.getDeviceId(), tbMsg, null); 175 clusterService.pushMsgToRuleEngine(msg.getTenantId(), msg.getDeviceId(), tbMsg, null);
@@ -100,7 +100,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi @@ -100,7 +100,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
100 @Override 100 @Override
101 public void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest src, Consumer<RuleEngineDeviceRpcResponse> consumer) { 101 public void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest src, Consumer<RuleEngineDeviceRpcResponse> consumer) {
102 ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(), 102 ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(),
103 - src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted()); 103 + src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getAdditionalInfo());
104 forwardRpcRequestToDeviceActor(request, response -> { 104 forwardRpcRequestToDeviceActor(request, response -> {
105 if (src.isRestApiCall()) { 105 if (src.isRestApiCall()) {
106 sendRpcResponseToTbCore(src.getOriginServiceId(), response); 106 sendRpcResponseToTbCore(src.getOriginServiceId(), response);
@@ -36,6 +36,7 @@ public class DataConstants { @@ -36,6 +36,7 @@ public class DataConstants {
36 public static final String ALARM_CONDITION_REPEATS = "alarmConditionRepeats"; 36 public static final String ALARM_CONDITION_REPEATS = "alarmConditionRepeats";
37 public static final String ALARM_CONDITION_DURATION = "alarmConditionDuration"; 37 public static final String ALARM_CONDITION_DURATION = "alarmConditionDuration";
38 public static final String PERSISTENT = "persistent"; 38 public static final String PERSISTENT = "persistent";
  39 + public static final String ADDITIONAL_INFO = "additionalInfo";
39 public static final String COAP_TRANSPORT_NAME = "COAP"; 40 public static final String COAP_TRANSPORT_NAME = "COAP";
40 public static final String LWM2M_TRANSPORT_NAME = "LWM2M"; 41 public static final String LWM2M_TRANSPORT_NAME = "LWM2M";
41 public static final String MQTT_TRANSPORT_NAME = "MQTT"; 42 public static final String MQTT_TRANSPORT_NAME = "MQTT";
@@ -33,6 +33,7 @@ public class Rpc extends BaseData<RpcId> implements HasTenantId { @@ -33,6 +33,7 @@ public class Rpc extends BaseData<RpcId> implements HasTenantId {
33 private JsonNode request; 33 private JsonNode request;
34 private JsonNode response; 34 private JsonNode response;
35 private RpcStatus status; 35 private RpcStatus status;
  36 + private JsonNode additionalInfo;
36 37
37 public Rpc() { 38 public Rpc() {
38 super(); 39 super();
@@ -50,5 +51,6 @@ public class Rpc extends BaseData<RpcId> implements HasTenantId { @@ -50,5 +51,6 @@ public class Rpc extends BaseData<RpcId> implements HasTenantId {
50 this.request = rpc.getRequest(); 51 this.request = rpc.getRequest();
51 this.response = rpc.getResponse(); 52 this.response = rpc.getResponse();
52 this.status = rpc.getStatus(); 53 this.status = rpc.getStatus();
  54 + this.additionalInfo = rpc.getAdditionalInfo();
53 } 55 }
54 } 56 }
@@ -15,6 +15,7 @@ @@ -15,6 +15,7 @@
15 */ 15 */
16 package org.thingsboard.server.common.msg.rpc; 16 package org.thingsboard.server.common.msg.rpc;
17 17
  18 +import com.fasterxml.jackson.annotation.JsonIgnore;
18 import lombok.Data; 19 import lombok.Data;
19 import org.thingsboard.server.common.data.id.DeviceId; 20 import org.thingsboard.server.common.data.id.DeviceId;
20 import org.thingsboard.server.common.data.id.TenantId; 21 import org.thingsboard.server.common.data.id.TenantId;
@@ -35,5 +36,7 @@ public class ToDeviceRpcRequest implements Serializable { @@ -35,5 +36,7 @@ public class ToDeviceRpcRequest implements Serializable {
35 private final long expirationTime; 36 private final long expirationTime;
36 private final ToDeviceRpcRequestBody body; 37 private final ToDeviceRpcRequestBody body;
37 private final boolean persisted; 38 private final boolean persisted;
  39 + @JsonIgnore
  40 + private final String additionalInfo;
38 } 41 }
39 42
@@ -519,6 +519,7 @@ public class ModelConstants { @@ -519,6 +519,7 @@ public class ModelConstants {
519 public static final String RPC_REQUEST = "request"; 519 public static final String RPC_REQUEST = "request";
520 public static final String RPC_RESPONSE = "response"; 520 public static final String RPC_RESPONSE = "response";
521 public static final String RPC_STATUS = "status"; 521 public static final String RPC_STATUS = "status";
  522 + public static final String RPC_ADDITIONAL_INFO = ADDITIONAL_INFO_PROPERTY;
522 523
523 /** 524 /**
524 * Edge constants. 525 * Edge constants.
@@ -36,6 +36,7 @@ import javax.persistence.Enumerated; @@ -36,6 +36,7 @@ import javax.persistence.Enumerated;
36 import javax.persistence.Table; 36 import javax.persistence.Table;
37 import java.util.UUID; 37 import java.util.UUID;
38 38
  39 +import static org.thingsboard.server.dao.model.ModelConstants.RPC_ADDITIONAL_INFO;
39 import static org.thingsboard.server.dao.model.ModelConstants.RPC_DEVICE_ID; 40 import static org.thingsboard.server.dao.model.ModelConstants.RPC_DEVICE_ID;
40 import static org.thingsboard.server.dao.model.ModelConstants.RPC_EXPIRATION_TIME; 41 import static org.thingsboard.server.dao.model.ModelConstants.RPC_EXPIRATION_TIME;
41 import static org.thingsboard.server.dao.model.ModelConstants.RPC_REQUEST; 42 import static org.thingsboard.server.dao.model.ModelConstants.RPC_REQUEST;
@@ -72,6 +73,10 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> { @@ -72,6 +73,10 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> {
72 @Column(name = RPC_STATUS) 73 @Column(name = RPC_STATUS)
73 private RpcStatus status; 74 private RpcStatus status;
74 75
  76 + @Type(type = "json")
  77 + @Column(name = RPC_ADDITIONAL_INFO)
  78 + private JsonNode additionalInfo;
  79 +
75 public RpcEntity() { 80 public RpcEntity() {
76 super(); 81 super();
77 } 82 }
@@ -85,6 +90,7 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> { @@ -85,6 +90,7 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> {
85 this.request = rpc.getRequest(); 90 this.request = rpc.getRequest();
86 this.response = rpc.getResponse(); 91 this.response = rpc.getResponse();
87 this.status = rpc.getStatus(); 92 this.status = rpc.getStatus();
  93 + this.additionalInfo = rpc.getAdditionalInfo();
88 } 94 }
89 95
90 @Override 96 @Override
@@ -97,6 +103,7 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> { @@ -97,6 +103,7 @@ public class RpcEntity extends BaseSqlEntity<Rpc> implements BaseEntity<Rpc> {
97 rpc.setRequest(request); 103 rpc.setRequest(request);
98 rpc.setResponse(response); 104 rpc.setResponse(response);
99 rpc.setStatus(status); 105 rpc.setStatus(status);
  106 + rpc.setAdditionalInfo(additionalInfo);
100 return rpc; 107 return rpc;
101 } 108 }
102 } 109 }
@@ -582,5 +582,6 @@ CREATE TABLE IF NOT EXISTS rpc ( @@ -582,5 +582,6 @@ CREATE TABLE IF NOT EXISTS rpc (
582 expiration_time bigint NOT NULL, 582 expiration_time bigint NOT NULL,
583 request varchar(10000000) NOT NULL, 583 request varchar(10000000) NOT NULL,
584 response varchar(10000000), 584 response varchar(10000000),
  585 + additional_info varchar(10000000),
585 status varchar(255) NOT NULL 586 status varchar(255) NOT NULL
586 ); 587 );
@@ -616,6 +616,7 @@ CREATE TABLE IF NOT EXISTS rpc ( @@ -616,6 +616,7 @@ CREATE TABLE IF NOT EXISTS rpc (
616 expiration_time bigint NOT NULL, 616 expiration_time bigint NOT NULL,
617 request varchar(10000000) NOT NULL, 617 request varchar(10000000) NOT NULL,
618 response varchar(10000000), 618 response varchar(10000000),
  619 + additional_info varchar(10000000),
619 status varchar(255) NOT NULL 620 status varchar(255) NOT NULL
620 ); 621 );
621 622
@@ -40,5 +40,6 @@ public final class RuleEngineDeviceRpcRequest { @@ -40,5 +40,6 @@ public final class RuleEngineDeviceRpcRequest {
40 private final String body; 40 private final String body;
41 private final long expirationTime; 41 private final long expirationTime;
42 private final boolean restApiCall; 42 private final boolean restApiCall;
  43 + private final String additionalInfo;
43 44
44 } 45 }
@@ -100,6 +100,8 @@ public class TbSendRPCRequestNode implements TbNode { @@ -100,6 +100,8 @@ public class TbSendRPCRequestNode implements TbNode {
100 params = gson.toJson(paramsEl); 100 params = gson.toJson(paramsEl);
101 } 101 }
102 102
  103 + String additionalInfo = gson.toJson(json.get(DataConstants.ADDITIONAL_INFO));
  104 +
103 RuleEngineDeviceRpcRequest request = RuleEngineDeviceRpcRequest.builder() 105 RuleEngineDeviceRpcRequest request = RuleEngineDeviceRpcRequest.builder()
104 .oneway(oneway) 106 .oneway(oneway)
105 .method(json.get("method").getAsString()) 107 .method(json.get("method").getAsString())
@@ -112,6 +114,7 @@ public class TbSendRPCRequestNode implements TbNode { @@ -112,6 +114,7 @@ public class TbSendRPCRequestNode implements TbNode {
112 .expirationTime(expirationTime) 114 .expirationTime(expirationTime)
113 .restApiCall(restApiCall) 115 .restApiCall(restApiCall)
114 .persisted(persisted) 116 .persisted(persisted)
  117 + .additionalInfo(additionalInfo)
115 .build(); 118 .build();
116 119
117 ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> { 120 ctx.getRpcService().sendRpcRequestToDevice(request, ruleEngineDeviceRpcResponse -> {