tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

added upgrade script

pull/14225/head
IrynaMatveieva 12 months ago
parent
a8525c650c
commit
76606f63ba
6 changed files with 101 additions and 16 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 19
      application/src/main/data/upgrade/basic/schema_update.sql
  2. 85
      application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
  3. 2
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
  4. 2
      application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
  5. 4
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java
  6. 5
      common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java

19
application/src/main/data/upgrade/basic/schema_update.sql
View File

@ -69,3 +69,22 @@ ALTER TABLE calculated_field DROP CONSTRAINT IF EXISTS calculated_field_unq_key;
ALTER TABLE calculated_field ADD CONSTRAINT calculated_field_unq_key UNIQUE (entity_id, type, name);
-- CALCULATED FIELD UNIQUE CONSTRAINT UPDATE END
-- CALCULATED FIELD OUTPUT STRATEGY UPGRADE START
UPDATE calculated_field
SET configuration = jsonb_set(
configuration::jsonb,
'{output}',
(configuration::jsonb -> 'output')
|| jsonb_build_object(
'strategy',
jsonb_build_object(
'type', 'PUSH_TO_RULE_ENGINE'
)
),
false
)
WHERE (configuration::jsonb -> 'output' -> 'strategy') IS NULL;
-- CALCULATED FIELD OUTPUT STRATEGY UPGRADE END

85
application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java
View File

@ -27,12 +27,15 @@ import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.rule.engine.api.AttributesSaveRequest;
import org.thingsboard.rule.engine.api.AttributesSaveRequest.Strategy;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.cf.configuration.Argument;
import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
import org.thingsboard.server.common.data.cf.configuration.AttributeSkipRuleEngineOutputStrategy;
import org.thingsboard.server.common.data.cf.configuration.OutputType;
import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration;
import org.thingsboard.server.common.data.cf.configuration.TimeSeriesSkipRuleEngineOutputStrategy;
@ -71,6 +74,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION;
@ -344,8 +348,8 @@ public abstract class AbstractCalculatedFieldProcessingService {
SettableFuture<Void> future = SettableFuture.create();
switch (type) {
case ATTRIBUTES -> saveAttributes(tenantId, entityId, jsonResult, cfIds, future);
case TIME_SERIES -> saveTimeSeries(tenantId, entityId, jsonResult, ((TimeSeriesSkipRuleEngineOutputStrategy) cfResult.getOutputStrategy()).getTtl(), cfIds, System.currentTimeMillis(), TimeseriesSaveRequest.Strategy.PROCESS_ALL, future);
case ATTRIBUTES -> saveAttributes(tenantId, entityId, cfResult, cfIds, future);
case TIME_SERIES -> saveTimeSeries(tenantId, entityId, cfResult, cfIds, System.currentTimeMillis(), future);
}
if (log.isTraceEnabled()) {
@ -365,20 +369,54 @@ public abstract class AbstractCalculatedFieldProcessingService {
}
}
private void saveAttributes(TenantId tenantId, EntityId entityId, JsonElement jsonResult, List<CalculatedFieldId> cfIds, SettableFuture<Void> future) {
private void saveAttributes(TenantId tenantId, EntityId entityId, TelemetryCalculatedFieldResult cfResult, List<CalculatedFieldId> cfIds, SettableFuture<Void> future) {
if (!(cfResult.getOutputStrategy() instanceof AttributeSkipRuleEngineOutputStrategy outputStrategy)) {
return;
}
JsonElement jsonResult = JsonParser.parseString(Objects.requireNonNull(cfResult.stringValue()));
AttributesSaveRequest.Strategy strategy = new Strategy(outputStrategy.isSaveAttribute(), outputStrategy.isSendWsUpdate(), outputStrategy.isProcessCfs());
List<AttributeKvEntry> attributeKvEntries = JsonConverter.convertToAttributes(jsonResult);
tsSubService.saveAttributesInternal(AttributesSaveRequest.builder()
.tenantId(tenantId)
.entityId(entityId)
.entries(attributeKvEntries)
.strategy(AttributesSaveRequest.Strategy.PROCESS_ALL)
.previousCalculatedFieldIds(cfIds)
.future(future)
.build()
);
if (!outputStrategy.isUpdateAttributesOnlyOnValueChange()) {
tsSubService.saveAttributesInternal(AttributesSaveRequest.builder()
.tenantId(tenantId)
.entityId(entityId)
.entries(attributeKvEntries)
.strategy(strategy)
.previousCalculatedFieldIds(cfIds)
.future(future)
.build()
);
return;
}
List<String> keys = attributeKvEntries.stream().map(KvEntry::getKey).collect(Collectors.toList());
ListenableFuture<List<AttributeKvEntry>> findFuture = attributesService.find(tenantId, entityId, cfResult.getScope(), keys);
DonAsynchron.withCallback(findFuture,
existingAttributes -> {
List<AttributeKvEntry> attributesChanged = filterChangedAttr(existingAttributes, attributeKvEntries);
tsSubService.saveAttributesInternal(AttributesSaveRequest.builder()
.tenantId(tenantId)
.entityId(entityId)
.entries(attributesChanged)
.strategy(strategy)
.previousCalculatedFieldIds(cfIds)
.future(future)
.build()
);
},
future::setException,
MoreExecutors.directExecutor());
}
private void saveTimeSeries(TenantId tenantId, EntityId entityId, JsonElement jsonResult, Long ttl, List<CalculatedFieldId> cfIds, long ts, TimeseriesSaveRequest.Strategy strategy, SettableFuture<Void> future) {
private void saveTimeSeries(TenantId tenantId, EntityId entityId, TelemetryCalculatedFieldResult cfResult, List<CalculatedFieldId> cfIds, long ts, SettableFuture<Void> future) {
if (!(cfResult.getOutputStrategy() instanceof TimeSeriesSkipRuleEngineOutputStrategy outputStrategy)) {
return;
}
JsonElement jsonResult = JsonParser.parseString(Objects.requireNonNull(cfResult.stringValue()));
Map<Long, List<KvEntry>> tsKvMap = JsonConverter.convertToTelemetry(jsonResult, ts);
List<TsKvEntry> tsEntries = new ArrayList<>();
for (Map.Entry<Long, List<KvEntry>> tsKvEntry : tsKvMap.entrySet()) {
@ -386,15 +424,34 @@ public abstract class AbstractCalculatedFieldProcessingService {
tsEntries.add(new BasicTsKvEntry(tsKvEntry.getKey(), kvEntry));
}
}
TimeseriesSaveRequest.Strategy strategy = new TimeseriesSaveRequest.Strategy(outputStrategy.isSaveTimeSeries(), outputStrategy.isSaveLatest(), outputStrategy.isSendWsUpdate(), outputStrategy.isProcessCfs());
tsSubService.saveTimeseriesInternal(TimeseriesSaveRequest.builder()
.tenantId(tenantId)
.entityId(entityId)
.entries(tsEntries)
.ttl(ttl)
.ttl(outputStrategy.getTtl())
.strategy(strategy)
.previousCalculatedFieldIds(cfIds)
.future(future)
.build());
}
private List<AttributeKvEntry> filterChangedAttr(List<AttributeKvEntry> existingAttributes, List<AttributeKvEntry> newAttributes) {
if (existingAttributes == null || existingAttributes.isEmpty()) {
return newAttributes;
}
Map<String, AttributeKvEntry> currentAttrMap = existingAttributes.stream()
.collect(Collectors.toMap(AttributeKvEntry::getKey, Function.identity(), (existing, replacement) -> existing));
return newAttributes.stream()
.filter(item -> {
AttributeKvEntry cacheAttr = currentAttrMap.get(item.getKey());
return cacheAttr == null
|| !Objects.equals(item.getValue(), cacheAttr.getValue()) //JSON and String can be equals by value, but different by type
|| !Objects.equals(item.getDataType(), cacheAttr.getDataType());
})
.collect(Collectors.toList());
}
}

2
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java
View File

@ -17,7 +17,7 @@ package org.thingsboard.server.service.cf;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.logging.log4j.util.TriConsumer;
import org.apache.commons.lang3.function.TriConsumer;
import org.springframework.stereotype.Service;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
import org.thingsboard.server.actors.calculatedField.MultipleTbCallback;

2
application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java
View File

@ -1235,7 +1235,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes
output.setName("fahrenheitTemp");
output.setType(OutputType.TIME_SERIES);
output.setDecimalsByDefault(1);
output.setStrategy(new TimeSeriesSkipRuleEngineOutputStrategy(1000L));
output.setStrategy(new TimeSeriesSkipRuleEngineOutputStrategy(1000L, true, true, true, true));
config.setOutput(output);

4
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java
View File

@ -26,4 +26,8 @@ public class AttributeSkipRuleEngineOutputStrategy extends SkipRuleEngineOutputS
private boolean updateAttributesOnlyOnValueChange;
private boolean saveAttribute;
private boolean sendWsUpdate;
private boolean processCfs;
}

5
common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java
View File

@ -26,4 +26,9 @@ public class TimeSeriesSkipRuleEngineOutputStrategy extends SkipRuleEngineOutput
private long ttl;
private boolean saveTimeSeries;
private boolean saveLatest;
private boolean sendWsUpdate;
private boolean processCfs;
}

Write Preview
Loading…
Cancel
Save