Browse Source

Merge pull request #5154 from thingsboard/master

Merge master to Develop 3.3.2
pull/5169/head
Andrew Shvayka 5 years ago
committed by GitHub
parent
commit
3d2e86b62f
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 18
      application/src/main/data/json/system/widget_bundles/cards.json
  2. 10
      application/src/main/data/json/system/widget_bundles/maps.json
  3. 9
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  4. 174
      application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
  5. 2
      application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java
  6. 6
      application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
  7. 50
      application/src/main/java/org/thingsboard/server/controller/EdgeController.java
  8. 29
      application/src/main/java/org/thingsboard/server/controller/QueueController.java
  9. 3
      application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java
  10. 5
      application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java
  11. 2
      application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
  12. 7
      application/src/main/resources/thingsboard.yml
  13. 8
      application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java
  14. 9
      application/src/test/java/org/thingsboard/server/transport/coap/telemetry/attributes/AbstractCoapAttributesIntegrationTest.java
  15. 5
      application/src/test/java/org/thingsboard/server/transport/coap/telemetry/attributes/AbstractCoapAttributesProtoIntegrationTest.java
  16. 3
      application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java
  17. 5
      application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java
  18. 66
      application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
  19. 9
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java
  20. 5
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/attributes/AbstractMqttAttributesProtoIntegrationTest.java
  21. 3
      application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java
  22. 3
      application/src/test/resources/application-test.properties
  23. 28
      common/cluster-api/src/main/java/org/thingsboard/server/queue/QueueService.java
  24. 4
      common/cluster-api/src/main/proto/queue.proto
  25. 9
      common/dao-api/pom.xml
  26. 8
      common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java
  27. 5
      common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java
  28. 2
      common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java
  29. 2
      common/data/src/main/java/org/thingsboard/server/common/data/widget/WidgetTypeDetails.java
  30. 2
      common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java
  31. 1
      common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java
  32. 62
      common/queue/src/main/java/org/thingsboard/server/queue/DefaultQueueService.java
  33. 8
      common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java
  34. 2
      common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java
  35. 93
      common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java
  36. 12
      common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java
  37. 3
      common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java
  38. 7
      common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java
  39. 2
      common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2MUplinkMsgHandler.java
  40. 3
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java
  41. 28
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
  42. 17
      common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java
  43. 3
      common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java
  44. 7
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java
  45. 117
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
  46. 42
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionActivityData.java
  47. 7
      common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java
  48. 13
      dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java
  49. 10
      dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
  50. 2
      msa/js-executor/package.json
  51. 2
      msa/tb/pom.xml
  52. 2
      msa/web-ui/package.json
  53. 2
      rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java
  54. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java
  55. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java
  56. 2
      transport/coap/src/main/resources/tb-coap-transport.yml
  57. 2
      transport/http/src/main/resources/tb-http-transport.yml
  58. 2
      transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
  59. 2
      transport/mqtt/src/main/resources/tb-mqtt-transport.yml
  60. 2
      transport/snmp/src/main/resources/tb-snmp-transport.yml
  61. 9
      ui-ngx/angular.json
  62. 4
      ui-ngx/package.json
  63. 1
      ui-ngx/src/app/core/api/widget-api.models.ts
  64. 5
      ui-ngx/src/app/core/api/widget-subscription.ts
  65. 2
      ui-ngx/src/app/core/services/menu.service.ts
  66. 2
      ui-ngx/src/app/modules/home/components/device/device-credentials-mqtt-basic.component.ts
  67. 1
      ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.ts
  68. 6
      ui-ngx/src/app/modules/home/components/profile/device/common/power-mode-setting.component.html
  69. 14
      ui-ngx/src/app/modules/home/components/widget/lib/flot-widget.models.ts
  70. 18
      ui-ngx/src/app/modules/home/components/widget/lib/maps/common-maps-utils.ts
  71. 18
      ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.html
  72. 130
      ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.scss
  73. 116
      ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.ts
  74. 3
      ui-ngx/src/app/modules/home/components/widget/lib/multiple-input-widget.component.ts
  75. 2
      ui-ngx/src/app/modules/home/components/widget/lib/qrcode-widget.component.ts
  76. 7
      ui-ngx/src/app/modules/home/components/widget/widget-components.module.ts
  77. 3
      ui-ngx/src/app/modules/home/components/widget/widget.component.ts
  78. 1
      ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.ts
  79. 11
      ui-ngx/src/app/modules/home/pages/user/user.component.html
  80. 17
      ui-ngx/src/app/modules/home/pages/user/user.component.ts
  81. 9
      ui-ngx/src/app/shared/components/json-form/json-form.component.ts
  82. 4
      ui-ngx/src/app/shared/components/json-form/react/json-form-ace-editor.tsx
  83. 33
      ui-ngx/src/app/shared/components/json-form/react/json-form-markdown.tsx
  84. 4
      ui-ngx/src/app/shared/components/json-form/react/json-form-schema-form.tsx
  85. 2
      ui-ngx/src/app/shared/components/led-light.component.ts
  86. 99
      ui-ngx/src/app/shared/components/markdown.factory.ts
  87. 7
      ui-ngx/src/app/shared/components/queue/queue-type-list.component.html
  88. 52
      ui-ngx/src/app/shared/components/queue/queue-type-list.component.ts
  89. 6
      ui-ngx/src/app/shared/components/time/timewindow-panel.component.scss
  90. 15
      ui-ngx/src/app/shared/models/time/time.models.ts
  91. 1
      ui-ngx/src/app/shared/models/widget.models.ts
  92. 15
      ui-ngx/src/app/shared/shared.module.ts
  93. 10
      ui-ngx/src/assets/locale/locale.constant-en_US.json
  94. 2
      ui-ngx/src/assets/locale/locale.constant-ko_KR.json
  95. 4
      ui-ngx/src/assets/locale/locale.constant-ru_RU.json
  96. 2
      ui-ngx/src/assets/locale/locale.constant-sl_SI.json
  97. 4
      ui-ngx/src/assets/locale/locale.constant-uk_UA.json
  98. 2
      ui-ngx/src/tsconfig.app.json
  99. 275
      ui-ngx/src/typings/leadflet-editable.d.ts
  100. 3
      ui-ngx/tsconfig.json

18
application/src/main/data/json/system/widget_bundles/cards.json

File diff suppressed because one or more lines are too long

10
application/src/main/data/json/system/widget_bundles/maps.json

File diff suppressed because one or more lines are too long

9
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@ -400,6 +400,14 @@ public class ActorSystemContext {
@Getter
private String debugPerTenantLimitsConfiguration;
@Value("${actors.rpc.sequential:false}")
@Getter
private boolean rpcSequential;
@Value("${actors.rpc.max_retries:5}")
@Getter
private int maxRpcRetries;
@Getter
@Setter
private TbActorSystem actorSystem;
@ -478,7 +486,6 @@ public class ActorSystemContext {
return partitionService.resolve(serviceType, queueName, tenantId, entityId);
}
public String getServiceId() {
return serviceInfoProvider.getServiceId();
}

174
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java

@ -48,6 +48,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.page.SortOrder;
import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.RelationTypeGroup;
import org.thingsboard.server.common.data.rpc.Rpc;
@ -79,9 +80,9 @@ import org.thingsboard.server.gen.transport.TransportProtos.SessionType;
import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg;
import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg;
import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionInfoProto;
import org.thingsboard.server.gen.transport.TransportProtos.ToDevicePersistedRpcResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseStatusMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg;
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportUpdateCredentialsProto;
@ -98,9 +99,11 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.UUID;
import java.util.function.Consumer;
@ -119,6 +122,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
private final Map<UUID, SessionInfo> attributeSubscriptions;
private final Map<UUID, SessionInfo> rpcSubscriptions;
private final Map<Integer, ToDeviceRpcRequestMetadata> toDeviceRpcPendingMap;
private final boolean rpcSequential;
private int rpcSeq = 0;
private String deviceName;
@ -130,9 +134,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
super(systemContext);
this.tenantId = tenantId;
this.deviceId = deviceId;
this.rpcSequential = systemContext.isRpcSequential();
this.attributeSubscriptions = new HashMap<>();
this.rpcSubscriptions = new HashMap<>();
this.toDeviceRpcPendingMap = new HashMap<>();
this.toDeviceRpcPendingMap = new LinkedHashMap<>();
this.sessions = new LinkedHashMapRemoveEldest<>(systemContext.getMaxConcurrentSessionsPerDevice(), this::notifyTransportAboutClosedSessionMaxSessionsLimit);
if (initAttributes()) {
restoreSessions();
@ -183,19 +188,19 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
if (timeout <= 0) {
log.debug("[{}][{}] Ignoring message due to exp time reached, {}", deviceId, request.getId(), request.getExpirationTime());
if (persisted) {
createRpc(request, RpcStatus.TIMEOUT);
createRpc(request, RpcStatus.EXPIRED);
}
return;
} else if (persisted) {
createRpc(request, RpcStatus.QUEUED);
}
boolean sent;
boolean sent = false;
if (systemContext.isEdgesEnabled() && edgeId != null) {
log.debug("[{}][{}] device is related to edge [{}]. Saving RPC request to edge queue", tenantId, deviceId, edgeId.getId());
saveRpcRequestToEdgeQueue(request, rpcRequest.getRequestId());
sent = true;
} else {
} else if (isSendNewRpcAvailable()) {
sent = rpcSubscriptions.size() > 0;
Set<UUID> syncSessionSet = new HashSet<>();
rpcSubscriptions.forEach((key, value) -> {
@ -227,6 +232,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
}
}
private boolean isSendNewRpcAvailable() {
return !rpcSequential || toDeviceRpcPendingMap.values().stream().filter(md -> !md.isDelivered()).findAny().isEmpty();
}
private Rpc createRpc(ToDeviceRpcRequest request, RpcStatus status) {
Rpc rpc = new Rpc(new RpcId(request.getId()));
rpc.setCreatedTime(System.currentTimeMillis());
@ -266,16 +275,26 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
void processRemoveRpc(TbActorCtx context, RemoveRpcActorMsg msg) {
log.debug("[{}] Processing remove rpc command", msg.getRequestId());
Integer requestId = null;
for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry : toDeviceRpcPendingMap.entrySet()) {
if (entry.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) {
requestId = entry.getKey();
Map.Entry<Integer, ToDeviceRpcRequestMetadata> entry = null;
for (Map.Entry<Integer, ToDeviceRpcRequestMetadata> e : toDeviceRpcPendingMap.entrySet()) {
if (e.getValue().getMsg().getMsg().getId().equals(msg.getRequestId())) {
entry = e;
break;
}
}
if (requestId != null) {
toDeviceRpcPendingMap.remove(requestId);
if (entry != null) {
if (entry.getValue().isDelivered()) {
toDeviceRpcPendingMap.remove(entry.getKey());
} else {
Optional<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> firstRpc = getFirstRpc();
if (firstRpc.isPresent() && entry.getKey().equals(firstRpc.get().getKey())) {
toDeviceRpcPendingMap.remove(entry.getKey());
sendNextPendingRequest(context);
} else {
toDeviceRpcPendingMap.remove(entry.getKey());
}
}
}
}
@ -290,14 +309,17 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
if (requestMd != null) {
log.debug("[{}] RPC request [{}] timeout detected!", deviceId, msg.getId());
if (requestMd.getMsg().getMsg().isPersisted()) {
systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.TIMEOUT, null);
systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), RpcStatus.EXPIRED, null);
}
systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
null, requestMd.isSent() ? RpcError.TIMEOUT : RpcError.NO_ACTIVE_CONNECTION));
if (!requestMd.isDelivered()) {
sendNextPendingRequest(context);
}
}
}
private void sendPendingRequests(TbActorCtx context, UUID sessionId, SessionInfoProto sessionInfo) {
private void sendPendingRequests(TbActorCtx context, UUID sessionId, String nodeId) {
SessionType sessionType = getSessionType(sessionId);
if (!toDeviceRpcPendingMap.isEmpty()) {
log.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId);
@ -309,20 +331,33 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
log.debug("[{}] No pending RPC messages for new async session [{}]", deviceId, sessionId);
}
Set<Integer> sentOneWayIds = new HashSet<>();
if (sessionType == SessionType.ASYNC) {
toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds));
if (rpcSequential) {
getFirstRpc().ifPresent(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
} else if (sessionType == SessionType.ASYNC) {
toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
} else {
toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds));
toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, nodeId, sentOneWayIds));
}
sentOneWayIds.stream().filter(id -> !toDeviceRpcPendingMap.get(id).getMsg().getMsg().isPersisted()).forEach(toDeviceRpcPendingMap::remove);
}
private Optional<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> getFirstRpc() {
return toDeviceRpcPendingMap.entrySet().stream().filter(e -> !e.getValue().isDelivered()).findFirst();
}
private void sendNextPendingRequest(TbActorCtx context) {
if (rpcSequential) {
rpcSubscriptions.forEach((id, s) -> sendPendingRequests(context, id, s.getNodeId()));
}
}
private Consumer<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> processPendingRpc(TbActorCtx context, UUID sessionId, String nodeId, Set<Integer> sentOneWayIds) {
return entry -> {
ToDeviceRpcRequest request = entry.getValue().getMsg().getMsg();
ToDeviceRpcRequestBody body = request.getBody();
if (request.isOneway()) {
if (request.isOneway() && !rpcSequential) {
sentOneWayIds.add(entry.getKey());
systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(request.getId(), null, null));
}
@ -355,7 +390,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
processSubscriptionCommands(context, sessionInfo, msg.getSubscribeToRPC());
}
if (msg.hasSendPendingRPC()) {
sendPendingRequests(context, getSessionId(sessionInfo), sessionInfo);
sendPendingRequests(context, getSessionId(sessionInfo), sessionInfo.getNodeId());
}
if (msg.hasGetAttributes()) {
handleGetAttributesRequest(context, sessionInfo, msg.getGetAttributes());
@ -369,8 +404,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
if (msg.hasClaimDevice()) {
handleClaimDeviceMsg(context, sessionInfo, msg.getClaimDevice());
}
if (msg.hasPersistedRpcResponseMsg()) {
processPersistedRpcResponses(context, sessionInfo, msg.getPersistedRpcResponseMsg());
if (msg.hasRpcResponseStatusMsg()) {
processRpcResponseStatus(context, sessionInfo, msg.getRpcResponseStatusMsg());
}
if (msg.hasUplinkNotificationMsg()) {
processUplinkNotificationMsg(context, sessionInfo, msg.getUplinkNotificationMsg());
@ -530,38 +565,63 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
boolean success = requestMd != null;
if (success) {
boolean hasError = StringUtils.isNotEmpty(responseMsg.getError());
String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
payload, null));
if (requestMd.getMsg().getMsg().isPersisted()) {
RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
JsonNode response;
try {
response = JacksonUtil.toJsonNode(payload);
} catch (IllegalArgumentException e) {
response = JacksonUtil.newObjectNode().put("error", payload);
try {
String payload = hasError ? responseMsg.getError() : responseMsg.getPayload();
systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(
new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(),
payload, null));
if (requestMd.getMsg().getMsg().isPersisted()) {
RpcStatus status = hasError ? RpcStatus.FAILED : RpcStatus.SUCCESSFUL;
JsonNode response;
try {
response = JacksonUtil.toJsonNode(payload);
} catch (IllegalArgumentException e) {
response = JacksonUtil.newObjectNode().put("error", payload);
}
systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
}
} finally {
if (hasError) {
sendNextPendingRequest(context);
}
systemContext.getTbRpcService().save(tenantId, new RpcId(requestMd.getMsg().getMsg().getId()), status, response);
}
} else {
log.debug("[{}] Rpc command response [{}] is stale!", deviceId, responseMsg.getRequestId());
}
}
private void processPersistedRpcResponses(TbActorCtx context, SessionInfoProto sessionInfo, ToDevicePersistedRpcResponseMsg responseMsg) {
private void processRpcResponseStatus(TbActorCtx context, SessionInfoProto sessionInfo, ToDeviceRpcResponseStatusMsg responseMsg) {
UUID rpcId = new UUID(responseMsg.getRequestIdMSB(), responseMsg.getRequestIdLSB());
RpcStatus status = RpcStatus.valueOf(responseMsg.getStatus());
ToDeviceRpcRequestMetadata md;
if (RpcStatus.DELIVERED.equals(status)) {
md = toDeviceRpcPendingMap.get(responseMsg.getRequestId());
} else {
md = toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
}
ToDeviceRpcRequestMetadata md = toDeviceRpcPendingMap.get(responseMsg.getRequestId());
if (md != null) {
systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
if (status.equals(RpcStatus.DELIVERED)) {
if (md.getMsg().getMsg().isOneway()) {
toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
if (rpcSequential) {
systemContext.getTbCoreDeviceRpcService().processRpcResponseFromDeviceActor(new FromDeviceRpcResponse(rpcId, null, null));
}
} else {
md.setDelivered(true);
}
} else if (status.equals(RpcStatus.TIMEOUT)) {
Integer maxRpcRetries = md.getMsg().getMsg().getRetries();
maxRpcRetries = maxRpcRetries == null ? systemContext.getMaxRpcRetries() : Math.min(maxRpcRetries, systemContext.getMaxRpcRetries());
if (maxRpcRetries <= md.getRetries()) {
toDeviceRpcPendingMap.remove(responseMsg.getRequestId());
status = RpcStatus.FAILED;
} else {
md.setRetries(md.getRetries() + 1);
}
}
if (md.getMsg().getMsg().isPersisted()) {
systemContext.getTbRpcService().save(tenantId, new RpcId(rpcId), status, null);
}
if (status != RpcStatus.SENT) {
sendNextPendingRequest(context);
}
} else {
log.info("[{}][{}] Rpc has already removed from pending map.", deviceId, rpcId);
}
@ -601,7 +661,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
sessionMD.setSubscribedToRPC(true);
log.debug("[{}] Registering rpc subscription for session [{}]", deviceId, sessionId);
rpcSubscriptions.put(sessionId, sessionMD.getSessionInfo());
sendPendingRequests(context, sessionId, sessionInfo);
sendPendingRequests(context, sessionId, sessionInfo.getNodeId());
dumpSessions();
}
}
@ -638,20 +698,22 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
UUID sessionId = getSessionId(sessionInfoProto);
Objects.requireNonNull(sessionId);
SessionInfoMetaData sessionMD = sessions.computeIfAbsent(sessionId,
id -> new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfoProto.getNodeId()), subscriptionInfo.getLastActivityTime()));
sessionMD.setLastActivityTime(subscriptionInfo.getLastActivityTime());
sessionMD.setSubscribedToAttributes(subscriptionInfo.getAttributeSubscription());
sessionMD.setSubscribedToRPC(subscriptionInfo.getRpcSubscription());
if (subscriptionInfo.getAttributeSubscription()) {
attributeSubscriptions.putIfAbsent(sessionId, sessionMD.getSessionInfo());
}
if (subscriptionInfo.getRpcSubscription()) {
rpcSubscriptions.putIfAbsent(sessionId, sessionMD.getSessionInfo());
SessionInfoMetaData sessionMD = sessions.get(sessionId);
if (sessionMD != null) {
sessionMD.setLastActivityTime(subscriptionInfo.getLastActivityTime());
sessionMD.setSubscribedToAttributes(subscriptionInfo.getAttributeSubscription());
sessionMD.setSubscribedToRPC(subscriptionInfo.getRpcSubscription());
if (subscriptionInfo.getAttributeSubscription()) {
attributeSubscriptions.putIfAbsent(sessionId, sessionMD.getSessionInfo());
}
if (subscriptionInfo.getRpcSubscription()) {
rpcSubscriptions.putIfAbsent(sessionId, sessionMD.getSessionInfo());
}
}
systemContext.getDeviceStateService().onDeviceActivity(tenantId, deviceId, subscriptionInfo.getLastActivityTime());
dumpSessions();
if (sessionMD != null) {
dumpSessions();
}
}
void processCredentialsUpdate(TbActorMsg msg) {
@ -869,7 +931,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
void init(TbActorCtx ctx) {
schedulePeriodicMsgWithDelay(ctx, SessionTimeoutCheckMsg.instance(), systemContext.getSessionReportTimeout(), systemContext.getSessionReportTimeout());
PageLink pageLink = new PageLink(1024);
PageLink pageLink = new PageLink(1024, 0, null, new SortOrder("createdTime"));
PageData<Rpc> pageData;
do {
pageData = systemContext.getTbRpcService().findAllByDeviceIdAndStatus(tenantId, deviceId, RpcStatus.QUEUED, pageLink);
@ -877,7 +939,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor {
ToDeviceRpcRequest msg = JacksonUtil.convertValue(rpc.getRequest(), ToDeviceRpcRequest.class);
long timeout = rpc.getExpirationTime() - System.currentTimeMillis();
if (timeout <= 0) {
rpc.setStatus(RpcStatus.TIMEOUT);
rpc.setStatus(RpcStatus.EXPIRED);
systemContext.getTbRpcService().save(tenantId, rpc);
} else {
registerPendingRpcRequest(ctx, new ToDeviceRpcRequestActorMsg(systemContext.getServiceId(), msg), false, creteToDeviceRpcRequestMsg(msg), timeout);

2
application/src/main/java/org/thingsboard/server/actors/device/ToDeviceRpcRequestMetadata.java

@ -25,4 +25,6 @@ import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg;
public class ToDeviceRpcRequestMetadata {
private final ToDeviceRpcRequestActorMsg msg;
private final boolean sent;
private int retries;
private boolean delivered;
}

6
application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java

@ -75,11 +75,12 @@ public abstract class AbstractRpcController extends BaseController {
SecurityUser currentUser = getCurrentUser();
TenantId tenantId = currentUser.getTenantId();
final DeferredResult<ResponseEntity> response = new DeferredResult<>();
long timeout = rpcRequestBody.has("timeout") ? rpcRequestBody.get("timeout").asLong() : defaultTimeout;
long expTime = System.currentTimeMillis() + Math.max(minTimeout, timeout);
long timeout = rpcRequestBody.has(DataConstants.TIMEOUT) ? rpcRequestBody.get(DataConstants.TIMEOUT).asLong() : defaultTimeout;
long expTime = rpcRequestBody.has(DataConstants.EXPIRATION_TIME) ? rpcRequestBody.get(DataConstants.EXPIRATION_TIME).asLong() : System.currentTimeMillis() + Math.max(minTimeout, timeout);
UUID rpcRequestUUID = rpcRequestBody.has("requestUUID") ? UUID.fromString(rpcRequestBody.get("requestUUID").asText()) : UUID.randomUUID();
boolean persisted = rpcRequestBody.has(DataConstants.PERSISTENT) && rpcRequestBody.get(DataConstants.PERSISTENT).asBoolean();
String additionalInfo = JacksonUtil.toString(rpcRequestBody.get(DataConstants.ADDITIONAL_INFO));
Integer retries = rpcRequestBody.has(DataConstants.RETRIES) ? rpcRequestBody.get(DataConstants.RETRIES).asInt() : null;
accessValidator.validate(currentUser, Operation.RPC_CALL, deviceId, new HttpValidationCallback(response, new FutureCallback<>() {
@Override
public void onSuccess(@Nullable DeferredResult<ResponseEntity> result) {
@ -90,6 +91,7 @@ public abstract class AbstractRpcController extends BaseController {
expTime,
body,
persisted,
retries,
additionalInfo
);
deviceRpcService.processRestApiRpcRequest(rpcRequest, fromDeviceRpcResponse -> reply(new LocalRequestMetaData(rpcRequest, currentUser, result), fromDeviceRpcResponse, timeoutStatus, noActiveConnectionStatus), currentUser);

50
application/src/main/java/org/thingsboard/server/controller/EdgeController.java

@ -15,8 +15,11 @@
*/
package org.thingsboard.server.controller;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
@ -59,6 +62,7 @@ import java.util.stream.Collectors;
@RestController
@TbCoreComponent
@Slf4j
@RequestMapping("/api")
public class EdgeController extends BaseController {
@ -544,27 +548,6 @@ public class EdgeController extends BaseController {
}
}
@RequestMapping(value = "/license/checkInstance", method = RequestMethod.POST)
@ResponseBody
public Object checkInstance(@RequestBody Object request) throws ThingsboardException {
try {
return edgeService.checkInstance(request);
} catch (Exception e) {
throw new ThingsboardException(e, ThingsboardErrorCode.SUBSCRIPTION_VIOLATION);
}
}
@RequestMapping(value = "/license/activateInstance", params = {"licenseSecret", "releaseDate"}, method = RequestMethod.POST)
@ResponseBody
public Object activateInstance(@RequestParam String licenseSecret,
@RequestParam String releaseDate) throws ThingsboardException {
try {
return edgeService.activateInstance(licenseSecret, releaseDate);
} catch (Exception e) {
throw new ThingsboardException(e, ThingsboardErrorCode.SUBSCRIPTION_VIOLATION);
}
}
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
@RequestMapping(value = "/edge/missingToRelatedRuleChains/{edgeId}", method = RequestMethod.GET)
@ResponseBody
@ -583,4 +566,29 @@ public class EdgeController extends BaseController {
private void cleanUpLicenseKey(Edge edge) {
edge.setEdgeLicenseKey(null);
}
@RequestMapping(value = "/license/checkInstance", method = RequestMethod.POST)
@ResponseBody
public ResponseEntity<JsonNode> checkInstance(@RequestBody JsonNode request) throws ThingsboardException {
log.debug("Checking instance [{}]", request);
try {
return edgeService.checkInstance(request);
} catch (Exception e) {
log.error("Error occurred: [{}]", e.getMessage(), e);
throw new ThingsboardException(e, ThingsboardErrorCode.SUBSCRIPTION_VIOLATION);
}
}
@RequestMapping(value = "/license/activateInstance", params = {"licenseSecret", "releaseDate"}, method = RequestMethod.POST)
@ResponseBody
public ResponseEntity<JsonNode> activateInstance(@RequestParam String licenseSecret,
@RequestParam String releaseDate) throws ThingsboardException {
log.debug("Activating instance [{}], [{}]", licenseSecret, releaseDate);
try {
return edgeService.activateInstance(licenseSecret, releaseDate);
} catch (Exception e) {
log.error("Error occurred: [{}]", e.getMessage(), e);
throw new ThingsboardException(e, ThingsboardErrorCode.SUBSCRIPTION_VIOLATION);
}
}
}

29
application/src/main/java/org/thingsboard/server/controller/QueueController.java

@ -15,7 +15,7 @@
*/
package org.thingsboard.server.controller;
import org.springframework.beans.factory.annotation.Autowired;
import lombok.RequiredArgsConstructor;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@ -24,41 +24,26 @@ import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import org.thingsboard.server.common.data.exception.ThingsboardException;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
import org.thingsboard.server.queue.QueueService;
import org.thingsboard.server.queue.util.TbCoreComponent;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.Set;
@RestController
@TbCoreComponent
@RequestMapping("/api")
@RequiredArgsConstructor
public class QueueController extends BaseController {
@Autowired(required = false)
private TbQueueRuleEngineSettings ruleEngineSettings;
private final QueueService queueService;
@PreAuthorize("hasAuthority('TENANT_ADMIN')")
@RequestMapping(value = "/tenant/queues", params = {"serviceType"}, method = RequestMethod.GET)
@ResponseBody
public List<String> getTenantQueuesByServiceType(@RequestParam String serviceType) throws ThingsboardException {
public Set<String> getTenantQueuesByServiceType(@RequestParam String serviceType) throws ThingsboardException {
checkParameter("serviceType", serviceType);
try {
ServiceType type = ServiceType.valueOf(serviceType);
switch (type) {
case TB_RULE_ENGINE:
if (ruleEngineSettings == null) {
return Arrays.asList("Main", "HighPriority", "SequentialByOriginator");
}
return ruleEngineSettings.getQueues().stream()
.map(TbRuleEngineQueueConfiguration::getName)
.collect(Collectors.toList());
default:
return Collections.emptyList();
}
return queueService.getQueuesByServiceType(ServiceType.valueOf(serviceType));
} catch (Exception e) {
throw handleException(e);
}

3
application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@ -200,7 +200,8 @@ public class ThingsboardInstallService {
dataUpdateService.updateData("3.2.2");
systemDataLoaderService.createOAuth2Templates();
case "3.3.0":
log.info("Upgrading ThingsBoard from version 3.3.0 to 3.3.1 ...");
log.info("Updating system data...");
systemDataLoaderService.updateSystemWidgets();
break;

5
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbCoreDeviceRpcService.java

@ -166,6 +166,11 @@ public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService {
metaData.putValue("oneway", Boolean.toString(msg.isOneway()));
metaData.putValue(DataConstants.PERSISTENT, Boolean.toString(msg.isPersisted()));
if (msg.getRetries() != null) {
metaData.putValue(DataConstants.RETRIES, msg.getRetries().toString());
}
Device device = deviceService.findDeviceById(msg.getTenantId(), msg.getDeviceId());
if (device != null) {
metaData.putValue("deviceName", device.getName());

2
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java

@ -101,7 +101,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
@Override
public void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest src, Consumer<RuleEngineDeviceRpcResponse> consumer) {
ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(),
src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getAdditionalInfo());
src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getRetries(), src.getAdditionalInfo());
forwardRpcRequestToDeviceActor(request, response -> {
if (src.isRestApiCall()) {
sendRpcResponseToTbCore(src.getOriginServiceId(), response);

7
application/src/main/resources/thingsboard.yml

@ -326,6 +326,9 @@ actors:
queue_size: "${ACTORS_RULE_TRANSACTION_QUEUE_SIZE:15000}"
# Time in milliseconds for transaction to complete
duration: "${ACTORS_RULE_TRANSACTION_DURATION:60000}"
rpc:
max_retries: "${ACTORS_RPC_MAX_RETRIES:5}"
sequential: "${ACTORS_RPC_SEQUENTIAL:false}"
statistics:
# Enable/disable actor statistics
enabled: "${ACTORS_STATISTICS_ENABLED:true}"
@ -563,7 +566,7 @@ js:
transport:
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
@ -699,7 +702,7 @@ transport:
# Edges parameters
edges:
enabled: "${EDGES_ENABLED:false}"
enabled: "${EDGES_ENABLED:true}"
rpc:
port: "${EDGES_RPC_PORT:7070}"
client_max_keep_alive_time_sec: "${EDGES_RPC_CLIENT_MAX_KEEP_ALIVE_TIME_SEC:300}"

8
application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java

@ -21,11 +21,13 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.test.util.ReflectionTestUtils;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.queue.QueueService;
import org.thingsboard.server.queue.discovery.HashPartitionService;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
@ -57,6 +59,7 @@ public class HashPartitionServiceTest {
private TenantRoutingInfoService routingInfoService;
private ApplicationEventPublisher applicationEventPublisher;
private TbQueueRuleEngineSettings ruleEngineSettings;
private QueueService queueService;
private String hashFunctionName = "sha256";
@ -67,10 +70,12 @@ public class HashPartitionServiceTest {
applicationEventPublisher = mock(ApplicationEventPublisher.class);
routingInfoService = mock(TenantRoutingInfoService.class);
ruleEngineSettings = mock(TbQueueRuleEngineSettings.class);
queueService = mock(QueueService.class);
clusterRoutingService = new HashPartitionService(discoveryService,
routingInfoService,
applicationEventPublisher,
ruleEngineSettings
ruleEngineSettings,
queueService
);
when(ruleEngineSettings.getQueues()).thenReturn(Collections.emptyList());
ReflectionTestUtils.setField(clusterRoutingService, "coreTopic", "tb.core");
@ -82,6 +87,7 @@ public class HashPartitionServiceTest {
.setTenantIdLSB(TenantId.NULL_UUID.getLeastSignificantBits())
.addAllServiceTypes(Collections.singletonList(ServiceType.TB_CORE.name()))
.build();
// when(queueService.resolve(Mockito.any(), Mockito.anyString())).thenAnswer(i -> i.getArguments()[1]);
// when(discoveryService.getServiceInfo()).thenReturn(currentServer);
List<TransportProtos.ServiceInfo> otherServers = new ArrayList<>();
for (int i = 1; i < SERVER_COUNT; i++) {

9
application/src/test/java/org/thingsboard/server/transport/coap/telemetry/attributes/AbstractCoapAttributesIntegrationTest.java

@ -154,15 +154,6 @@ public abstract class AbstractCoapAttributesIntegrationTest extends AbstractCoap
case "key1":
assertEquals("", value);
break;
case "key2":
assertEquals(false, value);
break;
case "key3":
assertEquals(0.0, value);
break;
case "key4":
assertEquals(0, value);
break;
case "key5":
assertNotNull(value);
assertEquals(2, ((LinkedHashMap) value).size());

5
application/src/test/java/org/thingsboard/server/transport/coap/telemetry/attributes/AbstractCoapAttributesProtoIntegrationTest.java

@ -123,12 +123,9 @@ public abstract class AbstractCoapAttributesProtoIntegrationTest extends Abstrac
assertNotNull(postAttributesMsgDescriptor);
DynamicMessage postAttributesMsg = postAttributesBuilder
.setField(postAttributesMsgDescriptor.findFieldByName("key1"), "")
.setField(postAttributesMsgDescriptor.findFieldByName("key2"), false)
.setField(postAttributesMsgDescriptor.findFieldByName("key3"), 0.0)
.setField(postAttributesMsgDescriptor.findFieldByName("key4"), 0)
.setField(postAttributesMsgDescriptor.findFieldByName("key5"), jsonObject)
.build();
processAttributesTest(Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), true);
processAttributesTest(Arrays.asList("key1", "key5"), postAttributesMsg.toByteArray(), true);
}
}

3
application/src/test/java/org/thingsboard/server/transport/coap/telemetry/timeseries/AbstractCoapTimeseriesProtoIntegrationTest.java

@ -207,10 +207,9 @@ public abstract class AbstractCoapTimeseriesProtoIntegrationTest extends Abstrac
.setField(postTelemetryMsgDescriptor.findFieldByName("key1"), "")
.setField(postTelemetryMsgDescriptor.findFieldByName("key2"), false)
.setField(postTelemetryMsgDescriptor.findFieldByName("key3"), 0.0)
.setField(postTelemetryMsgDescriptor.findFieldByName("key4"), 0)
.setField(postTelemetryMsgDescriptor.findFieldByName("key5"), jsonObject)
.build();
processTestPostTelemetry(postTelemetryMsg.toByteArray(), Arrays.asList("key1", "key2", "key3", "key4", "key5"), false, true);
processTestPostTelemetry(postTelemetryMsg.toByteArray(), Arrays.asList("key1", "key2", "key3", "key5"), false, true);
}
@Test

5
application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcDefaultIntegrationTest.java

@ -89,6 +89,11 @@ public abstract class AbstractMqttServerSideRpcDefaultIntegrationTest extends Ab
processTwoWayRpcTest();
}
@Test
public void testSequenceServerMqttTwoWayRpc() throws Exception {
processSequenceTwoWayRpcTest();
}
@Test
public void testGatewayServerMqttOneWayRpc() throws Exception {
processOneWayRpcTestGateway("Gateway Device OneWay RPC");

66
application/src/test/java/org/thingsboard/server/transport/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java

@ -16,6 +16,7 @@
package org.thingsboard.server.transport.mqtt.rpc;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nimbusds.jose.util.StandardCharset;
import io.netty.handler.codec.mqtt.MqttQoS;
@ -27,13 +28,16 @@ import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.Assert;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.transport.mqtt.AbstractMqttIntegrationTest;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@ -101,6 +105,32 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
Assert.assertEquals(expected, result);
}
protected void processSequenceTwoWayRpcTest() throws Exception {
List<String> expected = new ArrayList<>();
List<String> result = new ArrayList<>();
String deviceId = savedDevice.getId().getId().toString();
for (int i = 0; i < 10; i++) {
ObjectNode request = JacksonUtil.newObjectNode();
request.put("method", "test");
request.put("params", i);
expected.add(JacksonUtil.toString(request));
request.put("persistent", true);
doPostAsync("/api/rpc/twoway/" + deviceId, JacksonUtil.toString(request), String.class, status().isOk());
}
MqttAsyncClient client = getMqttAsyncClient(accessToken);
client.setManualAcks(true);
CountDownLatch latch = new CountDownLatch(10);
TestSequenceMqttCallback callback = new TestSequenceMqttCallback(client, latch, result);
client.setCallback(callback);
client.subscribe(MqttTopics.DEVICE_RPC_REQUESTS_SUB_TOPIC, 1);
latch.await(10, TimeUnit.SECONDS);
Assert.assertEquals(expected, result);
}
protected void processTwoWayRpcTestGateway(String deviceName) throws Exception {
MqttAsyncClient client = getMqttAsyncClient(gatewayAccessToken);
@ -213,4 +243,38 @@ public abstract class AbstractMqttServerSideRpcIntegrationTest extends AbstractM
}
}
protected class TestSequenceMqttCallback implements MqttCallback {
private final MqttAsyncClient client;
private final CountDownLatch latch;
private final List<String> expected;
TestSequenceMqttCallback(MqttAsyncClient client, CountDownLatch latch, List<String> expected) {
this.client = client;
this.latch = latch;
this.expected = expected;
}
@Override
public void connectionLost(Throwable throwable) {
}
@Override
public void messageArrived(String requestTopic, MqttMessage mqttMessage) throws Exception {
log.info("Message Arrived: " + Arrays.toString(mqttMessage.getPayload()));
expected.add(new String(mqttMessage.getPayload()));
String responseTopic = requestTopic.replace("request", "response");
var qoS = mqttMessage.getQos();
client.messageArrivedComplete(mqttMessage.getId(), qoS);
client.publish(responseTopic, processMessageArrived(requestTopic, mqttMessage));
latch.countDown();
}
@Override
public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
}
}
}

9
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java

@ -192,15 +192,6 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
case "key1":
assertEquals("", value);
break;
case "key2":
assertEquals(false, value);
break;
case "key3":
assertEquals(0.0, value);
break;
case "key4":
assertEquals(0, value);
break;
case "key5":
assertNotNull(value);
assertEquals(2, ((LinkedHashMap) value).size());

5
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/attributes/AbstractMqttAttributesProtoIntegrationTest.java

@ -119,12 +119,9 @@ public abstract class AbstractMqttAttributesProtoIntegrationTest extends Abstrac
assertNotNull(postAttributesMsgDescriptor);
DynamicMessage postAttributesMsg = postAttributesBuilder
.setField(postAttributesMsgDescriptor.findFieldByName("key1"), "")
.setField(postAttributesMsgDescriptor.findFieldByName("key2"), false)
.setField(postAttributesMsgDescriptor.findFieldByName("key3"), 0.0)
.setField(postAttributesMsgDescriptor.findFieldByName("key4"), 0)
.setField(postAttributesMsgDescriptor.findFieldByName("key5"), jsonObject)
.build();
processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postAttributesMsg.toByteArray(), true);
processAttributesTest(POST_DATA_ATTRIBUTES_TOPIC, Arrays.asList("key1", "key5"), postAttributesMsg.toByteArray(), true);
}
@Test

3
application/src/test/java/org/thingsboard/server/transport/mqtt/telemetry/timeseries/AbstractMqttTimeseriesProtoIntegrationTest.java

@ -204,10 +204,9 @@ public abstract class AbstractMqttTimeseriesProtoIntegrationTest extends Abstrac
.setField(postTelemetryMsgDescriptor.findFieldByName("key1"), "")
.setField(postTelemetryMsgDescriptor.findFieldByName("key2"), false)
.setField(postTelemetryMsgDescriptor.findFieldByName("key3"), 0.0)
.setField(postTelemetryMsgDescriptor.findFieldByName("key4"), 0)
.setField(postTelemetryMsgDescriptor.findFieldByName("key5"), jsonObject)
.build();
processTelemetryTest(POST_DATA_TELEMETRY_TOPIC, Arrays.asList("key1", "key2", "key3", "key4", "key5"), postTelemetryMsg.toByteArray(), false, true);
processTelemetryTest(POST_DATA_TELEMETRY_TOPIC, Arrays.asList("key1", "key2", "key3", "key5"), postTelemetryMsg.toByteArray(), false, true);
}
@Test

3
application/src/test/resources/application-test.properties

@ -6,4 +6,5 @@ edges.storage.sleep_between_batches=500
transport.lwm2m.server.security.key_alias=server
transport.lwm2m.server.security.key_password=server
transport.lwm2m.bootstrap.security.key_alias=server
transport.lwm2m.bootstrap.security.key_password=server
transport.lwm2m.bootstrap.security.key_password=server
actors.rpc.sequential=true

28
common/cluster-api/src/main/java/org/thingsboard/server/queue/QueueService.java

@ -0,0 +1,28 @@
/**
* Copyright © 2016-2021 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.queue;
import org.thingsboard.server.common.msg.queue.ServiceType;
import java.util.Set;
public interface QueueService {
Set<String> getQueuesByServiceType(ServiceType serviceType);
String resolve(ServiceType serviceType, String queueName);
}

4
common/cluster-api/src/main/proto/queue.proto

@ -346,7 +346,7 @@ message UplinkNotificationMsg {
int64 uplinkTs = 1;
}
message ToDevicePersistedRpcResponseMsg {
message ToDeviceRpcResponseStatusMsg {
int32 requestId = 1;
int64 requestIdMSB = 2;
int64 requestIdLSB = 3;
@ -456,7 +456,7 @@ message TransportToDeviceActorMsg {
SubscriptionInfoProto subscriptionInfo = 7;
ClaimDeviceMsg claimDevice = 8;
ProvisionDeviceRequestMsg provisionDevice = 9;
ToDevicePersistedRpcResponseMsg persistedRpcResponseMsg = 10;
ToDeviceRpcResponseStatusMsg rpcResponseStatusMsg = 10;
SendPendingRPCMsg sendPendingRPC = 11;
UplinkNotificationMsg uplinkNotificationMsg = 12;
}

9
common/dao-api/pom.xml

@ -40,6 +40,10 @@
<groupId>org.thingsboard.common</groupId>
<artifactId>data</artifactId>
</dependency>
<dependency>
<groupId>org.thingsboard.common</groupId>
<artifactId>cluster-api</artifactId>
</dependency>
<dependency>
<groupId>org.thingsboard.common</groupId>
<artifactId>message</artifactId>
@ -81,6 +85,11 @@
<artifactId>spring-boot-autoconfigure</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.datastax.oss</groupId>
<artifactId>java-driver-core</artifactId>

8
common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java

@ -15,16 +15,16 @@
*/
package org.thingsboard.server.dao.edge;
import com.fasterxml.jackson.databind.JsonNode;
import com.google.common.util.concurrent.ListenableFuture;
import org.springframework.http.ResponseEntity;
import org.thingsboard.server.common.data.EntitySubtype;
import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeInfo;
import org.thingsboard.server.common.data.edge.EdgeSearchQuery;
import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DashboardId;
import org.thingsboard.server.common.data.id.EdgeId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.RuleChainId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
@ -86,9 +86,9 @@ public interface EdgeService {
PageData<EdgeId> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId, PageLink pageLink);
Object checkInstance(Object request);
ResponseEntity<JsonNode> checkInstance(JsonNode request);
Object activateInstance(String licenseSecret, String releaseDate);
ResponseEntity<JsonNode> activateInstance(String licenseSecret, String releaseDate);
String findMissingToRelatedRuleChains(TenantId tenantId, EdgeId edgeId);
}

5
common/data/src/main/java/org/thingsboard/server/common/data/DataConstants.java

@ -36,7 +36,10 @@ public class DataConstants {
public static final String ALARM_CONDITION_REPEATS = "alarmConditionRepeats";
public static final String ALARM_CONDITION_DURATION = "alarmConditionDuration";
public static final String PERSISTENT = "persistent";
public static final String TIMEOUT = "timeout";
public static final String EXPIRATION_TIME = "expirationTime";
public static final String ADDITIONAL_INFO = "additionalInfo";
public static final String RETRIES = "retries";
public static final String COAP_TRANSPORT_NAME = "COAP";
public static final String LWM2M_TRANSPORT_NAME = "LWM2M";
public static final String MQTT_TRANSPORT_NAME = "MQTT";
@ -85,9 +88,11 @@ public class DataConstants {
public static final String RPC_CALL_FROM_SERVER_TO_DEVICE = "RPC_CALL_FROM_SERVER_TO_DEVICE";
public static final String RPC_QUEUED = "RPC_QUEUED";
public static final String RPC_SENT = "RPC_SENT";
public static final String RPC_DELIVERED = "RPC_DELIVERED";
public static final String RPC_SUCCESSFUL = "RPC_SUCCESSFUL";
public static final String RPC_TIMEOUT = "RPC_TIMEOUT";
public static final String RPC_EXPIRED = "RPC_EXPIRED";
public static final String RPC_FAILED = "RPC_FAILED";
public static final String RPC_DELETED = "RPC_DELETED";

2
common/data/src/main/java/org/thingsboard/server/common/data/rpc/RpcStatus.java

@ -16,5 +16,5 @@
package org.thingsboard.server.common.data.rpc;
public enum RpcStatus {
QUEUED, DELIVERED, SUCCESSFUL, TIMEOUT, FAILED
QUEUED, SENT, DELIVERED, SUCCESSFUL, TIMEOUT, EXPIRED, FAILED
}

2
common/data/src/main/java/org/thingsboard/server/common/data/widget/WidgetTypeDetails.java

@ -15,11 +15,13 @@
*/
package org.thingsboard.server.common.data.widget;
import com.fasterxml.jackson.annotation.JsonPropertyOrder;
import lombok.Data;
import org.thingsboard.server.common.data.id.WidgetTypeId;
import org.thingsboard.server.common.data.validation.NoXss;
@Data
@JsonPropertyOrder({ "alias", "name", "image", "description", "descriptor" })
public class WidgetTypeDetails extends WidgetType {
private String image;

2
common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java

@ -92,7 +92,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
channel = builder.build();
EdgeRpcServiceGrpc.EdgeRpcServiceStub stub = EdgeRpcServiceGrpc.newStub(channel);
log.info("[{}] Sending a connect request to the TB!", edgeKey);
this.inputStream = stub.handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDownlink, onError));
this.inputStream = stub.withCompression("gzip").handleMsgs(initOutputStream(edgeKey, onUplinkResponse, onEdgeUpdate, onDownlink, onError));
this.inputStream.onNext(RequestMsg.newBuilder()
.setMsgType(RequestMsgType.CONNECT_RPC_MESSAGE)
.setConnectRequestMsg(ConnectRequestMsg.newBuilder().setEdgeRoutingKey(edgeKey).setEdgeSecret(edgeSecret).build())

1
common/message/src/main/java/org/thingsboard/server/common/msg/rpc/ToDeviceRpcRequest.java

@ -36,6 +36,7 @@ public class ToDeviceRpcRequest implements Serializable {
private final long expirationTime;
private final ToDeviceRpcRequestBody body;
private final boolean persisted;
private final Integer retries;
@JsonIgnore
private final String additionalInfo;
}

62
common/queue/src/main/java/org/thingsboard/server/queue/DefaultQueueService.java

@ -0,0 +1,62 @@
/**
* Copyright © 2016-2021 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.queue;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings;
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration;
import javax.annotation.PostConstruct;
import java.util.Collections;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.stream.Collectors;
@Service
@RequiredArgsConstructor
public class DefaultQueueService implements QueueService {
private final TbQueueRuleEngineSettings ruleEngineSettings;
private Set<String> ruleEngineQueues;
@PostConstruct
public void init() {
ruleEngineQueues = ruleEngineSettings.getQueues().stream()
.map(TbRuleEngineQueueConfiguration::getName).collect(Collectors.toCollection(LinkedHashSet::new));
}
@Override
public Set<String> getQueuesByServiceType(ServiceType type) {
if (type == ServiceType.TB_RULE_ENGINE) {
return ruleEngineQueues;
} else {
return Collections.emptySet();
}
}
@Override
public String resolve(ServiceType serviceType, String queueName) {
if (StringUtils.isEmpty(queueName) || !getQueuesByServiceType(serviceType).contains(queueName)) {
return ServiceQueue.MAIN;
} else {
return queueName;
}
}
}

8
common/queue/src/main/java/org/thingsboard/server/queue/discovery/HashPartitionService.java

@ -22,6 +22,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.queue.ServiceQueue;
@ -30,6 +31,7 @@ import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.gen.transport.TransportProtos.ServiceInfo;
import org.thingsboard.server.queue.QueueService;
import org.thingsboard.server.queue.discovery.event.ClusterTopologyChangeEvent;
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.queue.discovery.event.ServiceListChangedEvent;
@ -64,6 +66,7 @@ public class HashPartitionService implements PartitionService {
private final TbServiceInfoProvider serviceInfoProvider;
private final TenantRoutingInfoService tenantRoutingInfoService;
private final TbQueueRuleEngineSettings tbQueueRuleEngineSettings;
private final QueueService queueService;
private final ConcurrentMap<ServiceQueue, String> partitionTopics = new ConcurrentHashMap<>();
private final ConcurrentMap<ServiceQueue, Integer> partitionSizes = new ConcurrentHashMap<>();
private final ConcurrentMap<TenantId, TenantRoutingInfo> tenantRoutingInfoMap = new ConcurrentHashMap<>();
@ -81,11 +84,13 @@ public class HashPartitionService implements PartitionService {
public HashPartitionService(TbServiceInfoProvider serviceInfoProvider,
TenantRoutingInfoService tenantRoutingInfoService,
ApplicationEventPublisher applicationEventPublisher,
TbQueueRuleEngineSettings tbQueueRuleEngineSettings) {
TbQueueRuleEngineSettings tbQueueRuleEngineSettings,
QueueService queueService) {
this.serviceInfoProvider = serviceInfoProvider;
this.tenantRoutingInfoService = tenantRoutingInfoService;
this.applicationEventPublisher = applicationEventPublisher;
this.tbQueueRuleEngineSettings = tbQueueRuleEngineSettings;
this.queueService = queueService;
}
@PostConstruct
@ -106,6 +111,7 @@ public class HashPartitionService implements PartitionService {
@Override
public TopicPartitionInfo resolve(ServiceType serviceType, String queueName, TenantId tenantId, EntityId entityId) {
queueName = queueService.resolve(serviceType, queueName);
return resolve(new ServiceQueue(serviceType, queueName), tenantId, entityId);
}

2
common/queue/src/main/java/org/thingsboard/server/queue/settings/TbQueueRuleEngineSettings.java

@ -36,8 +36,6 @@ public class TbQueueRuleEngineSettings {
private String topic;
private List<TbRuleEngineQueueConfiguration> queues;
//TODO 2.5 ybondarenko: make sure the queue names are valid to all queue providers.
// See how they are used in TbRuleEngineQueueFactory.createToRuleEngineMsgConsumer and all producers
@PostConstruct
public void validate() {
queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> {

93
common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/client/DefaultCoapClientContext.java

@ -42,6 +42,7 @@ import org.thingsboard.server.common.data.device.profile.ProtoTransportPayloadCo
import org.thingsboard.server.common.data.device.profile.TransportPayloadTypeConfiguration;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.DeviceProfileId;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.msg.session.FeatureType;
import org.thingsboard.server.common.msg.session.SessionMsgType;
import org.thingsboard.server.common.transport.SessionMsgListener;
@ -192,29 +193,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
client.lock();
try {
long uplinkTime = client.updateLastUplinkTime(uplinkTs);
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;
}
long timeout = getTimeout(client, powerMode, profileSettings);
Future<Void> sleepTask = client.getSleepTask();
if (sleepTask != null) {
sleepTask.cancel(false);
@ -234,6 +213,33 @@ public class DefaultCoapClientContext implements CoapClientContext {
}
}
private long getTimeout(TbCoapClientState client, PowerMode powerMode, PowerSavingConfiguration profileSettings) {
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;
}
return timeout;
}
private boolean registerFeatureObservation(TbCoapClientState state, String token, CoapExchange exchange, FeatureType featureType) {
state.lock();
try {
@ -524,17 +530,38 @@ public class DefaultCoapClientContext implements CoapClientContext {
Response response = state.getAdaptor().convertToPublish(conRequest, msg, state.getConfiguration().getRpcRequestDynamicMessageBuilder());
int requestId = getNextMsgId();
response.setMID(requestId);
if (msg.getPersisted() && conRequest) {
if (conRequest) {
PowerMode powerMode = state.getPowerMode();
PowerSavingConfiguration profileSettings = null;
if (powerMode == null) {
var clientProfile = getProfile(state.getProfileId());
if (clientProfile.isPresent()) {
profileSettings = clientProfile.get().getClientSettings();
if (profileSettings != null) {
powerMode = profileSettings.getPowerMode();
}
}
}
transportContext.getRpcAwaitingAck().put(requestId, msg);
transportContext.getScheduler().schedule(() -> {
transportContext.getRpcAwaitingAck().remove(requestId);
}, Math.max(0, msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(requestId);
if (rpcRequestMsg != null) {
transportService.process(state.getSession(), msg, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
}
}, Math.min(getTimeout(state, powerMode, profileSettings), msg.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> {
TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
if (rpcRequestMsg != null) {
transportService.process(state.getSession(), rpcRequestMsg, TransportServiceCallback.EMPTY);
transportService.process(state.getSession(), rpcRequestMsg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
}
}, null));
}, id -> {
TransportProtos.ToDeviceRpcRequestMsg rpcRequestMsg = transportContext.getRpcAwaitingAck().remove(id);
if (rpcRequestMsg != null) {
transportService.process(state.getSession(), msg, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
}
}));
}
if (conRequest) {
response.addMessageObserver(new TbCoapMessageObserver(requestId, id -> awake(state), id -> asleep(state)));
@ -553,8 +580,12 @@ public class DefaultCoapClientContext implements CoapClientContext {
transportService.process(state.getSession(),
TransportProtos.ToDeviceRpcResponseMsg.newBuilder()
.setRequestId(msg.getRequestId()).setError(error).build(), TransportServiceCallback.EMPTY);
} else if (msg.getPersisted() && !conRequest && sent) {
transportService.process(state.getSession(), msg, TransportServiceCallback.EMPTY);
} else if (sent) {
if (!conRequest) {
transportService.process(state.getSession(), msg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
} else if (msg.getPersisted()) {
transportService.process(state.getSession(), msg, RpcStatus.SENT, TransportServiceCallback.EMPTY);
}
}
}
}
@ -723,7 +754,7 @@ public class DefaultCoapClientContext implements CoapClientContext {
private void cancelRpcSubscription(TbCoapClientState state) {
if (state.getRpc() != null) {
clientsByToken.remove(state.getRpc().getToken());
CoapExchange exchange = state.getAttrs().getExchange();
CoapExchange exchange = state.getRpc().getExchange();
state.setRpc(null);
transportService.process(state.getSession(),
TransportProtos.SubscribeToRPCMsg.newBuilder().setUnsubscribe(true).build(),

12
common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java

@ -115,7 +115,6 @@ public class DeviceApiController implements TbTransportService {
TransportService transportService = transportContext.getTransportService();
transportService.process(sessionInfo, JsonConverter.convertToAttributesProto(new JsonParser().parse(json)),
new HttpOkCallback(responseWriter));
reportActivity(sessionInfo);
}));
return responseWriter;
}
@ -129,7 +128,6 @@ public class DeviceApiController implements TbTransportService {
TransportService transportService = transportContext.getTransportService();
transportService.process(sessionInfo, JsonConverter.convertToTelemetryProto(new JsonParser().parse(json)),
new HttpOkCallback(responseWriter));
reportActivity(sessionInfo);
}));
return responseWriter;
}
@ -409,7 +407,7 @@ public class DeviceApiController implements TbTransportService {
public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg msg) {
log.trace("[{}] Received RPC command to device", sessionId);
responseWriter.setResult(new ResponseEntity<>(JsonConverter.toJson(msg, true).toString(), HttpStatus.OK));
transportService.process(sessionInfo, msg, TransportServiceCallback.EMPTY);
transportService.process(sessionInfo, msg, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
}
@Override
@ -419,14 +417,6 @@ public class DeviceApiController implements TbTransportService {
}
private void reportActivity(SessionInfoProto sessionInfo) {
transportContext.getTransportService().process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder()
.setAttributeSubscription(false)
.setRpcSubscription(false)
.setLastActivityTime(System.currentTimeMillis())
.build(), TransportServiceCallback.EMPTY);
}
private static MediaType parseMediaType(String contentType) {
try {
return MediaType.parseMediaType(contentType);

3
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/DefaultLwM2MRpcRequestHandler.java

@ -21,7 +21,9 @@ import org.eclipse.leshan.core.ResponseCode;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil;
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.TransportServiceCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.util.TbLwM2mTransportComponent;
import org.thingsboard.server.transport.lwm2m.config.LwM2MTransportServerConfig;
@ -158,6 +160,7 @@ public class DefaultLwM2MRpcRequestHandler implements LwM2MRpcRequestHandler {
throw new IllegalArgumentException("Unsupported operation: " + operationType.name());
}
}
transportService.process(client.getSession(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
} catch (IllegalArgumentException e) {
this.sendErrorRpcResponse(sessionInfo, rpcRequest.getRequestId(), ResponseCode.BAD_REQUEST, e.getMessage());
}

7
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/rpc/RpcDownlinkRequestCallbackProxy.java

@ -19,6 +19,7 @@ import org.eclipse.leshan.core.ResponseCode;
import org.eclipse.leshan.core.request.exception.ClientSleepingException;
import org.thingsboard.common.util.JacksonUtil;
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.TransportServiceCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
@ -44,7 +45,7 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
@Override
public void onSuccess(R request, T response) {
transportService.process(client.getSession(), this.request, TransportServiceCallback.EMPTY);
transportService.process(client.getSession(), this.request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
sendRpcReplyOnSuccess(response);
if (callback != null) {
callback.onSuccess(request, response);
@ -61,7 +62,9 @@ public abstract class RpcDownlinkRequestCallbackProxy<R, T> implements DownlinkR
@Override
public void onError(String params, Exception e) {
if (!(e instanceof TimeoutException || e instanceof ClientSleepingException)) {
if (e instanceof TimeoutException) {
transportService.process(client.getSession(), this.request, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
} else if (!(e instanceof ClientSleepingException)) {
sendRpcReplyOnError(e);
}
if (callback != null) {

2
common/transport/lwm2m/src/main/java/org/thingsboard/server/transport/lwm2m/server/uplink/DefaultLwM2MUplinkMsgHandler.java

@ -908,7 +908,7 @@ public class DefaultLwM2MUplinkMsgHandler extends LwM2MExecutorAwareService impl
* @param sessionInfo -
*/
private void reportActivityAndRegister(SessionInfoProto sessionInfo) {
if (sessionInfo != null && transportService.reportActivity(sessionInfo) == null) {
if (sessionInfo != null && !transportService.hasSession(sessionInfo)) {
sessionManager.register(sessionInfo);
this.reportActivitySubscription(sessionInfo);
}

3
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportContext.java

@ -68,4 +68,7 @@ public class MqttTransportContext extends TransportContext {
@Value("${transport.mqtt.msg_queue_size_per_device_limit:100}")
private int messageQueueSizePerDeviceLimit;
@Getter
@Value("${transport.mqtt.timeout:10000}")
private long timeout;
}

28
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java

@ -50,6 +50,7 @@ import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
import org.thingsboard.server.common.data.id.OtaPackageId;
import org.thingsboard.server.common.data.ota.OtaPackageType;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.msg.EncryptionUtil;
import org.thingsboard.server.common.msg.tools.TbRateLimitsException;
import org.thingsboard.server.common.transport.SessionMsgListener;
@ -272,7 +273,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
int msgId = ((MqttPubAckMessage) msg).variableHeader().messageId();
TransportProtos.ToDeviceRpcRequestMsg rpcRequest = rpcAwaitingAck.remove(msgId);
if (rpcRequest != null) {
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY);
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
}
break;
default:
@ -849,20 +850,27 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
try {
deviceSessionCtx.getPayloadAdaptor().convertToPublish(deviceSessionCtx, rpcRequest).ifPresent(payload -> {
int msgId = ((MqttPublishMessage) payload).variableHeader().packetId();
if (rpcRequest.getPersisted() && isAckExpected(payload)) {
if (isAckExpected(payload)) {
rpcAwaitingAck.put(msgId, rpcRequest);
context.getScheduler().schedule(() -> {
rpcAwaitingAck.remove(msgId);
}, Math.max(0, rpcRequest.getExpirationTime() - System.currentTimeMillis()), TimeUnit.MILLISECONDS);
TransportProtos.ToDeviceRpcRequestMsg msg = rpcAwaitingAck.remove(msgId);
if (msg != null) {
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.TIMEOUT, TransportServiceCallback.EMPTY);
}
}, Math.max(0, Math.min(deviceSessionCtx.getContext().getTimeout(), rpcRequest.getExpirationTime() - System.currentTimeMillis())), TimeUnit.MILLISECONDS);
}
var cf = publish(payload, deviceSessionCtx);
if (rpcRequest.getPersisted() && !isAckExpected(payload)) {
cf.addListener(result -> {
if (result.cause() == null) {
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, TransportServiceCallback.EMPTY);
cf.addListener(result -> {
if (result.cause() == null) {
if (!isAckExpected(payload)) {
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
} else if (rpcRequest.getPersisted()) {
transportService.process(deviceSessionCtx.getSessionInfo(), rpcRequest, RpcStatus.SENT, TransportServiceCallback.EMPTY);
}
});
}
} else {
// TODO: send error
}
});
});
} catch (Exception e) {
transportService.process(deviceSessionCtx.getSessionInfo(),

17
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java

@ -16,8 +16,10 @@
package org.thingsboard.server.transport.mqtt.session;
import io.netty.channel.ChannelFuture;
import io.netty.handler.codec.mqtt.MqttMessage;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.transport.SessionMsgListener;
import org.thingsboard.server.common.transport.TransportService;
import org.thingsboard.server.common.transport.TransportServiceCallback;
@ -102,9 +104,14 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
payload -> {
ChannelFuture channelFuture = parent.writeAndFlush(payload);
if (request.getPersisted()) {
channelFuture.addListener(future -> {
if (future.cause() == null) {
transportService.process(getSessionInfo(), request, TransportServiceCallback.EMPTY);
channelFuture.addListener(result -> {
if (result.cause() == null) {
if (!isAckExpected(payload)) {
transportService.process(getSessionInfo(), request, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
} else if (request.getPersisted()) {
transportService.process(getSessionInfo(), request, RpcStatus.SENT, TransportServiceCallback.EMPTY);
}
}
});
}
@ -129,4 +136,8 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple
// This feature is not supported in the TB IoT Gateway yet.
}
private boolean isAckExpected(MqttMessage message) {
return message.fixedHeader().qosLevel().value() > 0;
}
}

3
common/transport/snmp/src/main/java/org/thingsboard/server/transport/snmp/session/DeviceSessionContext.java

@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.device.data.SnmpDeviceTransportConfiguration;
import org.thingsboard.server.common.data.device.profile.SnmpDeviceProfileTransportConfiguration;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.transport.SessionMsgListener;
import org.thingsboard.server.common.transport.TransportServiceCallback;
import org.thingsboard.server.common.transport.session.DeviceAwareSessionContext;
@ -142,7 +143,7 @@ public class DeviceSessionContext extends DeviceAwareSessionContext implements S
public void onToDeviceRpcRequest(UUID sessionId, ToDeviceRpcRequestMsg toDeviceRequest) {
log.trace("[{}] Received RPC command to device", sessionId);
snmpTransportContext.getSnmpTransportService().onToDeviceRpcRequest(this, toDeviceRequest);
snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, TransportServiceCallback.EMPTY);
snmpTransportContext.getTransportService().process(getSessionInfo(), toDeviceRequest, RpcStatus.DELIVERED, TransportServiceCallback.EMPTY);
}
@Override

7
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/TransportService.java

@ -17,6 +17,7 @@ package org.thingsboard.server.common.transport;
import org.thingsboard.server.common.data.DeviceProfile;
import org.thingsboard.server.common.data.DeviceTransportType;
import org.thingsboard.server.common.data.rpc.RpcStatus;
import org.thingsboard.server.common.transport.auth.GetOrCreateDeviceFromGatewayResponse;
import org.thingsboard.server.common.transport.auth.ValidateDeviceCredentialsResponse;
import org.thingsboard.server.common.transport.service.SessionMetaData;
@ -112,7 +113,7 @@ public interface TransportService {
void process(SessionInfoProto sessionInfo, ToServerRpcRequestMsg msg, TransportServiceCallback<Void> callback);
void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, TransportServiceCallback<Void> callback);
void process(SessionInfoProto sessionInfo, ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback<Void> callback);
void process(SessionInfoProto sessionInfo, SubscriptionInfoProto msg, TransportServiceCallback<Void> callback);
@ -126,7 +127,7 @@ public interface TransportService {
SessionMetaData registerSyncSession(SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout);
SessionMetaData reportActivity(SessionInfoProto sessionInfo);
void reportActivity(SessionInfoProto sessionInfo);
void deregisterSession(SessionInfoProto sessionInfo);
@ -135,4 +136,6 @@ public interface TransportService {
void notifyAboutUplink(SessionInfoProto sessionInfo, TransportProtos.UplinkNotificationMsg build, TransportServiceCallback<Void> empty);
ExecutorService getCallbackExecutor();
boolean hasSession(SessionInfoProto sessionInfo);
}

117
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java

@ -95,10 +95,12 @@ import org.thingsboard.server.queue.util.TbTransportComponent;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -162,6 +164,7 @@ public class DefaultTransportService implements TransportService {
private ExecutorService mainConsumerExecutor;
private final ConcurrentMap<UUID, SessionMetaData> sessions = new ConcurrentHashMap<>();
private final ConcurrentMap<UUID, SessionActivityData> sessionsActivity = new ConcurrentHashMap<>();
private final Map<String, RpcRequestMetadata> toServerRpcPendingMap = new ConcurrentHashMap<>();
private volatile boolean stopped = false;
@ -545,8 +548,11 @@ public class DefaultTransportService implements TransportService {
@Override
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SubscribeToAttributeUpdatesMsg msg, TransportServiceCallback<Void> callback) {
if (checkLimits(sessionInfo, msg, callback)) {
SessionMetaData sessionMetaData = reportActivityInternal(sessionInfo);
sessionMetaData.setSubscribedToAttributes(!msg.getUnsubscribe());
SessionMetaData sessionMetaData = sessions.get(toSessionId(sessionInfo));
if (sessionMetaData != null) {
sessionMetaData.setSubscribedToAttributes(!msg.getUnsubscribe());
}
reportActivityInternal(sessionInfo);
sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setSubscribeToAttributes(msg).build(),
new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, callback));
}
@ -555,8 +561,11 @@ public class DefaultTransportService implements TransportService {
@Override
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SubscribeToRPCMsg msg, TransportServiceCallback<Void> callback) {
if (checkLimits(sessionInfo, msg, callback)) {
SessionMetaData sessionMetaData = reportActivityInternal(sessionInfo);
sessionMetaData.setSubscribedToRPC(!msg.getUnsubscribe());
SessionMetaData sessionMetaData = sessions.get(toSessionId(sessionInfo));
if (sessionMetaData != null) {
sessionMetaData.setSubscribedToRPC(!msg.getUnsubscribe());
}
reportActivityInternal(sessionInfo);
sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setSubscribeToRPC(msg).build(),
new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, callback));
}
@ -580,22 +589,18 @@ public class DefaultTransportService implements TransportService {
}
@Override
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, TransportServiceCallback<Void> callback) {
if (msg.getPersisted()) {
RpcStatus status = msg.getOneway() ? RpcStatus.SUCCESSFUL : RpcStatus.DELIVERED;
TransportProtos.ToDevicePersistedRpcResponseMsg responseMsg = TransportProtos.ToDevicePersistedRpcResponseMsg.newBuilder()
.setRequestId(msg.getRequestId())
.setRequestIdLSB(msg.getRequestIdLSB())
.setRequestIdMSB(msg.getRequestIdMSB())
.setStatus(status.name())
.build();
if (checkLimits(sessionInfo, responseMsg, callback)) {
reportActivityInternal(sessionInfo);
sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setPersistedRpcResponseMsg(responseMsg).build(),
new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
}
public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToDeviceRpcRequestMsg msg, RpcStatus rpcStatus, TransportServiceCallback<Void> callback) {
TransportProtos.ToDeviceRpcResponseStatusMsg responseMsg = TransportProtos.ToDeviceRpcResponseStatusMsg.newBuilder()
.setRequestId(msg.getRequestId())
.setRequestIdLSB(msg.getRequestIdLSB())
.setRequestIdMSB(msg.getRequestIdMSB())
.setStatus(rpcStatus.name())
.build();
if (checkLimits(sessionInfo, responseMsg, callback)) {
reportActivityInternal(sessionInfo);
sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo).setRpcResponseStatusMsg(responseMsg).build(),
new ApiStatsProxyCallback<>(getTenantId(sessionInfo), getCustomerId(sessionInfo), 1, TransportServiceCallback.EMPTY));
}
}
@ -668,52 +673,61 @@ public class DefaultTransportService implements TransportService {
}
@Override
public SessionMetaData reportActivity(TransportProtos.SessionInfoProto sessionInfo) {
return reportActivityInternal(sessionInfo);
public void reportActivity(TransportProtos.SessionInfoProto sessionInfo) {
reportActivityInternal(sessionInfo);
}
private SessionMetaData reportActivityInternal(TransportProtos.SessionInfoProto sessionInfo) {
private void reportActivityInternal(TransportProtos.SessionInfoProto sessionInfo) {
UUID sessionId = toSessionId(sessionInfo);
SessionMetaData sessionMetaData = sessions.get(sessionId);
if (sessionMetaData != null) {
sessionMetaData.updateLastActivityTime();
}
return sessionMetaData;
SessionActivityData sessionMetaData = sessionsActivity.computeIfAbsent(sessionId, id -> new SessionActivityData(sessionInfo));
sessionMetaData.updateLastActivityTime();
}
private void checkInactivityAndReportActivity() {
long expTime = System.currentTimeMillis() - sessionInactivityTimeout;
sessions.forEach((uuid, sessionMD) -> {
long lastActivityTime = sessionMD.getLastActivityTime();
TransportProtos.SessionInfoProto sessionInfo = sessionMD.getSessionInfo();
if (sessionInfo.getGwSessionIdMSB() != 0 &&
sessionInfo.getGwSessionIdLSB() != 0) {
SessionMetaData gwMetaData = sessions.get(new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB()));
Set<UUID> sessionsToRemove = new HashSet<>();
sessionsActivity.forEach((uuid, sessionAD) -> {
long lastActivityTime = sessionAD.getLastActivityTime();
SessionMetaData sessionMD = sessions.get(uuid);
if (sessionMD != null) {
sessionAD.setSessionInfo(sessionMD.getSessionInfo());
} else {
sessionsToRemove.add(uuid);
}
TransportProtos.SessionInfoProto sessionInfo = sessionAD.getSessionInfo();
if (sessionInfo.getGwSessionIdMSB() != 0 && sessionInfo.getGwSessionIdLSB() != 0) {
var gwSessionId = new UUID(sessionInfo.getGwSessionIdMSB(), sessionInfo.getGwSessionIdLSB());
SessionMetaData gwMetaData = sessions.get(gwSessionId);
SessionActivityData gwActivityData = sessionsActivity.get(gwSessionId);
if (gwMetaData != null && gwMetaData.isOverwriteActivityTime()) {
lastActivityTime = Math.max(gwMetaData.getLastActivityTime(), lastActivityTime);
lastActivityTime = Math.max(gwActivityData.getLastActivityTime(), lastActivityTime);
}
}
if (lastActivityTime < expTime) {
if (log.isDebugEnabled()) {
log.debug("[{}] Session has expired due to last activity time: {}", toSessionId(sessionInfo), lastActivityTime);
if (sessionMD != null) {
if (log.isDebugEnabled()) {
log.debug("[{}] Session has expired due to last activity time: {}", toSessionId(sessionInfo), lastActivityTime);
}
sessions.remove(uuid);
sessionsToRemove.add(uuid);
process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null);
TransportProtos.SessionCloseNotificationProto sessionCloseNotificationProto = TransportProtos.SessionCloseNotificationProto
.newBuilder()
.setMessage("Session has expired due to last activity time!")
.build();
sessionMD.getListener().onRemoteSessionCloseCommand(uuid, sessionCloseNotificationProto);
}
process(sessionInfo, getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null);
sessions.remove(uuid);
TransportProtos.SessionCloseNotificationProto sessionCloseNotificationProto = TransportProtos.SessionCloseNotificationProto
.newBuilder()
.setMessage("session has expired due to last activity time!")
.build();
sessionMD.getListener().onRemoteSessionCloseCommand(uuid, sessionCloseNotificationProto);
} else {
if (lastActivityTime > sessionMD.getLastReportedActivityTime()) {
if (lastActivityTime > sessionAD.getLastReportedActivityTime()) {
final long lastActivityTimeFinal = lastActivityTime;
process(sessionInfo, TransportProtos.SubscriptionInfoProto.newBuilder()
.setAttributeSubscription(sessionMD.isSubscribedToAttributes())
.setRpcSubscription(sessionMD.isSubscribedToRPC())
.setAttributeSubscription(sessionMD != null && sessionMD.isSubscribedToAttributes())
.setRpcSubscription(sessionMD != null && sessionMD.isSubscribedToRPC())
.setLastActivityTime(lastActivityTime).build(), new TransportServiceCallback<Void>() {
@Override
public void onSuccess(Void msg) {
sessionMD.setLastReportedActivityTime(lastActivityTimeFinal);
sessionAD.setLastReportedActivityTime(lastActivityTimeFinal);
}
@Override
@ -724,6 +738,8 @@ public class DefaultTransportService implements TransportService {
}
}
});
// Removes all closed or short-lived sessions.
sessionsToRemove.forEach(sessionsActivity::remove);
}
@Override
@ -1146,4 +1162,9 @@ public class DefaultTransportService implements TransportService {
public ExecutorService getCallbackExecutor() {
return transportCallbackExecutor;
}
@Override
public boolean hasSession(TransportProtos.SessionInfoProto sessionInfo) {
return sessions.containsKey(toSessionId(sessionInfo));
}
}

42
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionActivityData.java

@ -0,0 +1,42 @@
/**
* Copyright © 2016-2021 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.common.transport.service;
import lombok.Data;
import org.thingsboard.server.common.transport.SessionMsgListener;
import org.thingsboard.server.gen.transport.TransportProtos;
import java.util.concurrent.ScheduledFuture;
/**
* Created by ashvayka on 15.10.18.
*/
@Data
public class SessionActivityData {
private volatile TransportProtos.SessionInfoProto sessionInfo;
private volatile long lastActivityTime;
private volatile long lastReportedActivityTime;
SessionActivityData(TransportProtos.SessionInfoProto sessionInfo) {
this.sessionInfo = sessionInfo;
}
void updateLastActivityTime() {
this.lastActivityTime = System.currentTimeMillis();
}
}

7
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/SessionMetaData.java

@ -32,8 +32,6 @@ public class SessionMetaData {
private final SessionMsgListener listener;
private volatile ScheduledFuture scheduledFuture;
private volatile long lastActivityTime;
private volatile long lastReportedActivityTime;
private volatile boolean subscribedToAttributes;
private volatile boolean subscribedToRPC;
private volatile boolean overwriteActivityTime;
@ -42,14 +40,9 @@ public class SessionMetaData {
this.sessionInfo = sessionInfo;
this.sessionType = sessionType;
this.listener = listener;
this.lastActivityTime = System.currentTimeMillis();
this.scheduledFuture = null;
}
void updateLastActivityTime() {
this.lastActivityTime = System.currentTimeMillis();
}
void setScheduledFuture(ScheduledFuture scheduledFuture) {
this.scheduledFuture = scheduledFuture;
}

13
dao/src/main/java/org/thingsboard/server/dao/device/DeviceProfileServiceImpl.java

@ -28,7 +28,7 @@ import com.squareup.wire.schema.internal.parser.ProtoFileElement;
import com.squareup.wire.schema.internal.parser.ProtoParser;
import com.squareup.wire.schema.internal.parser.TypeElement;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.thingsboard.server.common.data.StringUtils;
import org.hibernate.exception.ConstraintViolationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cache.Cache;
@ -63,6 +63,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.page.PageLink;
import org.thingsboard.server.common.data.rule.RuleChain;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.dao.dashboard.DashboardService;
import org.thingsboard.server.dao.entity.AbstractEntityService;
import org.thingsboard.server.dao.exception.DataValidationException;
@ -72,6 +73,7 @@ import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.dao.service.PaginatedRemover;
import org.thingsboard.server.dao.service.Validator;
import org.thingsboard.server.dao.tenant.TenantDao;
import org.thingsboard.server.queue.QueueService;
import java.util.Arrays;
import java.util.Collections;
@ -82,7 +84,6 @@ import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import static com.google.protobuf.FieldType.MESSAGE;
import static org.thingsboard.server.common.data.CacheConstants.DEVICE_PROFILE_CACHE;
import static org.thingsboard.server.dao.service.Validator.validateId;
@ -104,6 +105,9 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D
return "[Transport Configuration] invalid " + schemaName + " provided!";
}
@Autowired(required = false)
private QueueService queueService;
@Autowired
private DeviceProfileDao deviceProfileDao;
@ -373,6 +377,11 @@ public class DeviceProfileServiceImpl extends AbstractEntityService implements D
throw new DataValidationException("Another default device profile is present in scope of current tenant!");
}
}
if (!StringUtils.isEmpty(deviceProfile.getDefaultQueueName()) && queueService != null){
if(!queueService.getQueuesByServiceType(ServiceType.TB_RULE_ENGINE).contains(deviceProfile.getDefaultQueueName())){
throw new DataValidationException("Device profile is referencing to non-existent queue!");
}
}
if (deviceProfile.getProvisionType() == null) {
deviceProfile.setProvisionType(DeviceProfileProvisionType.DISABLED);
}

10
dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java

@ -15,6 +15,7 @@
*/
package org.thingsboard.server.dao.edge;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
@ -34,6 +35,7 @@ import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.CacheEvict;
import org.springframework.cache.annotation.Cacheable;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
@ -543,16 +545,16 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
}
@Override
public Object checkInstance(Object request) {
return this.restTemplate.postForEntity(EDGE_LICENSE_SERVER_ENDPOINT + "/api/license/checkInstance", request, Object.class, new Object[0]);
public ResponseEntity<JsonNode> checkInstance(JsonNode request) {
return this.restTemplate.postForEntity(EDGE_LICENSE_SERVER_ENDPOINT + "/api/license/checkInstance", request, JsonNode.class);
}
@Override
public Object activateInstance(String edgeLicenseSecret, String releaseDate) {
public ResponseEntity<JsonNode> activateInstance(String edgeLicenseSecret, String releaseDate) {
Map<String, String> params = new HashMap<>();
params.put("licenseSecret", edgeLicenseSecret);
params.put("releaseDate", releaseDate);
return this.restTemplate.postForEntity(EDGE_LICENSE_SERVER_ENDPOINT + "/api/license/activateInstance?licenseSecret={licenseSecret}&releaseDate={releaseDate}", (Object) null, Object.class, params);
return this.restTemplate.postForEntity(EDGE_LICENSE_SERVER_ENDPOINT + "/api/license/activateInstance?licenseSecret={licenseSecret}&releaseDate={releaseDate}", null, JsonNode.class, params);
}
@Override

2
msa/js-executor/package.json

@ -1,7 +1,7 @@
{
"name": "thingsboard-js-executor",
"private": true,
"version": "3.3.0",
"version": "3.3.1",
"description": "ThingsBoard JavaScript Executor Microservice",
"main": "server.js",
"bin": "server.js",

2
msa/tb/pom.xml

@ -39,7 +39,7 @@
<tb-postgres.docker.name>tb-postgres</tb-postgres.docker.name>
<tb-cassandra.docker.name>tb-cassandra</tb-cassandra.docker.name>
<pkg.installFolder>/usr/share/${pkg.name}</pkg.installFolder>
<pkg.upgradeVersion>3.3.0</pkg.upgradeVersion>
<pkg.upgradeVersion>3.3.1</pkg.upgradeVersion>
</properties>
<dependencies>

2
msa/web-ui/package.json

@ -1,7 +1,7 @@
{
"name": "thingsboard-web-ui",
"private": true,
"version": "3.3.0",
"version": "3.3.1",
"description": "ThingsBoard Web UI Microservice",
"main": "server.js",
"bin": "server.js",

2
rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineDeviceRpcRequest.java

@ -41,5 +41,5 @@ public final class RuleEngineDeviceRpcRequest {
private final long expirationTime;
private final boolean restApiCall;
private final String additionalInfo;
private final Integer retries;
}

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/filter/TbMsgTypeSwitchNode.java

@ -33,7 +33,7 @@ import org.thingsboard.server.common.msg.session.SessionMsgType;
type = ComponentType.FILTER,
name = "message type switch",
configClazz = EmptyNodeConfiguration.class,
relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Failed", "RPC Deleted",
relationTypes = {"Post attributes", "Post telemetry", "RPC Request from Device", "RPC Request to Device", "RPC Queued", "RPC Sent", "RPC Delivered", "RPC Successful", "RPC Timeout", "RPC Expired", "RPC Failed", "RPC Deleted",
"Activity Event", "Inactivity Event", "Connect Event", "Disconnect Event", "Entity Created", "Entity Updated", "Entity Deleted", "Entity Assigned",
"Entity Unassigned", "Attributes Updated", "Attributes Deleted", "Alarm Acknowledged", "Alarm Cleared", "Other", "Entity Assigned From Tenant", "Entity Assigned To Tenant",
"Timeseries Updated", "Timeseries Deleted"},
@ -97,12 +97,16 @@ public class TbMsgTypeSwitchNode implements TbNode {
relationType = "Timeseries Deleted";
} else if (msg.getType().equals(DataConstants.RPC_QUEUED)) {
relationType = "RPC Queued";
} else if (msg.getType().equals(DataConstants.RPC_SENT)) {
relationType = "RPC Sent";
} else if (msg.getType().equals(DataConstants.RPC_DELIVERED)) {
relationType = "RPC Delivered";
} else if (msg.getType().equals(DataConstants.RPC_SUCCESSFUL)) {
relationType = "RPC Successful";
} else if (msg.getType().equals(DataConstants.RPC_TIMEOUT)) {
relationType = "RPC Timeout";
} else if (msg.getType().equals(DataConstants.RPC_EXPIRED)) {
relationType = "RPC Expired";
} else if (msg.getType().equals(DataConstants.RPC_FAILED)) {
relationType = "RPC Failed";
} else if (msg.getType().equals(DataConstants.RPC_DELETED)) {

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCRequestNode.java

@ -89,9 +89,12 @@ public class TbSendRPCRequestNode implements TbNode {
tmp = msg.getMetaData().getValue("originServiceId");
String originServiceId = !StringUtils.isEmpty(tmp) ? tmp : null;
tmp = msg.getMetaData().getValue("expirationTime");
tmp = msg.getMetaData().getValue(DataConstants.EXPIRATION_TIME);
long expirationTime = !StringUtils.isEmpty(tmp) ? Long.parseLong(tmp) : (System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(config.getTimeoutInSeconds()));
tmp = msg.getMetaData().getValue(DataConstants.RETRIES);
Integer retries = !StringUtils.isEmpty(tmp) ? Integer.parseInt(tmp) : null;
String params;
JsonElement paramsEl = json.get("params");
if (paramsEl.isJsonPrimitive()) {
@ -112,6 +115,7 @@ public class TbSendRPCRequestNode implements TbNode {
.requestUUID(requestUUID)
.originServiceId(originServiceId)
.expirationTime(expirationTime)
.retries(retries)
.restApiCall(restApiCall)
.persisted(persisted)
.additionalInfo(additionalInfo)

2
transport/coap/src/main/resources/tb-coap-transport.yml

@ -113,7 +113,7 @@ transport:
dtls_session_report_timeout: "${TB_COAP_X509_DTLS_SESSION_REPORT_TIMEOUT:1800000}"
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"

2
transport/http/src/main/resources/tb-http-transport.yml

@ -85,7 +85,7 @@ transport:
max_request_timeout: "${HTTP_MAX_REQUEST_TIMEOUT:300000}"
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"

2
transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml

@ -87,7 +87,7 @@ redis:
transport:
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:false}"

2
transport/mqtt/src/main/resources/tb-mqtt-transport.yml

@ -118,7 +118,7 @@ transport:
skip_validity_check_for_client_cert: "${MQTT_SSL_SKIP_VALIDITY_CHECK_FOR_CLIENT_CERT:false}"
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"

2
transport/snmp/src/main/resources/tb-snmp-transport.yml

@ -50,7 +50,7 @@ transport:
underlying_protocol: "${SNMP_UNDERLYING_PROTOCOL:udp}"
sessions:
inactivity_timeout: "${TB_TRANSPORT_SESSIONS_INACTIVITY_TIMEOUT:300000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:30000}"
report_timeout: "${TB_TRANSPORT_SESSIONS_REPORT_TIMEOUT:3000}"
json:
# Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"

9
ui-ngx/angular.json

@ -78,7 +78,8 @@
"node_modules/leaflet/dist/leaflet.css",
"src/app/modules/home/components/widget/lib/maps/markers.scss",
"node_modules/leaflet.markercluster/dist/MarkerCluster.css",
"node_modules/leaflet.markercluster/dist/MarkerCluster.Default.css"
"node_modules/leaflet.markercluster/dist/MarkerCluster.Default.css",
"node_modules/prismjs/themes/prism.css"
],
"stylePreprocessorOptions": {
"includePaths": [
@ -88,7 +89,11 @@
"scripts": [
"node_modules/tinycolor2/dist/tinycolor-min.js",
"node_modules/split.js/dist/split.min.js",
"node_modules/systemjs/dist/system.js"
"node_modules/systemjs/dist/system.js",
"node_modules/marked/lib/marked.js",
"node_modules/prismjs/prism.js",
"node_modules/prismjs/components/prism-bash.min.js",
"node_modules/prismjs/components/prism-json.min.js"
],
"customWebpackConfig": {
"path": "./extra-webpack.config.js"

4
ui-ngx/package.json

@ -1,6 +1,6 @@
{
"name": "thingsboard",
"version": "3.3.0",
"version": "3.3.1",
"scripts": {
"ng": "ng",
"start": "node --max_old_space_size=8048 ./node_modules/@angular/cli/bin/ng serve --host 0.0.0.0 --open",
@ -72,6 +72,7 @@
"ngx-drag-drop": "^2.0.0",
"ngx-flowchart": "git://github.com/thingsboard/ngx-flowchart.git#master",
"ngx-hm-carousel": "^2.0.0-rc.1",
"ngx-markdown": "^10.1.1",
"ngx-sharebuttons": "^8.0.5",
"ngx-translate-messageformat-compiler": "^4.9.0",
"objectpath": "^2.0.0",
@ -112,6 +113,7 @@
"@types/js-beautify": "^1.13.1",
"@types/jstree": "^3.3.40",
"@types/leaflet": "1.5.17",
"@types/leaflet-editable": "^1.2.1",
"@types/leaflet-polylinedecorator": "^1.6.0",
"@types/leaflet.markercluster": "^1.4.4",
"@types/lodash": "^4.14.170",

1
ui-ngx/src/app/core/api/widget-api.models.ts

@ -243,6 +243,7 @@ export interface WidgetSubscriptionOptions {
legendConfig?: LegendConfig;
comparisonEnabled?: boolean;
timeForComparison?: moment_.unitOfTime.DurationConstructor;
comparisonCustomIntervalValue?: number;
decimals?: number;
units?: string;
callbacks?: WidgetSubscriptionCallbacks;

5
ui-ngx/src/app/core/api/widget-subscription.ts

@ -110,6 +110,7 @@ export class WidgetSubscription implements IWidgetSubscription {
units: string;
comparisonEnabled: boolean;
timeForComparison: ComparisonDuration;
comparisonCustomIntervalValue: number;
comparisonTimeWindow: WidgetTimewindow;
timewindowForComparison: SubscriptionTimewindow;
@ -234,6 +235,7 @@ export class WidgetSubscription implements IWidgetSubscription {
this.comparisonEnabled = options.comparisonEnabled && isHistoryTypeTimewindow(this.timeWindowConfig);
if (this.comparisonEnabled) {
this.timeForComparison = options.timeForComparison;
this.comparisonCustomIntervalValue = options.comparisonCustomIntervalValue;
this.comparisonTimeWindow = {};
this.timewindowForComparison = null;
@ -1183,7 +1185,8 @@ export class WidgetSubscription implements IWidgetSubscription {
}
private updateSubscriptionForComparison(): SubscriptionTimewindow {
this.timewindowForComparison = createTimewindowForComparison(this.subscriptionTimewindow, this.timeForComparison);
this.timewindowForComparison = createTimewindowForComparison(this.subscriptionTimewindow, this.timeForComparison,
this.comparisonCustomIntervalValue);
this.updateComparisonTimewindow();
return this.timewindowForComparison;
}

2
ui-ngx/src/app/core/services/menu.service.ts

@ -219,7 +219,7 @@ export class MenuService {
{
name: 'resource.resources-library',
icon: 'folder',
path: '/resources-library'
path: '/settings/resources-library'
}
]
}

2
ui-ngx/src/app/modules/home/components/device/device-credentials-mqtt-basic.component.ts

@ -59,7 +59,7 @@ export class DeviceCredentialsMqttBasicComponent implements ControlValueAccessor
constructor(public fb: FormBuilder) {
this.deviceCredentialsMqttFormGroup = this.fb.group({
clientId: [null, [Validators.pattern(/^[A-Za-z0-9]+$/)]],
clientId: [null],
userName: [null],
password: [null]
}, {validators: this.atLeastOne(Validators.required, ['clientId', 'userName'])});

1
ui-ngx/src/app/modules/home/components/profile/add-device-profile-dialog.component.ts

@ -187,6 +187,7 @@ export class AddDeviceProfileDialogComponent extends
name: this.deviceProfileDetailsFormGroup.get('name').value,
type: this.deviceProfileDetailsFormGroup.get('type').value,
image: this.deviceProfileDetailsFormGroup.get('image').value,
defaultQueueName: this.deviceProfileDetailsFormGroup.get('defaultQueueName').value,
transportType: this.transportConfigFormGroup.get('transportType').value,
provisionType: deviceProvisionConfiguration.type,
provisionDeviceKey,

6
ui-ngx/src/app/modules/home/components/profile/device/common/power-mode-setting.component.html

@ -29,14 +29,14 @@
<tb-time-unit-select [labelText]="'device-profile.edrx-cycle'"
[requiredText]="'device-profile.edrx-cycle-required'"
[patternText]="'device-profile.edrx-cycle-pattern'"
[minTime]="20480"
[minTime]="5120"
[minText]="'device-profile.edrx-cycle-min'"
formControlName="edrxCycle">
</tb-time-unit-select>
<tb-time-unit-select [labelText]="'device-profile.paging-transmission-window'"
[requiredText]="'device-profile.paging-transmission-window-required'"
[patternText]="'device-profile.paging-transmission-window-pattern'"
[minTime]="2000"
[minTime]="1280"
[minText]="'device-profile.paging-transmission-window-min'"
formControlName="pagingTransmissionWindow">
</tb-time-unit-select>
@ -45,7 +45,7 @@
[labelText]="'device-profile.psm-activity-timer'"
[requiredText]="'device-profile.psm-activity-timer-required'"
[patternText]="'device-profile.psm-activity-timer-pattern'"
[minTime]="2000"
[minTime]="1280"
[minText]="'device-profile.psm-activity-timer-min'"
formControlName="psmActivityTimer">
</tb-time-unit-select>

14
ui-ngx/src/app/modules/home/components/widget/lib/flot-widget.models.ts

@ -145,6 +145,7 @@ export interface TbFlotComparisonSettings {
comparisonEnabled: boolean;
timeForComparison: ComparisonDuration;
xaxisSecond: TbFlotSecondXAxisSettings;
comparisonCustomIntervalValue?: number;
}
export interface TbFlotThresholdsSettings {
@ -546,6 +547,11 @@ const chartSettingsSchemaForComparison: JsonSettingsSchema = {
type: 'string',
default: 'previousInterval'
},
comparisonCustomIntervalValue: {
title: 'Custom interval value (ms)',
type: 'number',
default: 7200000
},
xaxisSecond: {
title: 'Second X axis',
type: 'object',
@ -596,9 +602,17 @@ const chartSettingsSchemaForComparison: JsonSettingsSchema = {
{
value: 'years',
label: 'Year ago'
},
{
value: 'customInterval',
label: 'Custom interval'
}
]
},
{
key: 'comparisonCustomIntervalValue',
condition: 'model.timeForComparison === "customInterval"'
},
{
key: 'xaxisSecond',
items: [

18
ui-ngx/src/app/modules/home/components/widget/lib/maps/common-maps-utils.ts

@ -16,7 +16,7 @@
import { FormattedData, MapProviders, ReplaceInfo } from '@home/components/widget/lib/maps/map-models';
import {
createLabelFromDatasource,
createLabelFromDatasource, deepClone,
hashCode,
isDefined,
isDefinedAndNotNull,
@ -343,6 +343,22 @@ export function parseData(input: DatasourceData[]): FormattedData[] {
});
}
export function flatData(input: FormattedData[]): FormattedData {
let result: FormattedData = {} as FormattedData;
if (input.length) {
for (const toMerge of input) {
result = {...result, ...toMerge};
}
result.entityName = input[0].entityName;
result.entityId = input[0].entityId;
result.entityType = input[0].entityType;
result.$datasource = input[0].$datasource;
result.dsIndex = input[0].dsIndex;
result.deviceType = input[0].deviceType;
}
return result;
}
export function parseArray(input: DatasourceData[]): FormattedData[][] {
return _(input).groupBy(el => el?.datasource?.entityName)
.values().value().map((entityArray) =>

18
ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.html

@ -0,0 +1,18 @@
<!--
Copyright © 2016-2021 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.
-->
<markdown [data]="markdownText" class="tb-markdown-view {{markdownClass}}" (click)="markdownClick($event)"></markdown>

130
ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.scss

@ -0,0 +1,130 @@
/**
* Copyright © 2016-2021 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.
*/
.tb-markdown-view {
display: block;
::ng-deep {
h1 {
font-size: 32px;
padding-right: 60px;
}
h1, h2, h3, h4, h5, h6 {
line-height: normal;
font-weight: 500;
margin-bottom: 30px;
padding-bottom: 10px;
border-bottom: 1px solid #ccc;
}
p {
font-size: 16px;
font-weight: 400;
line-height: 1.25em;
margin: 0;
}
p+p {
margin-top: 10px;
}
table {
width: 100%;
border: 1px solid #ccc;
border-spacing: 0;
margin-top: 30px;
margin-bottom: 30px;
}
thead {
background-color: #555;
color: #fff;
}
th, td {
font-size: .85em;
padding: 8px;
margin: 0;
text-align: left;
}
td[align=center], th[align=center] {
text-align: center;
}
td[align=right], th[align=right] {
text-align: right;
}
tr:nth-child(even) {
background-color: #f7f7f7;
}
code:not([class*=language-]) {
background: #f5f5f5;
border-radius: 2px;
color: #dd4a68;
padding: 2px 4px;
}
div.code-wrapper {
position: relative;
button.clipboard-btn {
cursor: pointer;
margin: 0;
border: 0;
outline: none;
position: absolute;
top: 5px;
right: 5px;
background: #fff;
box-shadow: 0 1px 8px 0 rgba(0,0,0,0.2), 0 3px 4px 0 rgba(0,0,0,0.14), 0 3px 3px -2px rgba(0,0,0,0.12);
border-radius: 5px;
opacity: 0;
transition: opacity .3s;
padding: 3px 6px;
line-height: 16px;
img {
width: 18px;
}
&:hover {
background: #f9f9f9;
}
&:active {
background-color: #ececec;
box-shadow: inset 1px -1px 4px 0px rgba(0,0,0,0.4);
}
}
&:hover {
button.clipboard-btn {
opacity: .85;
}
}
}
th, td {
div.code-wrapper {
display: inline-block;
button.clipboard-btn {
top: -5px;
right: -30px;
padding: 0 3px;
}
}
}
}
}

116
ui-ngx/src/app/modules/home/components/widget/lib/markdown-widget.component.ts

@ -0,0 +1,116 @@
///
/// Copyright © 2016-2021 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.
///
import { ChangeDetectorRef, Component, Input, OnInit } from '@angular/core';
import { PageComponent } from '@shared/components/page.component';
import { WidgetContext } from '@home/models/widget-component.models';
import { Store } from '@ngrx/store';
import { AppState } from '@core/core.state';
import { DatasourceData } from '@shared/models/widget.models';
import { DataKeyType } from '@shared/models/telemetry/telemetry.models';
import {
fillPattern, flatData,
parseData,
parseFunction,
processPattern,
safeExecute
} from '@home/components/widget/lib/maps/common-maps-utils';
import { FormattedData } from '@home/components/widget/lib/maps/map-models';
import { hashCode, isNotEmptyStr } from '@core/utils';
import cssjs from '@core/css/css';
interface MarkdownWidgetSettings {
markdownTextPattern: string;
useMarkdownTextFunction: boolean;
markdownTextFunction: string;
markdownCss: string;
}
type MarkdownTextFunction = (data: FormattedData[]) => string;
@Component({
selector: 'tb-markdown-widget ',
templateUrl: './markdown-widget.component.html',
styleUrls: ['./markdown-widget.component.scss']
})
export class MarkdownWidgetComponent extends PageComponent implements OnInit {
settings: MarkdownWidgetSettings;
markdownTextFunction: MarkdownTextFunction;
@Input()
ctx: WidgetContext;
markdownText: string;
markdownClass: string;
constructor(protected store: Store<AppState>,
private cd: ChangeDetectorRef) {
super(store);
}
ngOnInit(): void {
this.ctx.$scope.markdownWidget = this;
this.settings = this.ctx.settings;
this.markdownTextFunction = this.settings.useMarkdownTextFunction ? parseFunction(this.settings.markdownTextFunction, ['data']) : null;
this.markdownClass = 'markdown-widget';
const cssString = this.settings.markdownCss;
if (isNotEmptyStr(cssString)) {
const cssParser = new cssjs();
cssParser.testMode = false;
this.markdownClass += '-' + hashCode(cssString);
cssParser.cssPreviewNamespace = 'tb-markdown-view.' + this.markdownClass;
cssParser.createStyleElement(this.markdownClass, cssString);
}
}
public onDataUpdated() {
let initialData: DatasourceData[];
if (this.ctx.data?.length) {
initialData = this.ctx.data;
} else if (this.ctx.datasources?.length) {
initialData = [
{
datasource: this.ctx.datasources[0],
dataKey: {
type: DataKeyType.attribute,
name: 'empty'
},
data: []
}
];
}
let markdownText: string;
if (initialData) {
const data = parseData(initialData);
markdownText = this.settings.useMarkdownTextFunction ?
safeExecute(this.markdownTextFunction, [data]) : this.settings.markdownTextPattern;
const allData = flatData(data);
const replaceInfo = processPattern(markdownText, allData);
markdownText = fillPattern(markdownText, replaceInfo, allData);
}
if (this.markdownText !== markdownText) {
this.markdownText = markdownText;
this.cd.detectChanges();
}
}
markdownClick($event: MouseEvent) {
this.ctx.actionsApi.elementClick($event);
}
}

3
ui-ngx/src/app/modules/home/components/widget/lib/multiple-input-widget.component.ts

@ -162,7 +162,8 @@ export class MultipleInputWidgetComponent extends PageComponent implements OnIni
private initializeConfig() {
if (this.settings.widgetTitle && this.settings.widgetTitle.length) {
this.ctx.widgetTitle = this.utils.customTranslation(this.settings.widgetTitle, this.settings.widgetTitle);
const titlePatternText = this.utils.customTranslation(this.settings.widgetTitle, this.settings.widgetTitle);
this.ctx.widgetTitle = createLabelFromDatasource(this.datasources[0], titlePatternText);
} else {
this.ctx.widgetTitle = this.ctx.widgetConfig.title;
}

2
ui-ngx/src/app/modules/home/components/widget/lib/qrcode-widget.component.ts

@ -101,7 +101,7 @@ export class QrCodeWidgetComponent extends PageComponent implements OnInit, Afte
const dataSourceData = data[0];
const pattern = this.settings.useQrCodeTextFunction ?
safeExecute(this.qrCodeTextFunction, [dataSourceData]) : this.settings.qrCodeTextPattern;
const replaceInfo = processPattern(pattern, data);
const replaceInfo = processPattern(pattern, dataSourceData);
qrCodeText = fillPattern(pattern, replaceInfo, dataSourceData);
}
this.updateQrCodeText(qrCodeText);

7
ui-ngx/src/app/modules/home/components/widget/widget-components.module.ts

@ -40,6 +40,7 @@ import { NavigationCardWidgetComponent } from '@home/components/widget/lib/navig
import { EdgesOverviewWidgetComponent } from '@home/components/widget/lib/edges-overview-widget.component';
import { JsonInputWidgetComponent } from '@home/components/widget/lib/json-input-widget.component';
import { QrCodeWidgetComponent } from '@home/components/widget/lib/qrcode-widget.component';
import { MarkdownWidgetComponent } from '@home/components/widget/lib/markdown-widget.component';
@NgModule({
declarations:
@ -60,7 +61,8 @@ import { QrCodeWidgetComponent } from '@home/components/widget/lib/qrcode-widget
GatewayFormComponent,
NavigationCardsWidgetComponent,
NavigationCardWidgetComponent,
QrCodeWidgetComponent
QrCodeWidgetComponent,
MarkdownWidgetComponent
],
imports: [
CommonModule,
@ -83,7 +85,8 @@ import { QrCodeWidgetComponent } from '@home/components/widget/lib/qrcode-widget
GatewayFormComponent,
NavigationCardsWidgetComponent,
NavigationCardWidgetComponent,
QrCodeWidgetComponent
QrCodeWidgetComponent,
MarkdownWidgetComponent
],
providers: [
CustomDialogService,

3
ui-ngx/src/app/modules/home/components/widget/widget.component.ts

@ -908,7 +908,8 @@ export class WidgetComponent extends PageComponent implements OnInit, AfterViewI
warnOnPageDataOverflow: this.typeParameters.warnOnPageDataOverflow,
ignoreDataUpdateOnIntervalTick: this.typeParameters.ignoreDataUpdateOnIntervalTick,
comparisonEnabled: comparisonSettings.comparisonEnabled,
timeForComparison: comparisonSettings.timeForComparison
timeForComparison: comparisonSettings.timeForComparison,
comparisonCustomIntervalValue: comparisonSettings.comparisonCustomIntervalValue
};
if (this.widget.type === widgetType.alarm) {
options.alarmSource = deepClone(this.widget.config.alarmSource);

1
ui-ngx/src/app/modules/home/components/wizard/device-wizard-dialog.component.ts

@ -289,6 +289,7 @@ export class DeviceWizardDialogComponent extends
const deviceProfile: DeviceProfile = {
name: this.deviceWizardFormGroup.get('newDeviceProfileTitle').value,
type: DeviceProfileType.DEFAULT,
defaultQueueName: this.deviceWizardFormGroup.get('defaultQueueName').value,
transportType: this.transportConfigFormGroup.get('transportType').value,
provisionType: deviceProvisionConfiguration.type,
provisionDeviceKey,

11
ui-ngx/src/app/modules/home/pages/user/user.component.html

@ -53,6 +53,17 @@
[fxShow]="!hideDelete() && !isEdit">
{{'user.delete' | translate }}
</button>
<div fxLayout="row" fxLayout.xs="column">
<button mat-raised-button
ngxClipboard
(cbOnSuccess)="onUserIdCopied($event)"
[cbContent]="entity?.id?.id"
[disabled]="(isLoading$ | async)"
[fxShow]="!isEdit">
<mat-icon svgIcon="mdi:clipboard-arrow-left"></mat-icon>
<span translate>user.copyId</span>
</button>
</div>
</div>
<div class="mat-padding" fxLayout="column">
<form [formGroup]="entityForm">

17
ui-ngx/src/app/modules/home/pages/user/user.component.ts

@ -25,6 +25,8 @@ import { map } from 'rxjs/operators';
import { Authority } from '@shared/models/authority.enum';
import { isDefinedAndNotNull, isUndefined } from '@core/utils';
import { EntityTableConfig } from '@home/models/entity/entities-table-config.models';
import { ActionNotificationShow } from '@app/core/notification/notification.actions';
import { TranslateService } from '@ngx-translate/core';
@Component({
selector: 'tb-user',
@ -44,7 +46,8 @@ export class UserComponent extends EntityComponent<User> {
@Optional() @Inject('entity') protected entityValue: User,
@Optional() @Inject('entitiesTableConfig') protected entitiesTableConfigValue: EntityTableConfig<User>,
public fb: FormBuilder,
protected cd: ChangeDetectorRef) {
protected cd: ChangeDetectorRef,
protected translate: TranslateService) {
super(store, fb, entityValue, entitiesTableConfigValue, cd);
}
@ -100,4 +103,16 @@ export class UserComponent extends EntityComponent<User> {
isDefinedAndNotNull(entity.additionalInfo.homeDashboardHideToolbar) ? entity.additionalInfo.homeDashboardHideToolbar : true}});
}
onUserIdCopied($event) {
this.store.dispatch(new ActionNotificationShow(
{
message: this.translate.instant('user.idCopiedMessage'),
type: 'success',
duration: 750,
verticalPosition: 'bottom',
horizontalPosition: 'right'
}
))
}
}

9
ui-ngx/src/app/shared/components/json-form/json-form.component.ts

@ -15,6 +15,7 @@
///
import {
ChangeDetectorRef,
Component,
ElementRef,
forwardRef,
@ -113,7 +114,8 @@ export class JsonFormComponent implements OnInit, ControlValueAccessor, Validato
constructor(public elementRef: ElementRef,
private translate: TranslateService,
private dialogs: DialogService,
protected store: Store<AppState>) {
protected store: Store<AppState>,
private cd: ChangeDetectorRef) {
}
ngOnInit(): void {
@ -216,8 +218,8 @@ export class JsonFormComponent implements OnInit, ControlValueAccessor, Validato
}
private onIconClick(key: (string | number)[],
val: string,
iconSelectedFn: (icon: string) => void) {
val: string,
iconSelectedFn: (icon: string) => void) {
this.dialogs.materialIconPicker(val).subscribe((icon) => {
if (icon && iconSelectedFn) {
iconSelectedFn(icon);
@ -229,6 +231,7 @@ export class JsonFormComponent implements OnInit, ControlValueAccessor, Validato
this.targetFullscreenElement = element;
this.isFullscreen = !this.isFullscreen;
this.fullscreenFinishFn = fullscreenFinishFn;
this.cd.markForCheck();
}
onFullscreenChanged(fullscreen: boolean) {

4
ui-ngx/src/app/shared/components/json-form/react/json-form-ace-editor.tsx

@ -159,8 +159,8 @@ class ThingsboardAceEditor extends React.Component<ThingsboardAceEditorProps, Th
<div className='json-form-ace-editor'>
<div className='title-panel'>
<label>{this.props.mode}</label>
<Button style={ styles.tidyButtonStyle }
className='tidy-button' onClick={this.onTidy}>Tidy</Button>
{ this.props.onTidy ? <Button style={ styles.tidyButtonStyle }
className='tidy-button' onClick={this.onTidy}>Tidy</Button> : null }
<Button style={ styles.tidyButtonStyle }
className='tidy-button' onClick={this.onToggleFull}>
{this.state.isFull ?

33
ui-ngx/src/app/shared/components/json-form/react/json-form-markdown.tsx

@ -0,0 +1,33 @@
/*
* Copyright © 2016-2021 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.
*/
import * as React from 'react';
import ThingsboardAceEditor from './json-form-ace-editor';
import { JsonFormFieldProps, JsonFormFieldState } from '@shared/components/json-form/react/json-form.models';
class ThingsboardMarkdown extends React.Component<JsonFormFieldProps, JsonFormFieldState> {
constructor(props) {
super(props);
}
render() {
return (
<ThingsboardAceEditor {...this.props} mode='markdown' {...this.state}></ThingsboardAceEditor>
);
}
}
export default ThingsboardMarkdown;

4
ui-ngx/src/app/shared/components/json-form/react/json-form-schema-form.tsx

@ -38,6 +38,7 @@ import { JsonFormData, JsonFormProps, onChangeFn, OnColorClickFn, OnIconClickFn
import _ from 'lodash';
import * as tinycolor_ from 'tinycolor2';
import { GroupInfo } from '@shared/models/widget.models';
import ThingsboardMarkdown from '@shared/components/json-form/react/json-form-markdown';
const tinycolor = tinycolor_;
@ -65,6 +66,7 @@ class ThingsboardSchemaForm extends React.Component<JsonFormProps, any> {
json: ThingsboardJson,
html: ThingsboardHtml,
css: ThingsboardCss,
markdown: ThingsboardMarkdown,
color: ThingsboardColor,
'rc-select': ThingsboardRcSelect,
fieldset: ThingsboardFieldSet,
@ -91,7 +93,7 @@ class ThingsboardSchemaForm extends React.Component<JsonFormProps, any> {
}
onIconClick(key: (string | number)[], val: string,
iconSelectedFn: (icon: string) => void) {
iconSelectedFn: (icon: string) => void) {
this.props.onIconClick(key, val, iconSelectedFn);
}

2
ui-ngx/src/app/shared/components/led-light.component.ts

@ -73,7 +73,7 @@ export class LedLightComponent implements OnInit, AfterViewInit, OnChanges {
for (const propName of Object.keys(changes)) {
const change = changes[propName];
if (!change.firstChange && change.currentValue !== change.previousValue) {
if (propName === 'enabled') {
if (propName === 'enabled' && this.circleElement) {
this.draw();
} else if (propName === 'size') {
this.update();

99
ui-ngx/src/app/shared/components/markdown.factory.ts

@ -0,0 +1,99 @@
///
/// Copyright © 2016-2021 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.
///
import { MarkedOptions, MarkedRenderer } from 'ngx-markdown';
export function markedOptionsFactory(): MarkedOptions {
const renderer = new MarkedRenderer();
const renderer2 = new MarkedRenderer();
const copyCodeBlock = '{:copy-code}';
let id = 1;
renderer.code = (code: string, language: string | undefined, isEscaped: boolean) => {
if (code.endsWith(copyCodeBlock)) {
code = code.substring(0, code.length - copyCodeBlock.length);
const content = renderer2.code(code, language, isEscaped);
id++;
return wrapCopyCode(id, content, code);
} else {
return renderer2.code(code, language, isEscaped);
}
};
renderer.tablecell = (content: string, flags: {
header: boolean;
align: 'center' | 'left' | 'right' | null;
}) => {
if (content.endsWith(copyCodeBlock)) {
content = content.substring(0, content.length - copyCodeBlock.length);
id++;
content = wrapCopyCode(id, content, content);
}
return renderer2.tablecell(content, flags);
};
return {
renderer,
headerIds: true,
gfm: true,
breaks: false,
pedantic: false,
smartLists: true,
smartypants: false,
};
}
function wrapCopyCode(id: number, content: string, code: string): string {
return '<div class="code-wrapper">' + content + '<span id="copyCodeId' + id + '" style="display: none;">' + code + '</span>' +
'<button id="copyCodeBtn' + id + '" onClick="markdownCopyCode(' + id + ')" ' +
'class="clipboard-btn"><img src="https://clipboardjs.com/assets/images/clippy.svg" alt="Copy to clipboard">' +
'</button></div>';
}
(window as any).markdownCopyCode = (id: number) => {
const text = $('#copyCodeId' + id).text();
navigator.clipboard.writeText(text).then(() => {
import('tooltipster').then(
() => {
const copyBtn = $('#copyCodeBtn' + id);
if (!copyBtn.hasClass('tooltipstered')) {
copyBtn.tooltipster(
{
content: 'Copied!',
theme: 'tooltipster-shadow',
delay: 0,
trigger: 'custom',
triggerClose: {
click: true,
tap: true,
scroll: true,
mouseleave: true
},
side: 'bottom',
distance: 12,
trackOrigin: true
}
);
}
const tooltip = copyBtn.tooltipster('instance');
tooltip.open();
}
);
});
};

7
ui-ngx/src/app/shared/components/queue/queue-type-list.component.html

@ -34,7 +34,12 @@
#queueAutocomplete="matAutocomplete"
[displayWith]="displayQueueFn">
<mat-option *ngFor="let queue of filteredQueues | async" [value]="queue">
<span [innerHTML]="queue | highlight:searchText"></span>
<span [innerHTML]="queue.queueName | highlight:searchText"></span>
</mat-option>
<mat-option *ngIf="!(filteredQueues | async)?.length" [value]="null">
<span>
{{ translate.get('queue.no-queues-matching', {queue: searchText}) | async }}
</span>
</mat-option>
</mat-autocomplete>
<mat-error *ngIf="queueFormGroup.get('queue').hasError('required')">

52
ui-ngx/src/app/shared/components/queue/queue-type-list.component.ts

@ -34,6 +34,10 @@ import { coerceBooleanProperty } from '@angular/cdk/coercion';
import { QueueService } from '@core/http/queue.service';
import { ServiceType } from '@shared/models/queue.models';
interface Queue {
queueName: string;
}
@Component({
selector: 'tb-queue-type-list',
templateUrl: './queue-type-list.component.html',
@ -48,7 +52,7 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
queueFormGroup: FormGroup;
modelValue: string | null;
modelValue: Queue | null;
private requiredValue: boolean;
get required(): boolean {
@ -67,9 +71,9 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
@ViewChild('queueInput', {static: true}) queueInput: ElementRef<HTMLInputElement>;
filteredQueues: Observable<Array<string>>;
filteredQueues: Observable<Array<Queue>>;
queues: Observable<Array<string>>;
queues: Observable<Array<Queue>>;
searchText = '';
@ -99,9 +103,15 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
debounceTime(150),
distinctUntilChanged(),
tap(value => {
this.updateView(value);
let modelValue;
if (typeof value === 'string' || !value) {
modelValue = null;
} else {
modelValue = value;
}
this.updateView(modelValue);
}),
map(value => value ? value : ''),
map(value => value ? (typeof value === 'string' ? value : value.queueName) : ''),
switchMap(queue => this.fetchQueues(queue) )
);
}
@ -123,8 +133,8 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
writeValue(value: string | null): void {
this.searchText = '';
this.modelValue = value;
this.queueFormGroup.get('queue').patchValue(value, {emitEvent: false});
this.modelValue = value ? { queueName: value } : null;
this.queueFormGroup.get('queue').patchValue(this.modelValue, {emitEvent: false});
this.dirty = true;
}
@ -135,42 +145,42 @@ export class QueueTypeListComponent implements ControlValueAccessor, OnInit, Aft
}
}
updateView(value: string | null) {
updateView(value: Queue | null) {
if (this.modelValue !== value) {
this.modelValue = value;
this.propagateChange(this.modelValue);
this.propagateChange(this.modelValue ? this.modelValue.queueName : null);
}
}
displayQueueFn(queue?: string): string | undefined {
return queue ? queue : undefined;
displayQueueFn(queue?: Queue): string | undefined {
return queue ? queue.queueName : undefined;
}
fetchQueues(searchText?: string): Observable<Array<string>> {
fetchQueues(searchText?: string): Observable<Array<Queue>> {
this.searchText = searchText;
return this.getQueues().pipe(
catchError(() => of([])),
catchError(() => of([] as Array<Queue>)),
map(queues => {
const result = queues.filter( queue => {
return searchText ? queue.toUpperCase().startsWith(searchText.toUpperCase()) : true;
return searchText ? queue.queueName.toUpperCase().startsWith(searchText.toUpperCase()) : true;
});
if (result.length) {
if (searchText && searchText.length && result.indexOf(searchText) === -1) {
result.push(searchText);
}
result.sort();
} else if (searchText && searchText.length) {
result.push(searchText);
result.sort((q1, q2) => q1.queueName.localeCompare(q2.queueName));
}
return result;
})
);
}
getQueues(): Observable<Array<string>> {
getQueues(): Observable<Array<Queue>> {
if (!this.queues) {
this.queues = this.queueService.
getTenantQueuesByServiceType(this.queueType, {ignoreLoading: true}).pipe(
map((queues) => {
return queues.map((queueName) => {
return { queueName };
});
}),
publishReplay(1),
refCount()
);

6
ui-ngx/src/app/shared/components/time/timewindow-panel.component.scss

@ -79,4 +79,10 @@
}
}
}
.mat-slider-horizontal .mat-slider-thumb-label {
width: 38px;
height: 38px;
top: -46px;
right: -19px;
}
}

15
ui-ngx/src/app/shared/models/time/time.models.ts

@ -15,7 +15,7 @@
///
import { TimeService } from '@core/services/time.service';
import { deepClone, isDefined, isUndefined } from '@app/core/utils';
import { deepClone, isDefined, isNumeric, isUndefined } from '@app/core/utils';
import * as moment_ from 'moment';
import * as momentTz from 'moment-timezone';
@ -28,7 +28,7 @@ export const DAY = 24 * HOUR;
export const WEEK = 7 * DAY;
export const YEAR = DAY * 365;
export type ComparisonDuration = moment_.unitOfTime.DurationConstructor | 'previousInterval';
export type ComparisonDuration = moment_.unitOfTime.DurationConstructor | 'previousInterval' | 'customInterval';
export enum TimewindowType {
REALTIME,
@ -640,7 +640,7 @@ export function calculateIntervalComparisonEndTime(interval: QuickTimeInterval,
}
export function createTimewindowForComparison(subscriptionTimewindow: SubscriptionTimewindow,
timeUnit: ComparisonDuration): SubscriptionTimewindow {
timeUnit: ComparisonDuration, customIntervalValue: number): SubscriptionTimewindow {
const timewindowForComparison: SubscriptionTimewindow = {
fixedWindow: null,
realtimeWindowMs: null,
@ -667,6 +667,15 @@ export function createTimewindowForComparison(subscriptionTimewindow: Subscripti
endTimeMs = subscriptionTimewindow.fixedWindow.startTimeMs;
startTimeMs = endTimeMs - timeInterval;
}
} else if (timeUnit === 'customInterval') {
if (isNumeric(customIntervalValue) && isFinite(customIntervalValue) && customIntervalValue > 0) {
const timeInterval = subscriptionTimewindow.fixedWindow.endTimeMs - subscriptionTimewindow.fixedWindow.startTimeMs;
endTimeMs = subscriptionTimewindow.fixedWindow.endTimeMs - Math.round(customIntervalValue);
startTimeMs = endTimeMs - timeInterval;
} else {
endTimeMs = subscriptionTimewindow.fixedWindow.endTimeMs;
startTimeMs = subscriptionTimewindow.fixedWindow.startTimeMs;
}
} else {
const timeInterval = subscriptionTimewindow.fixedWindow.endTimeMs - subscriptionTimewindow.fixedWindow.startTimeMs;
endTimeMs = moment(subscriptionTimewindow.fixedWindow.endTimeMs).subtract(1, timeUnit).valueOf();

1
ui-ngx/src/app/shared/models/widget.models.ts

@ -467,6 +467,7 @@ export interface WidgetActionDescriptor extends CustomActionDescriptor {
export interface WidgetComparisonSettings {
comparisonEnabled?: boolean;
timeForComparison?: moment_.unitOfTime.DurationConstructor;
comparisonCustomIntervalValue?: number;
}
export interface WidgetConfig {

15
ui-ngx/src/app/shared/shared.module.ts

@ -14,7 +14,7 @@
/// limitations under the License.
///
import { NgModule } from '@angular/core';
import { NgModule, SecurityContext } from '@angular/core';
import { CommonModule, DatePipe } from '@angular/common';
import { FooterComponent } from '@shared/components/footer.component';
import { LogoComponent } from '@shared/components/logo.component';
@ -78,6 +78,7 @@ import { DatetimePeriodComponent } from '@shared/components/time/datetime-period
import { EnumToArrayPipe } from '@shared/pipe/enum-to-array.pipe';
import { ClipboardModule } from 'ngx-clipboard';
import { ValueInputComponent } from '@shared/components/value-input.component';
import { MarkdownModule, MarkedOptions } from 'ngx-markdown';
import { FullscreenDirective } from '@shared/components/fullscreen.directive';
import { HighlightPipe } from '@shared/pipe/highlight.pipe';
import { DashboardAutocompleteComponent } from '@shared/components/dashboard-autocomplete.component';
@ -145,6 +146,7 @@ import { OtaPackageAutocompleteComponent } from '@shared/components/ota-package/
import { MAT_DATE_LOCALE } from '@angular/material/core';
import { CopyButtonComponent } from '@shared/components/button/copy-button.component';
import { TogglePasswordComponent } from '@shared/components/button/toggle-password.component';
import { markedOptionsFactory } from '@shared/components/markdown.factory';
@NgModule({
providers: [
@ -294,7 +296,15 @@ import { TogglePasswordComponent } from '@shared/components/button/toggle-passwo
NgxHmCarouselModule,
DndModule,
NgxFlowModule,
NgxFlowchartModule
NgxFlowchartModule,
// ngx-markdown
MarkdownModule.forRoot({
sanitize: SecurityContext.NONE,
markedOptions: {
provide: MarkedOptions,
useFactory: markedOptionsFactory
}
})
],
exports: [
FooterComponent,
@ -386,6 +396,7 @@ import { TogglePasswordComponent } from '@shared/components/button/toggle-passwo
NgxHmCarouselModule,
DndModule,
NgxFlowchartModule,
MarkdownModule,
ConfirmDialogComponent,
AlertDialogComponent,
TodoDialogComponent,

10
ui-ngx/src/assets/locale/locale.constant-en_US.json

@ -1082,7 +1082,7 @@
"default-rule-chain": "Default rule chain",
"mobile-dashboard": "Mobile dashboard",
"mobile-dashboard-hint": "Used by mobile application as a device details dashboard",
"select-queue-hint": "Select from a drop-down list or add a custom name.",
"select-queue-hint": "Select from a drop-down list.",
"delete-device-profile-title": "Are you sure you want to delete the device profile '{{deviceProfileName}}'?",
"delete-device-profile-text": "Be careful, after the confirmation the device profile and all related data will become unrecoverable.",
"delete-device-profiles-title": "Are you sure you want to delete { count, plural, 1 {1 device profile} other {# device profiles} }?",
@ -2263,6 +2263,7 @@
"total": "total",
"comparison-time-ago": {
"previousInterval": "(previous interval)",
"customInterval": "(custom interval)",
"days": "(day ago)",
"weeks": "(week ago)",
"months": "(month ago)",
@ -2569,7 +2570,8 @@
"queue": {
"select_name": "Select queue name",
"name": "Queue Name",
"name_required": "Queue name is required!"
"name_required": "Queue name is required!",
"no-queues-matching": "No queues matching '{{queue}}' were found."
},
"tenant": {
"tenant": "Tenant",
@ -2807,7 +2809,9 @@
"disable-account": "Disable User Account",
"enable-account": "Enable User Account",
"enable-account-message": "User account was successfully enabled!",
"disable-account-message": "User account was successfully disabled!"
"disable-account-message": "User account was successfully disabled!",
"copyId": "Copy user Id",
"idCopiedMessage": "User Id has been copied to clipboard"
},
"value": {
"type": "Value type",

2
ui-ngx/src/assets/locale/locale.constant-ko_KR.json

@ -959,7 +959,7 @@
"profile-configuration": "프로파일 설정",
"transport-configuration": "전송 설정",
"default-rule-chain": "기본 규칙 사슬",
"select-queue-hint": "Select from a drop-down list or add a custom name.",
"select-queue-hint": "Select from a drop-down list.",
"delete-device-profile-title": "Are you sure you want to delete the device profile '{{deviceProfileName}}'?",
"delete-device-profile-text": "Be careful, after the confirmation the device profile and all related data will become unrecoverable.",
"delete-device-profiles-title": "Are you sure you want to delete { count, plural, 1 {1 개 장치 프로파일} other {# 개 장치 프로파일} }?",

4
ui-ngx/src/assets/locale/locale.constant-ru_RU.json

@ -1510,7 +1510,9 @@
"disable-account": "Отключить учетную запись пользователя",
"enable-account": "Включить учетную запись пользователя",
"enable-account-message": "Учетная запись пользователя была успешно включена!",
"disable-account-message": "Учетная запись пользователя была успешно отключена!"
"disable-account-message": "Учетная запись пользователя была успешно отключена!",
"copyId": "Копировать ИД пользователя",
"idCopiedMessage": "ИД пользователя скопирован в буфер обмена"
},
"value": {
"type": "Тип значения",

2
ui-ngx/src/assets/locale/locale.constant-sl_SI.json

@ -959,7 +959,7 @@
"profile-configuration": "Profile configuration",
"transport-configuration": "Transport configuration",
"default-rule-chain": "Default rule chain",
"select-queue-hint": "Select from a drop-down list or add a custom name.",
"select-queue-hint": "Select from a drop-down list.",
"delete-device-profile-title": "Are you sure you want to delete the device profile '{{deviceProfileName}}'?",
"delete-device-profile-text": "Be careful, after the confirmation the device profile and all related data will become unrecoverable.",
"delete-device-profiles-title": "Are you sure you want to delete { count, plural, 1 {1 device profile} other {# device profiles} }?",

4
ui-ngx/src/assets/locale/locale.constant-uk_UA.json

@ -2078,7 +2078,9 @@
"disable-account": "Вимкнути обліковий запис користувача",
"enable-account": "Увімкнути обліковий запис користувача",
"enable-account-message": "Обліковий запис користувача успішно увімкнено!",
"disabled-account-message": "Обліковий запис користувача успішно вимкнено!"
"disabled-account-message": "Обліковий запис користувача успішно вимкнено!",
"copyId": "Копіювати Id користувача",
"idCopiedMessage": "Id користувача було скопійовано в буфер обміну"
},
"value": {
"type": "Тип значення",

2
ui-ngx/src/tsconfig.app.json

@ -3,7 +3,7 @@
"compilerOptions": {
"outDir": "../out-tsc/app",
"types": ["node", "jquery", "flot", "tooltipster", "tinycolor2", "js-beautify",
"react", "react-dom", "jstree", "raphael", "canvas-gauges", "leaflet", "leaflet.markercluster"]
"react", "react-dom", "jstree", "raphael", "canvas-gauges", "leaflet", "leaflet.markercluster", "leaflet-editable"]
},
"angularCompilerOptions": {
"fullTemplateTypeCheck": true

275
ui-ngx/src/typings/leadflet-editable.d.ts

@ -1,275 +0,0 @@
///
/// Copyright © 2016-2021 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.
///
import * as leaflet from 'leaflet';
declare module 'leaflet' {
/**
* Make geometries editable in Leaflet.
*
* This is not a plug and play UI, and will not. This is a minimal, lightweight, and fully extendable API to
* control editing of geometries. So you can easily build your own UI with your own needs and choices.
*/
interface EditableStatic {
new (map: Map, options: EditOptions): Editable;
}
/**
* Options to pass to L.Editable when instanciating.
*/
interface EditOptions extends leaflet.MapOptions{
/**
* Whether to create a L.Editable instance at map init or not.
*/
editable: boolean;
/**
* Class to be used when creating a new Polyline.
*/
polylineClass?: object;
/**
* Class to be used when creating a new Polygon.
*/
polygonClass?: object;
/**
* Class to be used when creating a new Marker.
*/
markerClass?: object;
/**
* CSS class to be added to the map container while drawing.
*/
drawingCSSClass?: string;
/**
* Layer used to store edit tools (vertex, line guide).
*/
editLayer?: LayerGroup<leaflet.Layer>;
/**
* Default layer used to store drawn features (marker, polyline).
*/
featuresLayer?: LayerGroup<Polyline|Polygon|Marker>;
/**
* Class to be used as vertex, for path editing.
*/
vertexMarkerClass?: object;
/**
* Class to be used as middle vertex, pulled by the user to create a new point in the middle of a path.
*/
middleMarkerClass?: object;
/**
* Class to be used as Polyline editor.
*/
polylineEditorClass?: object;
/**
* Class to be used as Polygon editor.
*/
polygonEditorClass?: object;
/**
* Class to be used as Marker editor.
*/
markerEditorClass?: object;
/**
* Options to be passed to the line guides.
*/
lineGuideOptions?: object;
/**
* Set this to true if you don't want middle markers.
*/
skipMiddleMarkers?: boolean;
}
/**
* Make geometries editable in Leaflet.
*
* This is not a plug and play UI, and will not. This is a minimal, lightweight, and fully extendable API to
* control editing of geometries. So you can easily build your own UI with your own needs and choices.
*/
interface Editable extends leaflet.Evented {
/**
* Options to pass to L.Editable when instanciating.
*/
options: EditOptions;
currentPolygon: Polyline|Polygon|Marker;
/**
* Start drawing a polyline. If latlng is given, a first point will be added. In any case, continuing on user
* click. If options is given, it will be passed to the polyline class constructor.
*/
startPolyline(latLng?: LatLng, options?: PolylineOptions): Polyline;
/**
* Start drawing a polygon. If latlng is given, a first point will be added. In any case, continuing on user
* click. If options is given, it will be passed to the polygon class constructor.
*/
startPolygon(latLng?: LatLng, options?: PolylineOptions): Polygon;
/**
* Start adding a marker. If latlng is given, the marker will be shown first at this point. In any case, it
* will follow the user mouse, and will have a final latlng on next click (or touch). If options is given,
* it will be passed to the marker class constructor.
*/
startMarker(latLng?: LatLng, options?: MarkerOptions): Marker;
/**
* When you need to stop any ongoing drawing, without needing to know which editor is active.
*/
stopDrawing(): void;
/**
* When you need to commit any ongoing drawing, without needing to know which editor is active.
*/
commitDrawing(): void;
}
let Editable: EditableStatic;
/**
* EditableMixin is included to L.Polyline, L.Polygon and L.Marker. It adds the following methods to them.
*
* When editing is enabled, the editor is accessible on the instance with the editor property.
*/
interface EditableMixin {
/**
* Enable editing, by creating an editor if not existing, and then calling enable on it.
*/
enableEdit(map: L.Map): any;
/**
* Disable editing, also remove the editor property reference.
*/
disableEdit(): void;
/**
* Enable or disable editing, according to current status.
*/
toggleEdit(): void;
/**
* Return true if current instance has an editor attached, and this editor is enabled.
*/
editEnabled(): boolean;
}
interface Map {
/**
* Options to pass to L.Editable when instanciating.
*/
editOptions: MapOptions;
/**
* L.Editable plugin instance.
*/
editTools: Editable;
}
// tslint:disable-next-line:no-empty-interface
interface Polyline extends EditableMixin {}
namespace Map {
interface MapOptions {
/**
* Whether to create a L.Editable instance at map init or not.
*/
editable?: boolean;
/**
* Options to pass to L.Editable when instanciating.
*/
editOptions?: MapOptions;
}
}
/**
* When editing a feature (marker, polyline), an editor is attached to it. This editor basically knows
* how to handle the edition.
*/
interface BaseEditor {
/**
* Set up the drawing tools for the feature to be editable.
*/
enable(): MarkerEditor|PolylineEditor|PolygonEditor;
/**
* Remove editing tools.
*/
disable(): MarkerEditor|PolylineEditor|PolygonEditor;
}
/**
* Inherit from L.Editable.BaseEditor.
* Inherited by L.Editable.PolylineEditor and L.Editable.PolygonEditor.
*/
interface PathEditor extends BaseEditor {
/**
* Rebuild edit elements (vertex, middlemarker, etc.).
*/
reset(): void;
}
/**
* Inherit from L.Editable.PathEditor.
*/
interface PolylineEditor extends PathEditor {
/**
* Set up drawing tools to continue the line forward.
*/
continueForward(): void;
/**
* Set up drawing tools to continue the line backward.
*/
continueBackward(): void;
}
/**
* Inherit from L.Editable.PathEditor.
*/
interface PolygonEditor extends PathEditor {
/**
* Set up drawing tools for creating a new hole on the polygon. If the latlng param is given, a first
* point is created.
*/
newHole(latlng: LatLng): void;
}
/**
* Inherit from L.Editable.BaseEditor.
*/
// tslint:disable-next-line:no-empty-interface
interface MarkerEditor extends BaseEditor {}
interface Marker extends EditableMixin, MarkerEditor {}
interface Polyline extends EditableMixin, PolylineEditor {}
interface Polygon extends EditableMixin, PolygonEditor {}
function map(element: string | HTMLElement, options?: EditOptions): Map;
}

3
ui-ngx/tsconfig.json

@ -20,8 +20,7 @@
"src/typings/jquery.flot.typings.d.ts",
"src/typings/jquery.jstree.typings.d.ts",
"src/typings/split.js.typings.d.ts",
"src/typings/add-marker.d.ts",
"src/typings/leaflet-editable.d.ts"
"src/typings/add-marker.d.ts"
],
"paths": {
"@app/*": ["src/app/*"],

Some files were not shown because too many files changed in this diff

Loading…
Cancel
Save