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

Merge pull request #10522 from AndriiLandiak/fix/edge-3.7

Browse Source 
Edge refactoring: using java 17
This commit is contained in:
Viacheslav Klimov 2024-04-10 11:50:38 +03:00 committed by GitHub
commit 756bac5135
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 94 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
View File

@ -19,6 +19,8 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
@ -116,9 +118,6 @@ import org.thingsboard.server.service.telemetry.AlarmSubscriptionService;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import org.thingsboard.server.service.transport.TbCoreToTransportService;
import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.concurrent.ConcurrentHashMap;

45
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -43,7 +43,6 @@ import org.thingsboard.server.common.data.id.RpcId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKey;
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;
@ -74,8 +73,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ClaimDeviceMsg;
import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheEntry;
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg;
import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
import org.thingsboard.server.gen.transport.TransportProtos.KeyValueType;
import org.thingsboard.server.gen.transport.TransportProtos.SessionCloseNotificationProto;
import org.thingsboard.server.gen.transport.TransportProtos.SessionEvent;
import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg;
@ -254,14 +251,12 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
private boolean isSendNewRpcAvailable() {
switch (rpcSubmitStrategy) {
case SEQUENTIAL_ON_ACK_FROM_DEVICE:
return toDeviceRpcPendingMap.values().stream().filter(md -> !md.isDelivered()).findAny().isEmpty();
case SEQUENTIAL_ON_RESPONSE_FROM_DEVICE:
return toDeviceRpcPendingMap.values().stream().filter(ToDeviceRpcRequestMetadata::isDelivered).findAny().isEmpty();
default:
return true;
}
return switch (rpcSubmitStrategy) {
case SEQUENTIAL_ON_ACK_FROM_DEVICE -> toDeviceRpcPendingMap.values().stream().filter(md -> !md.isDelivered()).findAny().isEmpty();
case SEQUENTIAL_ON_RESPONSE_FROM_DEVICE ->
toDeviceRpcPendingMap.values().stream().filter(ToDeviceRpcRequestMetadata::isDelivered).findAny().isEmpty();
default -> true;
};
}
private void createRpc(ToDeviceRpcRequest request, RpcStatus status) {
@ -927,34 +922,6 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
return systemContext.getEdgeEventService().saveAsync(edgeEvent);
}
private KeyValueProto toKeyValueProto(KvEntry kvEntry) {
KeyValueProto.Builder builder = KeyValueProto.newBuilder();
builder.setKey(kvEntry.getKey());
switch (kvEntry.getDataType()) {
case BOOLEAN:
builder.setType(KeyValueType.BOOLEAN_V);
builder.setBoolV(kvEntry.getBooleanValue().get());
break;
case DOUBLE:
builder.setType(KeyValueType.DOUBLE_V);
builder.setDoubleV(kvEntry.getDoubleValue().get());
break;
case LONG:
builder.setType(KeyValueType.LONG_V);
builder.setLongV(kvEntry.getLongValue().get());
break;
case STRING:
builder.setType(KeyValueType.STRING_V);
builder.setStringV(kvEntry.getStrValue().get());
break;
case JSON:
builder.setType(KeyValueType.JSON_V);
builder.setJsonV(kvEntry.getJsonValue().get());
break;
}
return builder.build();
}
void restoreSessions() {
if (systemContext.isLocalCacheType()) {
return;

12
application/src/main/java/org/thingsboard/server/controller/AbstractRpcController.java
View File

@ -122,15 +122,9 @@ public abstract class AbstractRpcController extends BaseController {
logRpcCall(rpcRequest, rpcError, null);
RpcError error = rpcError.get();
switch (error) {
case TIMEOUT:
responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
break;
case NO_ACTIVE_CONNECTION:
responseWriter.setResult(new ResponseEntity<>(noActiveConnectionStatus));
break;
default:
responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
break;
case TIMEOUT -> responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
case NO_ACTIVE_CONNECTION -> responseWriter.setResult(new ResponseEntity<>(noActiveConnectionStatus));
default -> responseWriter.setResult(new ResponseEntity<>(timeoutStatus));
}
} else {
Optional<String> responseData = response.getResponse();

10
application/src/main/java/org/thingsboard/server/controller/TelemetryController.java
View File

@ -25,7 +25,6 @@ import com.google.gson.JsonElement;
import com.google.gson.JsonParseException;
import com.google.gson.JsonParser;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.media.Content;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.responses.ApiResponse;
import io.swagger.v3.oas.annotations.responses.ApiResponses;
@ -36,7 +35,6 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.PathVariable;
@ -49,9 +47,6 @@ import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.context.request.async.DeferredResult;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardThreadFactory;
import org.thingsboard.server.common.data.kv.AggregationParams;
import org.thingsboard.server.common.data.kv.IntervalType;
import org.thingsboard.server.common.msg.rule.engine.DeviceAttributesEventNotificationMsg;
import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.EntityType;
@ -65,6 +60,7 @@ import org.thingsboard.server.common.data.id.EntityIdFactory;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UUIDBased;
import org.thingsboard.server.common.data.kv.Aggregation;
import org.thingsboard.server.common.data.kv.AggregationParams;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
import org.thingsboard.server.common.data.kv.BaseDeleteTsKvQuery;
@ -74,6 +70,7 @@ import org.thingsboard.server.common.data.kv.BooleanDataEntry;
import org.thingsboard.server.common.data.kv.DataType;
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
import org.thingsboard.server.common.data.kv.DoubleDataEntry;
import org.thingsboard.server.common.data.kv.IntervalType;
import org.thingsboard.server.common.data.kv.JsonDataEntry;
import org.thingsboard.server.common.data.kv.KvEntry;
import org.thingsboard.server.common.data.kv.LongDataEntry;
@ -379,7 +376,7 @@ public class TelemetryController extends BaseController {
public DeferredResult<ResponseEntity> saveEntityAttributesV1(
@Parameter(description = ENTITY_TYPE_PARAM_DESCRIPTION, required = true, schema = @Schema(defaultValue = "DEVICE")) @PathVariable("entityType") String entityType,
@Parameter(description = ENTITY_ID_PARAM_DESCRIPTION, required = true) @PathVariable("entityId") String entityIdStr,
@Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE"})) @PathVariable("scope")AttributeScope scope,
@Parameter(description = ATTRIBUTES_SCOPE_DESCRIPTION, schema = @Schema(allowableValues = {"SERVER_SCOPE", "SHARED_SCOPE"})) @PathVariable("scope") AttributeScope scope,
@io.swagger.v3.oas.annotations.parameters.RequestBody(description = ATTRIBUTES_JSON_REQUEST_DESCRIPTION, required = true) @RequestBody JsonNode request) throws ThingsboardException {
EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr);
return saveAttributes(getTenantId(), entityId, scope, request);
@ -918,4 +915,5 @@ public class TelemetryController extends BaseController {
}
return entry.getValue();
}
}

2
application/src/main/java/org/thingsboard/server/service/action/EntityActionService.java
View File

@ -250,7 +250,7 @@ public class EntityActionService {
return result;
}
private void addTimeseries(ObjectNode entityNode, List<TsKvEntry> timeseries) throws Exception {
private void addTimeseries(ObjectNode entityNode, List<TsKvEntry> timeseries) {
if (timeseries != null && !timeseries.isEmpty()) {
ArrayNode result = entityNode.putArray("timeseries");
Map<Long, List<TsKvEntry>> groupedTelemetry = timeseries.stream()

3
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -557,8 +557,7 @@ public final class EdgeGrpcSession implements Closeable {
}
case ATTRIBUTES_UPDATED, POST_ATTRIBUTES, ATTRIBUTES_DELETED, TIMESERIES_UPDATED ->
downlinkMsg = ctx.getTelemetryProcessor().convertTelemetryEventToDownlink(edge, edgeEvent);
default ->
log.warn("[{}][{}] Unsupported action type [{}]", this.tenantId, this.sessionId, edgeEvent.getAction());
default -> log.warn("[{}][{}] Unsupported action type [{}]", this.tenantId, this.sessionId, edgeEvent.getAction());
}
} catch (Exception e) {
log.error("[{}][{}] Exception during converting edge event to downlink msg", this.tenantId, this.sessionId, e);

16
application/src/main/java/org/thingsboard/server/service/edge/rpc/constructor/BaseMsgConstructorFactory.java
View File

@ -31,16 +31,10 @@ public abstract class BaseMsgConstructorFactory<T extends MsgConstructor, U exte
protected U v2Constructor;
public MsgConstructor getMsgConstructorByEdgeVersion(EdgeVersion edgeVersion) {
switch (edgeVersion) {
case V_3_3_0:
case V_3_3_3:
case V_3_4_0:
case V_3_6_0:
case V_3_6_1:
return v1Constructor;
case V_3_6_2:
default:
return v2Constructor;
}
return switch (edgeVersion) {
case V_3_3_0, V_3_3_3, V_3_4_0, V_3_6_0, V_3_6_1 -> v1Constructor;
default -> v2Constructor;
};
}
}

42
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/telemetry/BaseTelemetryProcessor.java
View File

@ -23,10 +23,13 @@ import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.Device;
@ -53,7 +56,6 @@ import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.TbMsgMetaData;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.msg.rule.engine.DeviceAttributesEventNotificationMsg;
import org.thingsboard.server.common.transport.util.JsonUtils;
import org.thingsboard.server.dao.model.ModelConstants;
@ -66,8 +68,6 @@ import org.thingsboard.server.queue.TbQueueProducer;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
import jakarta.annotation.Nullable;
import jakarta.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
@ -317,36 +317,22 @@ public abstract class BaseTelemetryProcessor extends BaseEdgeProcessor {
JsonNode body) {
EntityId entityId;
switch (entityType) {
case DEVICE:
entityId = new DeviceId(entityUUID);
break;
case ASSET:
entityId = new AssetId(entityUUID);
break;
case ENTITY_VIEW:
entityId = new EntityViewId(entityUUID);
break;
case DASHBOARD:
entityId = new DashboardId(entityUUID);
break;
case TENANT:
entityId = TenantId.fromUUID(entityUUID);
break;
case CUSTOMER:
entityId = new CustomerId(entityUUID);
break;
case USER:
entityId = new UserId(entityUUID);
break;
case EDGE:
entityId = new EdgeId(entityUUID);
break;
default:
case DEVICE -> entityId = new DeviceId(entityUUID);
case ASSET -> entityId = new AssetId(entityUUID);
case ENTITY_VIEW -> entityId = new EntityViewId(entityUUID);
case DASHBOARD -> entityId = new DashboardId(entityUUID);
case TENANT -> entityId = TenantId.fromUUID(entityUUID);
case CUSTOMER -> entityId = new CustomerId(entityUUID);
case USER -> entityId = new UserId(entityUUID);
case EDGE -> entityId = new EdgeId(entityUUID);
default -> {
log.warn("[{}] Unsupported edge event type [{}]", tenantId, entityType);
return null;
}
}
String bodyJackson = JacksonUtil.toString(body);
return bodyJackson == null ? null :
entityDataMsgConstructor.constructEntityDataMsg(tenantId, entityId, actionType, JsonParser.parseString(bodyJackson));
}
}

62
application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java
View File

@ -208,52 +208,22 @@ public class AccessValidator {
public void validate(SecurityUser currentUser, Operation operation, EntityId entityId, FutureCallback<ValidationResult> callback) {
switch (entityId.getEntityType()) {
case DEVICE:
validateDevice(currentUser, operation, entityId, callback);
return;
case DEVICE_PROFILE:
validateDeviceProfile(currentUser, operation, entityId, callback);
return;
case ASSET:
validateAsset(currentUser, operation, entityId, callback);
return;
case ASSET_PROFILE:
validateAssetProfile(currentUser, operation, entityId, callback);
return;
case RULE_CHAIN:
validateRuleChain(currentUser, operation, entityId, callback);
return;
case CUSTOMER:
validateCustomer(currentUser, operation, entityId, callback);
return;
case TENANT:
validateTenant(currentUser, operation, entityId, callback);
return;
case TENANT_PROFILE:
validateTenantProfile(currentUser, operation, entityId, callback);
return;
case USER:
validateUser(currentUser, operation, entityId, callback);
return;
case ENTITY_VIEW:
validateEntityView(currentUser, operation, entityId, callback);
return;
case EDGE:
validateEdge(currentUser, operation, entityId, callback);
return;
case API_USAGE_STATE:
validateApiUsageState(currentUser, operation, entityId, callback);
return;
case TB_RESOURCE:
validateResource(currentUser, operation, entityId, callback);
return;
case OTA_PACKAGE:
validateOtaPackage(currentUser, operation, entityId, callback);
return;
case RPC:
validateRpc(currentUser, operation, entityId, callback);
return;
default:
case DEVICE -> validateDevice(currentUser, operation, entityId, callback);
case DEVICE_PROFILE -> validateDeviceProfile(currentUser, operation, entityId, callback);
case ASSET -> validateAsset(currentUser, operation, entityId, callback);
case ASSET_PROFILE -> validateAssetProfile(currentUser, operation, entityId, callback);
case RULE_CHAIN -> validateRuleChain(currentUser, operation, entityId, callback);
case CUSTOMER -> validateCustomer(currentUser, operation, entityId, callback);
case TENANT -> validateTenant(currentUser, operation, entityId, callback);
case TENANT_PROFILE -> validateTenantProfile(currentUser, operation, entityId, callback);
case USER -> validateUser(currentUser, operation, entityId, callback);
case ENTITY_VIEW -> validateEntityView(currentUser, operation, entityId, callback);
case EDGE -> validateEdge(currentUser, operation, entityId, callback);
case API_USAGE_STATE -> validateApiUsageState(currentUser, operation, entityId, callback);
case TB_RESOURCE -> validateResource(currentUser, operation, entityId, callback);
case OTA_PACKAGE -> validateOtaPackage(currentUser, operation, entityId, callback);
case RPC -> validateRpc(currentUser, operation, entityId, callback);
default ->
//TODO: add support of other entities
throw new IllegalStateException("Not Implemented!");
}

19
application/src/main/java/org/thingsboard/server/service/sync/ie/importing/csv/AbstractBulkImportService.java
View File

@ -301,18 +301,13 @@ public abstract class AbstractBulkImportService<E extends HasId<? extends Entity
private final DataType dataType;
public JsonPrimitive toJsonPrimitive() {
switch (dataType) {
case STRING:
return new JsonPrimitive((String) value);
case LONG:
return new JsonPrimitive((Long) value);
case DOUBLE:
return new JsonPrimitive((Double) value);
case BOOLEAN:
return new JsonPrimitive((Boolean) value);
default:
return null;
}
return switch (dataType) {
case STRING -> new JsonPrimitive((String) value);
case LONG -> new JsonPrimitive((Long) value);
case DOUBLE -> new JsonPrimitive((Double) value);
case BOOLEAN -> new JsonPrimitive((Boolean) value);
default -> null;
};
}
public String stringValue() {

1
common/edge-api/src/main/java/org/thingsboard/edge/rpc/EdgeGrpcClient.java
View File

@ -248,4 +248,5 @@ public class EdgeGrpcClient implements EdgeRpcClient {
uplinkMsgLock.unlock();
}
}
}

20
common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
View File

@ -258,31 +258,31 @@ public class ProtoUtils {
.setLastUpdateTs(attributeKvEntry.getLastUpdateTs())
.setKey(attributeKvEntry.getKey());
switch (attributeKvEntry.getDataType()) {
case BOOLEAN:
case BOOLEAN -> {
attributeKvEntry.getBooleanValue().ifPresent(attributeValueBuilder::setBoolV);
attributeValueBuilder.setHasV(attributeKvEntry.getBooleanValue().isPresent());
attributeValueBuilder.setType(TransportProtos.KeyValueType.BOOLEAN_V);
break;
case STRING:
}
case STRING -> {
attributeKvEntry.getStrValue().ifPresent(attributeValueBuilder::setStringV);
attributeValueBuilder.setHasV(attributeKvEntry.getStrValue().isPresent());
attributeValueBuilder.setType(TransportProtos.KeyValueType.STRING_V);
break;
case DOUBLE:
}
case DOUBLE -> {
attributeKvEntry.getDoubleValue().ifPresent(attributeValueBuilder::setDoubleV);
attributeValueBuilder.setHasV(attributeKvEntry.getDoubleValue().isPresent());
attributeValueBuilder.setType(TransportProtos.KeyValueType.DOUBLE_V);
break;
case LONG:
}
case LONG -> {
attributeKvEntry.getLongValue().ifPresent(attributeValueBuilder::setLongV);
attributeValueBuilder.setHasV(attributeKvEntry.getLongValue().isPresent());
attributeValueBuilder.setType(TransportProtos.KeyValueType.LONG_V);
break;
case JSON:
}
case JSON -> {
attributeKvEntry.getJsonValue().ifPresent(attributeValueBuilder::setJsonV);
attributeValueBuilder.setHasV(attributeKvEntry.getJsonValue().isPresent());
attributeValueBuilder.setType(TransportProtos.KeyValueType.JSON_V);
break;
}
}
builder.addValues(attributeValueBuilder.build());
}

17
dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
View File

@ -21,6 +21,7 @@ import com.google.common.base.Function;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import jakarta.annotation.Nullable;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.hibernate.exception.ConstraintViolationException;
@ -31,7 +32,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.event.TransactionalEventListener;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.AttributeScope;
import org.thingsboard.server.common.data.EntitySubtype;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.StringUtils;
@ -71,7 +72,6 @@ import org.thingsboard.server.dao.service.Validator;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.dao.user.UserService;
import jakarta.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
@ -354,7 +354,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
return Futures.successfulAsList(futures);
}, MoreExecutors.directExecutor());
edges = Futures.transform(edges, new Function<List<Edge>, List<Edge>>() {
edges = Futures.transform(edges, new Function<>() {
@Nullable
@Override
public List<Edge> apply(@Nullable List<Edge> edgeList) {
@ -414,7 +414,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
}
private PaginatedRemover<TenantId, Edge> tenantEdgesRemover =
new PaginatedRemover<TenantId, Edge>() {
new PaginatedRemover<>() {
@Override
protected PageData<Edge> findEntities(TenantId tenantId, TenantId id, PageLink pageLink) {
@ -427,7 +427,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
}
};
private PaginatedRemover<CustomerId, Edge> customerEdgeUnassigner = new PaginatedRemover<CustomerId, Edge>() {
private PaginatedRemover<CustomerId, Edge> customerEdgeUnassigner = new PaginatedRemover<>() {
@Override
protected PageData<Edge> findEntities(TenantId tenantId, CustomerId id, PageLink pageLink) {
@ -513,7 +513,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
@Override
public String findMissingToRelatedRuleChains(TenantId tenantId, EdgeId edgeId, String tbRuleChainInputNodeClassName) {
List<RuleChain> edgeRuleChains = findEdgeRuleChains(tenantId, edgeId);
List<RuleChainId> edgeRuleChainIds = edgeRuleChains.stream().map(IdBased::getId).collect(Collectors.toList());
List<RuleChainId> edgeRuleChainIds = edgeRuleChains.stream().map(IdBased::getId).toList();
ObjectNode result = JacksonUtil.newObjectNode();
for (RuleChain edgeRuleChain : edgeRuleChains) {
List<RuleNode> ruleNodes =
@ -522,8 +522,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
List<RuleChainId> connectedRuleChains =
ruleNodes.stream()
.filter(rn -> rn.getType().equals(tbRuleChainInputNodeClassName))
.map(rn -> new RuleChainId(UUID.fromString(rn.getConfiguration().get("ruleChainId").asText())))
.collect(Collectors.toList());
.map(rn -> new RuleChainId(UUID.fromString(rn.getConfiguration().get("ruleChainId").asText()))).toList();
List<String> missingRuleChains = new ArrayList<>();
for (RuleChainId connectedRuleChain : connectedRuleChains) {
if (!edgeRuleChainIds.contains(connectedRuleChain)) {
@ -549,7 +548,7 @@ public class EdgeServiceImpl extends AbstractCachedEntityService<EdgeCacheKey, E
if (persistToTelemetry) {
futureKvEntry = timeseriesService.findLatest(tenantId, edgeId, key);
} else {
futureKvEntry = attributesService.find(tenantId, edgeId, DataConstants.SERVER_SCOPE, key);
futureKvEntry = attributesService.find(tenantId, edgeId, AttributeScope.SERVER_SCOPE, key);
}
return Futures.transformAsync(futureKvEntry, kvEntryOpt ->
Futures.immediateFuture(kvEntryOpt.flatMap(KvEntry::getBooleanValue).orElse(false)), MoreExecutors.directExecutor());

2
rest-client/src/main/java/org/thingsboard/rest/client/utils/RestJsonConverter.java
View File

@ -62,7 +62,7 @@ public class RestJsonConverter {
KvEntry entry = parseValue(key, ts.get(VALUE));
return new BasicTsKvEntry(ts.get(TS).asLong(), entry);
}
).collect(Collectors.toList()))
).toList())
);
return result;
} else {

14
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/AbstractTbMsgPushNode.java
View File

@ -87,24 +87,23 @@ public abstract class AbstractTbMsgPushNode<T extends BaseTbMsgPushNodeConfigura
Map<String, Object> entityBody = new HashMap<>();
JsonNode dataJson = JacksonUtil.toJsonNode(msg.getData());
switch (actionType) {
case ATTRIBUTES_UPDATED:
case POST_ATTRIBUTES:
case ATTRIBUTES_UPDATED, POST_ATTRIBUTES -> {
entityBody.put("kv", dataJson);
entityBody.put(SCOPE, getScope(metadata));
if (EdgeEventActionType.POST_ATTRIBUTES.equals(actionType)) {
entityBody.put("isPostAttributes", true);
}
break;
case ATTRIBUTES_DELETED:
}
case ATTRIBUTES_DELETED -> {
List<String> keys = JacksonUtil.convertValue(dataJson.get("attributes"), new TypeReference<>() {
});
entityBody.put("keys", keys);
entityBody.put(SCOPE, getScope(metadata));
break;
case TIMESERIES_UPDATED:
}
case TIMESERIES_UPDATED -> {
entityBody.put("data", dataJson);
entityBody.put("ts", msg.getMetaDataTs());
break;
}
}
return buildEvent(ctx.getTenantId(),
actionType,
@ -179,4 +178,5 @@ public abstract class AbstractTbMsgPushNode<T extends BaseTbMsgPushNodeConfigura
return msg.isTypeOneOf(POST_TELEMETRY_REQUEST, POST_ATTRIBUTES_REQUEST, ATTRIBUTES_UPDATED, ATTRIBUTES_DELETED, TIMESERIES_UPDATED,
ALARM, CONNECT_EVENT, DISCONNECT_EVENT, ACTIVITY_EVENT, INACTIVITY_EVENT, TO_SERVER_RPC_REQUEST);
}
}

1
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/BaseTbMsgPushNodeConfiguration.java
View File

@ -30,4 +30,5 @@ public class BaseTbMsgPushNodeConfiguration implements NodeConfiguration<BaseTbM
configuration.setScope(AttributeScope.SERVER_SCOPE.name());
return configuration;
}
}

5
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToCloudNodeConfiguration.java
View File

@ -17,7 +17,7 @@ package org.thingsboard.rule.engine.edge;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.AttributeScope;
@EqualsAndHashCode(callSuper = true)
@Data
@ -26,7 +26,8 @@ public class TbMsgPushToCloudNodeConfiguration extends BaseTbMsgPushNodeConfigur
@Override
public TbMsgPushToCloudNodeConfiguration defaultConfiguration() {
TbMsgPushToCloudNodeConfiguration configuration = new TbMsgPushToCloudNodeConfiguration();
configuration.setScope(DataConstants.SERVER_SCOPE);
configuration.setScope(AttributeScope.SERVER_SCOPE.name());
return configuration;
}
}

1
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java
View File

@ -155,4 +155,5 @@ public class TbMsgPushToEdgeNode extends AbstractTbMsgPushNode<TbMsgPushToEdgeNo
edgeEvent.setEdgeId(edgeId);
return ctx.getEdgeEventService().saveAsync(edgeEvent);
}
}

5
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNodeConfiguration.java
View File

@ -17,7 +17,7 @@ package org.thingsboard.rule.engine.edge;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.AttributeScope;
@EqualsAndHashCode(callSuper = true)
@Data
@ -26,7 +26,8 @@ public class TbMsgPushToEdgeNodeConfiguration extends BaseTbMsgPushNodeConfigura
@Override
public TbMsgPushToEdgeNodeConfiguration defaultConfiguration() {
TbMsgPushToEdgeNodeConfiguration configuration = new TbMsgPushToEdgeNodeConfiguration();
configuration.setScope(DataConstants.SERVER_SCOPE);
configuration.setScope(AttributeScope.SERVER_SCOPE.name());
return configuration;
}
}