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

Change wording from persistence to processing on backend

pull/12621/head
Dmytro Skarzhynets 2 years ago
parent
8472e4f74e
commit
12a8c070a6
No known key found for this signature in database GPG Key ID: 2B51652F224037DF
17 changed files with 160 additions and 160 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
  2. 2
      application/src/main/data/json/tenant/device_profile/rule_chain_template.json
  3. 2
      application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
  4. 4
      application/src/main/data/upgrade/basic/schema_update.sql
  5. 6
      monitoring/src/main/resources/root_rule_chain.json
  6. 2
      msa/black-box-tests/src/test/resources/MqttRuleNodeTestMetadata.json
  7. 52
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
  8. 28
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java
  9. 6
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicateProcessingStrategy.java
  10. 10
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessageProcessingStrategy.java
  11. 22
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/ProcessingStrategy.java
  12. 10
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/SkipProcessingStrategy.java
  13. 72
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java
  14. 76
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicateProcessingStrategyTest.java
  15. 6
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessageProcessingStrategyTest.java
  16. 14
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/ProcessingStrategyTest.java
  17. 6
      rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/SkipProcessingStrategyTest.java

2
application/src/main/data/json/edge/rule_chains/edge_root_rule_chain.json
View File

@ -37,7 +37,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}, },

2
application/src/main/data/json/tenant/device_profile/rule_chain_template.json
View File

@ -23,7 +23,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
} }

2
application/src/main/data/json/tenant/rule_chains/root_rule_chain.json
View File

@ -22,7 +22,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
} }

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

@ -225,7 +225,7 @@ DO $$
SET configuration = ( SET configuration = (
(configuration::jsonb - 'skipLatestPersistence') (configuration::jsonb - 'skipLatestPersistence')
|| jsonb_build_object( || jsonb_build_object(
'persistenceSettings', jsonb_build_object( 'processingSettings', jsonb_build_object(
'type', 'ADVANCED', 'type', 'ADVANCED',
'timeseries', jsonb_build_object('type', 'ON_EVERY_MESSAGE'), 'timeseries', jsonb_build_object('type', 'ON_EVERY_MESSAGE'),
'latest', jsonb_build_object('type', 'SKIP'), 'latest', jsonb_build_object('type', 'SKIP'),
@ -242,7 +242,7 @@ DO $$
SET configuration = ( SET configuration = (
(configuration::jsonb - 'skipLatestPersistence') (configuration::jsonb - 'skipLatestPersistence')
|| jsonb_build_object( || jsonb_build_object(
'persistenceSettings', jsonb_build_object( 'processingSettings', jsonb_build_object(
'type', 'ON_EVERY_MESSAGE' 'type', 'ON_EVERY_MESSAGE'
) )
) )

6
monitoring/src/main/resources/root_rule_chain.json
View File

@ -25,7 +25,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}, },
@ -281,7 +281,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}, },
@ -319,7 +319,7 @@
"configuration": { "configuration": {
"defaultTTL": 180, "defaultTTL": 180,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}, },

2
msa/black-box-tests/src/test/resources/MqttRuleNodeTestMetadata.json
View File

@ -40,7 +40,7 @@
"configuration": { "configuration": {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}, },

52
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNode.java
View File

@ -27,7 +27,7 @@ import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.rule.engine.api.util.TbNodeUtils; import org.thingsboard.rule.engine.api.util.TbNodeUtils;
import org.thingsboard.rule.engine.telemetry.strategy.PersistenceStrategy; import org.thingsboard.rule.engine.telemetry.strategy.ProcessingStrategy;
import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.StringUtils;
import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.TenantProfile;
@ -45,11 +45,11 @@ import java.util.Map;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Deduplicate; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Deduplicate;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.OnEveryMessage; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.WebSocketsOnly; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.WebSocketsOnly;
import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_REQUEST; import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_REQUEST;
@Slf4j @Slf4j
@ -58,7 +58,7 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_RE
name = "save time series", name = "save time series",
configClazz = TbMsgTimeseriesNodeConfiguration.class, configClazz = TbMsgTimeseriesNodeConfiguration.class,
nodeDescription = """ nodeDescription = """
Saves time series data with a configurable TTL and according to configured persistence strategies. Saves time series data with a configurable TTL and according to configured processing strategies.
""", """,
nodeDetails = """ nodeDetails = """
Node performs three <strong>actions:</strong> Node performs three <strong>actions:</strong>
@ -68,14 +68,14 @@ import static org.thingsboard.server.common.data.msg.TbMsgType.POST_TELEMETRY_RE
<li><strong>WebSockets:</strong> notify WebSockets subscriptions about time series data updates.</li> <li><strong>WebSockets:</strong> notify WebSockets subscriptions about time series data updates.</li>
</ul> </ul>
For each <em>action</em>, three <strong>persistence strategies</strong> are available: For each <em>action</em>, three <strong>processing strategies</strong> are available:
<ul> <ul>
<li><strong>On every message:</strong> perform the action for every message.</li> <li><strong>On every message:</strong> perform the action for every message.</li>
<li><strong>Deduplicate:</strong> perform the action only for the first message from a particular originator within a configurable interval.</li> <li><strong>Deduplicate:</strong> perform the action only for the first message from a particular originator within a configurable interval.</li>
<li><strong>Skip:</strong> never perform the action.</li> <li><strong>Skip:</strong> never perform the action.</li>
</ul> </ul>
<strong>Persistence strategies</strong> are configured using <em>persistence settings</em>, which support two modes: <strong>Processing strategies</strong> are configured using <em>processing settings</em>, which support two modes:
<ul> <ul>
<li><strong>Basic</strong> <li><strong>Basic</strong>
<ul> <ul>
@ -110,7 +110,7 @@ public class TbMsgTimeseriesNode implements TbNode {
private TbContext ctx; private TbContext ctx;
private long tenantProfileDefaultStorageTtl; private long tenantProfileDefaultStorageTtl;
private PersistenceSettings persistenceSettings; private ProcessingSettings processingSettings;
@Override @Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
@ -118,7 +118,7 @@ public class TbMsgTimeseriesNode implements TbNode {
this.ctx = ctx; this.ctx = ctx;
ctx.addTenantProfileListener(this::onTenantProfileUpdate); ctx.addTenantProfileListener(this::onTenantProfileUpdate);
onTenantProfileUpdate(ctx.getTenantProfile()); onTenantProfileUpdate(ctx.getTenantProfile());
persistenceSettings = config.getPersistenceSettings(); processingSettings = config.getProcessingSettings();
} }
private void onTenantProfileUpdate(TenantProfile tenantProfile) { private void onTenantProfileUpdate(TenantProfile tenantProfile) {
@ -175,25 +175,25 @@ public class TbMsgTimeseriesNode implements TbNode {
} }
private TimeseriesSaveRequest.Strategy determineSaveStrategy(long ts, UUID originatorUuid) { private TimeseriesSaveRequest.Strategy determineSaveStrategy(long ts, UUID originatorUuid) {
if (persistenceSettings instanceof OnEveryMessage) { if (processingSettings instanceof OnEveryMessage) {
return TimeseriesSaveRequest.Strategy.SAVE_ALL; return TimeseriesSaveRequest.Strategy.SAVE_ALL;
} }
if (persistenceSettings instanceof WebSocketsOnly) { if (processingSettings instanceof WebSocketsOnly) {
return TimeseriesSaveRequest.Strategy.WS_ONLY; return TimeseriesSaveRequest.Strategy.WS_ONLY;
} }
if (persistenceSettings instanceof Deduplicate deduplicate) { if (processingSettings instanceof Deduplicate deduplicate) {
boolean isFirstMsgInInterval = deduplicate.getPersistenceStrategy().shouldPersist(ts, originatorUuid); boolean isFirstMsgInInterval = deduplicate.getProcessingStrategy().shouldProcess(ts, originatorUuid);
return isFirstMsgInInterval ? TimeseriesSaveRequest.Strategy.SAVE_ALL : TimeseriesSaveRequest.Strategy.SKIP_ALL; return isFirstMsgInInterval ? TimeseriesSaveRequest.Strategy.SAVE_ALL : TimeseriesSaveRequest.Strategy.SKIP_ALL;
} }
if (persistenceSettings instanceof Advanced advanced) { if (processingSettings instanceof Advanced advanced) {
return new TimeseriesSaveRequest.Strategy( return new TimeseriesSaveRequest.Strategy(
advanced.timeseries().shouldPersist(ts, originatorUuid), advanced.timeseries().shouldProcess(ts, originatorUuid),
advanced.latest().shouldPersist(ts, originatorUuid), advanced.latest().shouldProcess(ts, originatorUuid),
advanced.webSockets().shouldPersist(ts, originatorUuid) advanced.webSockets().shouldProcess(ts, originatorUuid)
); );
} }
// should not happen // should not happen
throw new IllegalArgumentException("Unknown persistence settings type: " + persistenceSettings.getClass().getSimpleName()); throw new IllegalArgumentException("Unknown processing settings type: " + processingSettings.getClass().getSimpleName());
} }
@Override @Override
@ -209,14 +209,14 @@ public class TbMsgTimeseriesNode implements TbNode {
hasChanges = true; hasChanges = true;
JsonNode skipLatestPersistence = oldConfiguration.get("skipLatestPersistence"); JsonNode skipLatestPersistence = oldConfiguration.get("skipLatestPersistence");
if (skipLatestPersistence != null && "true".equals(skipLatestPersistence.asText())) { if (skipLatestPersistence != null && "true".equals(skipLatestPersistence.asText())) {
var skipLatestPersistenceSettings = new Advanced( var skipLatestProcessingSettings = new Advanced(
PersistenceStrategy.onEveryMessage(), ProcessingStrategy.onEveryMessage(),
PersistenceStrategy.skip(), ProcessingStrategy.skip(),
PersistenceStrategy.onEveryMessage() ProcessingStrategy.onEveryMessage()
); );
((ObjectNode) oldConfiguration).set("persistenceSettings", JacksonUtil.valueToTree(skipLatestPersistenceSettings)); ((ObjectNode) oldConfiguration).set("processingSettings", JacksonUtil.valueToTree(skipLatestProcessingSettings));
} else { } else {
((ObjectNode) oldConfiguration).set("persistenceSettings", JacksonUtil.valueToTree(new OnEveryMessage())); ((ObjectNode) oldConfiguration).set("processingSettings", JacksonUtil.valueToTree(new OnEveryMessage()));
} }
((ObjectNode) oldConfiguration).remove("skipLatestPersistence"); ((ObjectNode) oldConfiguration).remove("skipLatestPersistence");
break; break;

28
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeConfiguration.java
View File

@ -24,14 +24,14 @@ import jakarta.validation.constraints.NotNull;
import lombok.Data; import lombok.Data;
import lombok.Getter; import lombok.Getter;
import org.thingsboard.rule.engine.api.NodeConfiguration; import org.thingsboard.rule.engine.api.NodeConfiguration;
import org.thingsboard.rule.engine.telemetry.strategy.PersistenceStrategy; import org.thingsboard.rule.engine.telemetry.strategy.ProcessingStrategy;
import java.util.Objects; import java.util.Objects;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Deduplicate; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Deduplicate;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.OnEveryMessage; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage;
import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.PersistenceSettings.WebSocketsOnly; import static org.thingsboard.rule.engine.telemetry.TbMsgTimeseriesNodeConfiguration.ProcessingSettings.WebSocketsOnly;
@Data @Data
public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration<TbMsgTimeseriesNodeConfiguration> { public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration<TbMsgTimeseriesNodeConfiguration> {
@ -39,14 +39,14 @@ public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration<TbMsg
private long defaultTTL; private long defaultTTL;
private boolean useServerTs; private boolean useServerTs;
@NotNull @NotNull
private PersistenceSettings persistenceSettings; private TbMsgTimeseriesNodeConfiguration.ProcessingSettings processingSettings;
@Override @Override
public TbMsgTimeseriesNodeConfiguration defaultConfiguration() { public TbMsgTimeseriesNodeConfiguration defaultConfiguration() {
TbMsgTimeseriesNodeConfiguration configuration = new TbMsgTimeseriesNodeConfiguration(); TbMsgTimeseriesNodeConfiguration configuration = new TbMsgTimeseriesNodeConfiguration();
configuration.setDefaultTTL(0L); configuration.setDefaultTTL(0L);
configuration.setUseServerTs(false); configuration.setUseServerTs(false);
configuration.setPersistenceSettings(new OnEveryMessage()); configuration.setProcessingSettings(new OnEveryMessage());
return configuration; return configuration;
} }
@ -61,29 +61,29 @@ public class TbMsgTimeseriesNodeConfiguration implements NodeConfiguration<TbMsg
@JsonSubTypes.Type(value = Deduplicate.class, name = "DEDUPLICATE"), @JsonSubTypes.Type(value = Deduplicate.class, name = "DEDUPLICATE"),
@JsonSubTypes.Type(value = Advanced.class, name = "ADVANCED") @JsonSubTypes.Type(value = Advanced.class, name = "ADVANCED")
}) })
sealed interface PersistenceSettings permits OnEveryMessage, Deduplicate, WebSocketsOnly, Advanced { sealed interface ProcessingSettings permits OnEveryMessage, Deduplicate, WebSocketsOnly, Advanced {
record OnEveryMessage() implements PersistenceSettings {} record OnEveryMessage() implements ProcessingSettings {}
record WebSocketsOnly() implements PersistenceSettings {} record WebSocketsOnly() implements ProcessingSettings {}
@Getter @Getter
final class Deduplicate implements PersistenceSettings { final class Deduplicate implements ProcessingSettings {
private final int deduplicationIntervalSecs; private final int deduplicationIntervalSecs;
@JsonIgnore @JsonIgnore
private final PersistenceStrategy persistenceStrategy; private final ProcessingStrategy processingStrategy;
@JsonCreator @JsonCreator
Deduplicate(@JsonProperty("deduplicationIntervalSecs") int deduplicationIntervalSecs) { Deduplicate(@JsonProperty("deduplicationIntervalSecs") int deduplicationIntervalSecs) {
this.deduplicationIntervalSecs = deduplicationIntervalSecs; this.deduplicationIntervalSecs = deduplicationIntervalSecs;
persistenceStrategy = PersistenceStrategy.deduplicate(deduplicationIntervalSecs); processingStrategy = ProcessingStrategy.deduplicate(deduplicationIntervalSecs);
} }
} }
record Advanced(PersistenceStrategy timeseries, PersistenceStrategy latest, PersistenceStrategy webSockets) implements PersistenceSettings { record Advanced(ProcessingStrategy timeseries, ProcessingStrategy latest, ProcessingStrategy webSockets) implements ProcessingSettings {
public Advanced { public Advanced {
Objects.requireNonNull(timeseries); Objects.requireNonNull(timeseries);

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicatePersistenceStrategy.java → rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicateProcessingStrategy.java
View File

@ -26,7 +26,7 @@ import java.time.Duration;
import java.util.Set; import java.util.Set;
import java.util.UUID; import java.util.UUID;
final class DeduplicatePersistenceStrategy implements PersistenceStrategy { final class DeduplicateProcessingStrategy implements ProcessingStrategy {
private static final int MIN_DEDUPLICATION_INTERVAL_SECS = 1; private static final int MIN_DEDUPLICATION_INTERVAL_SECS = 1;
private static final int MAX_DEDUPLICATION_INTERVAL_SECS = (int) Duration.ofDays(1L).toSeconds(); private static final int MAX_DEDUPLICATION_INTERVAL_SECS = (int) Duration.ofDays(1L).toSeconds();
@ -42,7 +42,7 @@ final class DeduplicatePersistenceStrategy implements PersistenceStrategy {
private final LoadingCache<Long, Set<UUID>> deduplicationCache; private final LoadingCache<Long, Set<UUID>> deduplicationCache;
@JsonCreator @JsonCreator
public DeduplicatePersistenceStrategy(@JsonProperty("deduplicationIntervalSecs") int deduplicationIntervalSecs) { public DeduplicateProcessingStrategy(@JsonProperty("deduplicationIntervalSecs") int deduplicationIntervalSecs) {
if (deduplicationIntervalSecs < MIN_DEDUPLICATION_INTERVAL_SECS || deduplicationIntervalSecs > MAX_DEDUPLICATION_INTERVAL_SECS) { if (deduplicationIntervalSecs < MIN_DEDUPLICATION_INTERVAL_SECS || deduplicationIntervalSecs > MAX_DEDUPLICATION_INTERVAL_SECS) {
throw new IllegalArgumentException("Deduplication interval must be at least " + MIN_DEDUPLICATION_INTERVAL_SECS + " second(s) " + throw new IllegalArgumentException("Deduplication interval must be at least " + MIN_DEDUPLICATION_INTERVAL_SECS + " second(s) " +
"and at most " + MAX_DEDUPLICATION_INTERVAL_SECS + " second(s), was " + deduplicationIntervalSecs + " second(s)"); "and at most " + MAX_DEDUPLICATION_INTERVAL_SECS + " second(s), was " + deduplicationIntervalSecs + " second(s)");
@ -81,7 +81,7 @@ final class DeduplicatePersistenceStrategy implements PersistenceStrategy {
} }
@Override @Override
public boolean shouldPersist(long ts, UUID originatorUuid) { public boolean shouldProcess(long ts, UUID originatorUuid) {
long intervalNumber = ts / deduplicationIntervalMillis; long intervalNumber = ts / deduplicationIntervalMillis;
return deduplicationCache.get(intervalNumber).add(originatorUuid); return deduplicationCache.get(intervalNumber).add(originatorUuid);
} }

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessagePersistenceStrategy.java → rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessageProcessingStrategy.java
View File

@ -19,19 +19,19 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import java.util.UUID; import java.util.UUID;
final class OnEveryMessagePersistenceStrategy implements PersistenceStrategy { final class OnEveryMessageProcessingStrategy implements ProcessingStrategy {
private static final OnEveryMessagePersistenceStrategy INSTANCE = new OnEveryMessagePersistenceStrategy(); private static final OnEveryMessageProcessingStrategy INSTANCE = new OnEveryMessageProcessingStrategy();
private OnEveryMessagePersistenceStrategy() {} private OnEveryMessageProcessingStrategy() {}
@JsonCreator @JsonCreator
public static OnEveryMessagePersistenceStrategy getInstance() { public static OnEveryMessageProcessingStrategy getInstance() {
return INSTANCE; return INSTANCE;
} }
@Override @Override
public boolean shouldPersist(long ts, UUID originatorUuid) { public boolean shouldProcess(long ts, UUID originatorUuid) {
return true; return true;
} }

22
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/PersistenceStrategy.java → rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/ProcessingStrategy.java
View File

@ -26,24 +26,24 @@ import java.util.UUID;
property = "type" property = "type"
) )
@JsonSubTypes({ @JsonSubTypes({
@JsonSubTypes.Type(value = OnEveryMessagePersistenceStrategy.class, name = "ON_EVERY_MESSAGE"), @JsonSubTypes.Type(value = OnEveryMessageProcessingStrategy.class, name = "ON_EVERY_MESSAGE"),
@JsonSubTypes.Type(value = DeduplicatePersistenceStrategy.class, name = "DEDUPLICATE"), @JsonSubTypes.Type(value = DeduplicateProcessingStrategy.class, name = "DEDUPLICATE"),
@JsonSubTypes.Type(value = SkipPersistenceStrategy.class, name = "SKIP") @JsonSubTypes.Type(value = SkipProcessingStrategy.class, name = "SKIP")
}) })
public sealed interface PersistenceStrategy permits OnEveryMessagePersistenceStrategy, DeduplicatePersistenceStrategy, SkipPersistenceStrategy { public sealed interface ProcessingStrategy permits OnEveryMessageProcessingStrategy, DeduplicateProcessingStrategy, SkipProcessingStrategy {
static PersistenceStrategy onEveryMessage() { static ProcessingStrategy onEveryMessage() {
return OnEveryMessagePersistenceStrategy.getInstance(); return OnEveryMessageProcessingStrategy.getInstance();
} }
static PersistenceStrategy deduplicate(int deduplicationIntervalSecs) { static ProcessingStrategy deduplicate(int deduplicationIntervalSecs) {
return new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); return new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
} }
static PersistenceStrategy skip() { static ProcessingStrategy skip() {
return SkipPersistenceStrategy.getInstance(); return SkipProcessingStrategy.getInstance();
} }
boolean shouldPersist(long ts, UUID originatorUuid); boolean shouldProcess(long ts, UUID originatorUuid);
} }

10
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/SkipPersistenceStrategy.java → rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/telemetry/strategy/SkipProcessingStrategy.java
View File

@ -19,19 +19,19 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import java.util.UUID; import java.util.UUID;
final class SkipPersistenceStrategy implements PersistenceStrategy { final class SkipProcessingStrategy implements ProcessingStrategy {
private static final SkipPersistenceStrategy INSTANCE = new SkipPersistenceStrategy(); private static final SkipProcessingStrategy INSTANCE = new SkipProcessingStrategy();
private SkipPersistenceStrategy() {} private SkipProcessingStrategy() {}
@JsonCreator @JsonCreator
public static SkipPersistenceStrategy getInstance() { public static SkipProcessingStrategy getInstance() {
return INSTANCE; return INSTANCE;
} }
@Override @Override
public boolean shouldPersist(long ts, UUID originatorUuid) { public boolean shouldProcess(long ts, UUID originatorUuid) {
return false; return false;
} }

72
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/TbMsgTimeseriesNodeTest.java
View File

@ -34,7 +34,7 @@ import org.thingsboard.rule.engine.api.TbNode;
import org.thingsboard.rule.engine.api.TbNodeConfiguration; import org.thingsboard.rule.engine.api.TbNodeConfiguration;
import org.thingsboard.rule.engine.api.TbNodeException; import org.thingsboard.rule.engine.api.TbNodeException;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest;
import org.thingsboard.rule.engine.telemetry.strategy.PersistenceStrategy; import org.thingsboard.rule.engine.telemetry.strategy.ProcessingStrategy;
import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.adaptor.JsonConverter;
import org.thingsboard.server.common.data.TenantProfile; import org.thingsboard.server.common.data.TenantProfile;
import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.DeviceId;
@ -110,7 +110,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
@Test @Test
public void verifyDefaultConfig() { public void verifyDefaultConfig() {
assertThat(config.getDefaultTTL()).isEqualTo(0L); assertThat(config.getDefaultTTL()).isEqualTo(0L);
assertThat(config.getPersistenceSettings()).isInstanceOf(TbMsgTimeseriesNodeConfiguration.PersistenceSettings.OnEveryMessage.class); assertThat(config.getProcessingSettings()).isInstanceOf(TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage.class);
assertThat(config.isUseServerTs()).isFalse(); assertThat(config.isUseServerTs()).isFalse();
} }
@ -124,14 +124,14 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenPersistenceSettingsAreNull_whenValidatingConstraints_thenThrowsException() { public void givenProcessingSettingsAreNull_whenValidatingConstraints_thenThrowsException() {
// GIVEN // GIVEN
config.setPersistenceSettings(null); config.setProcessingSettings(null);
// WHEN-THEN // WHEN-THEN
assertThatThrownBy(() -> ConstraintValidator.validateFields(config)) assertThatThrownBy(() -> ConstraintValidator.validateFields(config))
.isInstanceOf(DataValidationException.class) .isInstanceOf(DataValidationException.class)
.hasMessage("Validation error: persistenceSettings must not be null"); .hasMessage("Validation error: processingSettings must not be null");
} }
@ParameterizedTest @ParameterizedTest
@ -216,15 +216,15 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenSkipLatestPersistenceSettingsAndTtlFromConfig_whenOnMsg_thenSaveTimeseriesUsingTtlFromConfig() throws TbNodeException { public void givenSkipLatestProcessingSettingsAndTtlFromConfig_whenOnMsg_thenSaveTimeseriesUsingTtlFromConfig() throws TbNodeException {
// GIVEN // GIVEN
config.setDefaultTTL(10L); config.setDefaultTTL(10L);
var timeseriesStrategy = PersistenceStrategy.onEveryMessage(); var timeseriesStrategy = ProcessingStrategy.onEveryMessage();
var latestStrategy = PersistenceStrategy.skip(); var latestStrategy = ProcessingStrategy.skip();
var webSockets = PersistenceStrategy.onEveryMessage(); var webSockets = ProcessingStrategy.onEveryMessage();
var persistenceSettings = new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced(timeseriesStrategy, latestStrategy, webSockets); var processingSettings = new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(timeseriesStrategy, latestStrategy, webSockets);
config.setPersistenceSettings(persistenceSettings); config.setProcessingSettings(processingSettings);
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -333,9 +333,9 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenOnEveryMessagePersistenceSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { public void givenOnEveryMessageProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.OnEveryMessage()); config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.OnEveryMessage());
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -368,9 +368,9 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenDeduplicatePersistenceSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistThisMessageOnlyFirstTime() throws TbNodeException { public void givenDeduplicateProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenPersistThisMessageOnlyFirstTime() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Deduplicate(10)); config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Deduplicate(10));
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -403,9 +403,9 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenWebsocketsOnlyPersistenceSettingsAndSameMessageTwoTimes_whenOnMsg_thenSendsOnlyWsUpdateTwoTimes() throws TbNodeException { public void givenWebSocketsOnlyProcessingSettingsAndSameMessageTwoTimes_whenOnMsg_thenSendsOnlyWsUpdateTwoTimes() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.WebSocketsOnly()); config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.WebSocketsOnly());
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -438,12 +438,12 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenAdvancedPersistenceSettingsWithOnEveryMessageStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException { public void givenAdvancedProcessingSettingsWithOnEveryMessageStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenPersistSameMessageTwoTimes() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced( config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(
PersistenceStrategy.onEveryMessage(), ProcessingStrategy.onEveryMessage(),
PersistenceStrategy.onEveryMessage(), ProcessingStrategy.onEveryMessage(),
PersistenceStrategy.onEveryMessage() ProcessingStrategy.onEveryMessage()
)); ));
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -477,12 +477,12 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenAdvancedPersistenceSettingsWithDifferentDeduplicateStrategyForEachAction_whenOnMsg_thenEvaluatesStrategiesForEachActionsIndependently() throws TbNodeException { public void givenAdvancedProcessingSettingsWithDifferentDeduplicateStrategyForEachAction_whenOnMsg_thenEvaluatesStrategiesForEachActionsIndependently() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced( config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(
PersistenceStrategy.deduplicate(1), ProcessingStrategy.deduplicate(1),
PersistenceStrategy.deduplicate(2), ProcessingStrategy.deduplicate(2),
PersistenceStrategy.deduplicate(3) ProcessingStrategy.deduplicate(3)
)); ));
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -528,12 +528,12 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
} }
@Test @Test
public void givenAdvancedPersistenceSettingsWithSkipStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenSkipsSameMessageTwoTimes() throws TbNodeException { public void givenAdvancedProcessingSettingsWithSkipStrategiesForAllActionsAndSameMessageTwoTimes_whenOnMsg_thenSkipsSameMessageTwoTimes() throws TbNodeException {
// GIVEN // GIVEN
config.setPersistenceSettings(new TbMsgTimeseriesNodeConfiguration.PersistenceSettings.Advanced( config.setProcessingSettings(new TbMsgTimeseriesNodeConfiguration.ProcessingSettings.Advanced(
PersistenceStrategy.skip(), ProcessingStrategy.skip(),
PersistenceStrategy.skip(), ProcessingStrategy.skip(),
PersistenceStrategy.skip() ProcessingStrategy.skip()
)); ));
node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config))); node.init(ctxMock, new TbNodeConfiguration(JacksonUtil.valueToTree(config)));
@ -577,7 +577,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
{ {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}"""), }"""),
@ -591,7 +591,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
{ {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}"""), }"""),
@ -606,7 +606,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
{ {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"
} }
}"""), }"""),
@ -621,7 +621,7 @@ public class TbMsgTimeseriesNodeTest extends AbstractRuleNodeUpgradeTest {
{ {
"defaultTTL": 0, "defaultTTL": 0,
"useServerTs": false, "useServerTs": false,
"persistenceSettings": { "processingSettings": {
"type": "ADVANCED", "type": "ADVANCED",
"timeseries": { "timeseries": {
"type": "ON_EVERY_MESSAGE" "type": "ON_EVERY_MESSAGE"

76
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicatePersistenceStrategyTest.java → rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/DeduplicateProcessingStrategyTest.java
View File

@ -28,27 +28,27 @@ import java.util.UUID;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy; import static org.assertj.core.api.Assertions.assertThatThrownBy;
class DeduplicatePersistenceStrategyTest { class DeduplicateProcessingStrategyTest {
final int deduplicationIntervalSecs = 10; final int deduplicationIntervalSecs = 10;
DeduplicatePersistenceStrategy strategy; DeduplicateProcessingStrategy strategy;
@BeforeEach @BeforeEach
void setup() { void setup() {
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
} }
@Test @Test
void shouldThrowWhenDeduplicationIntervalIsLessThanOneSecond() { void shouldThrowWhenDeduplicationIntervalIsLessThanOneSecond() {
assertThatThrownBy(() -> new DeduplicatePersistenceStrategy(0)) assertThatThrownBy(() -> new DeduplicateProcessingStrategy(0))
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 0 second(s)"); .hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 0 second(s)");
} }
@Test @Test
void shouldThrowWhenDeduplicationIntervalIsMoreThan24Hours() { void shouldThrowWhenDeduplicationIntervalIsMoreThan24Hours() {
assertThatThrownBy(() -> new DeduplicatePersistenceStrategy(86401)) assertThatThrownBy(() -> new DeduplicateProcessingStrategy(86401))
.isInstanceOf(IllegalArgumentException.class) .isInstanceOf(IllegalArgumentException.class)
.hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 86401 second(s)"); .hasMessageContaining("Deduplication interval must be at least 1 second(s) and at most 86400 second(s), was 86401 second(s)");
} }
@ -59,7 +59,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = 1; // min deduplication interval duration int deduplicationIntervalSecs = 1; // min deduplication interval duration
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -76,7 +76,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds(); // max deduplication interval duration int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds(); // max deduplication interval duration
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -93,7 +93,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -110,7 +110,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = 1; // min deduplication interval duration int deduplicationIntervalSecs = 1; // min deduplication interval duration
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -127,7 +127,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds(); int deduplicationIntervalSecs = (int) Duration.ofHours(1L).toSeconds();
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -144,7 +144,7 @@ class DeduplicatePersistenceStrategyTest {
int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration int deduplicationIntervalSecs = (int) Duration.ofDays(1L).toSeconds(); // max deduplication interval duration
// WHEN // WHEN
strategy = new DeduplicatePersistenceStrategy(deduplicationIntervalSecs); strategy = new DeduplicateProcessingStrategy(deduplicationIntervalSecs);
// THEN // THEN
var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache"); var deduplicationCache = (LoadingCache<Long, Set<UUID>>) ReflectionTestUtils.getField(strategy, "deduplicationCache");
@ -160,7 +160,7 @@ class DeduplicatePersistenceStrategyTest {
long ts = 1_000_000L; long ts = 1_000_000L;
UUID originator = UUID.randomUUID(); UUID originator = UUID.randomUUID();
assertThat(strategy.shouldPersist(ts, originator)).isTrue(); assertThat(strategy.shouldProcess(ts, originator)).isTrue();
} }
@Test @Test
@ -169,11 +169,11 @@ class DeduplicatePersistenceStrategyTest {
UUID originator = UUID.randomUUID(); UUID originator = UUID.randomUUID();
// Initial call should return true // Initial call should return true
assertThat(strategy.shouldPersist(baseTs, originator)).isTrue(); assertThat(strategy.shouldProcess(baseTs, originator)).isTrue();
// Subsequent call within the same interval should return false for the same originator // Subsequent call within the same interval should return false for the same originator
long withinSameIntervalTs = baseTs + 1000L; long withinSameIntervalTs = baseTs + 1000L;
assertThat(strategy.shouldPersist(withinSameIntervalTs, originator)).isFalse(); assertThat(strategy.shouldProcess(withinSameIntervalTs, originator)).isFalse();
} }
@Test @Test
@ -183,12 +183,12 @@ class DeduplicatePersistenceStrategyTest {
UUID originator2 = UUID.randomUUID(); UUID originator2 = UUID.randomUUID();
// First call for different originators in the same interval should return true independently // First call for different originators in the same interval should return true independently
assertThat(strategy.shouldPersist(baseTs, originator1)).isTrue(); assertThat(strategy.shouldProcess(baseTs, originator1)).isTrue();
assertThat(strategy.shouldPersist(baseTs, originator2)).isTrue(); assertThat(strategy.shouldProcess(baseTs, originator2)).isTrue();
// Subsequent calls for the same originators within the same interval should return false // Subsequent calls for the same originators within the same interval should return false
assertThat(strategy.shouldPersist(baseTs + 500L, originator1)).isFalse(); assertThat(strategy.shouldProcess(baseTs + 500L, originator1)).isFalse();
assertThat(strategy.shouldPersist(baseTs + 500L, originator2)).isFalse(); assertThat(strategy.shouldProcess(baseTs + 500L, originator2)).isFalse();
} }
@Test @Test
@ -197,11 +197,11 @@ class DeduplicatePersistenceStrategyTest {
long maxTs = Long.MAX_VALUE; long maxTs = Long.MAX_VALUE;
UUID originator = UUID.randomUUID(); UUID originator = UUID.randomUUID();
assertThat(strategy.shouldPersist(minTs, originator)).isTrue(); assertThat(strategy.shouldProcess(minTs, originator)).isTrue();
assertThat(strategy.shouldPersist(minTs + 1L, originator)).isFalse(); assertThat(strategy.shouldProcess(minTs + 1L, originator)).isFalse();
assertThat(strategy.shouldPersist(maxTs, originator)).isTrue(); assertThat(strategy.shouldProcess(maxTs, originator)).isTrue();
assertThat(strategy.shouldPersist(maxTs - 1L, originator)).isFalse(); assertThat(strategy.shouldProcess(maxTs - 1L, originator)).isFalse();
} }
@Test @Test
@ -213,22 +213,22 @@ class DeduplicatePersistenceStrategyTest {
long firstIntervalEnd = firstIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L; long firstIntervalEnd = firstIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L;
long firstIntervalMiddle = calculateMiddle(firstIntervalStart, firstIntervalEnd); long firstIntervalMiddle = calculateMiddle(firstIntervalStart, firstIntervalEnd);
assertThat(strategy.shouldPersist(firstIntervalStart, originator)).isTrue(); assertThat(strategy.shouldProcess(firstIntervalStart, originator)).isTrue();
assertThat(strategy.shouldPersist(firstIntervalStart + 1, originator)).isFalse(); assertThat(strategy.shouldProcess(firstIntervalStart + 1, originator)).isFalse();
assertThat(strategy.shouldPersist(firstIntervalMiddle, originator)).isFalse(); assertThat(strategy.shouldProcess(firstIntervalMiddle, originator)).isFalse();
assertThat(strategy.shouldPersist(firstIntervalEnd - 1, originator)).isFalse(); assertThat(strategy.shouldProcess(firstIntervalEnd - 1, originator)).isFalse();
assertThat(strategy.shouldPersist(firstIntervalEnd, originator)).isFalse(); assertThat(strategy.shouldProcess(firstIntervalEnd, originator)).isFalse();
// check 2nd interval // check 2nd interval
long secondIntervalStart = firstIntervalEnd + 1L; long secondIntervalStart = firstIntervalEnd + 1L;
long secondIntervalEnd = secondIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L; long secondIntervalEnd = secondIntervalStart + Duration.ofSeconds(deduplicationIntervalSecs).toMillis() - 1L;
long secondIntervalMiddle = calculateMiddle(secondIntervalStart, secondIntervalEnd); long secondIntervalMiddle = calculateMiddle(secondIntervalStart, secondIntervalEnd);
assertThat(strategy.shouldPersist(secondIntervalStart, originator)).isTrue(); assertThat(strategy.shouldProcess(secondIntervalStart, originator)).isTrue();
assertThat(strategy.shouldPersist(secondIntervalStart + 1, originator)).isFalse(); assertThat(strategy.shouldProcess(secondIntervalStart + 1, originator)).isFalse();
assertThat(strategy.shouldPersist(secondIntervalMiddle, originator)).isFalse(); assertThat(strategy.shouldProcess(secondIntervalMiddle, originator)).isFalse();
assertThat(strategy.shouldPersist(secondIntervalEnd - 1, originator)).isFalse(); assertThat(strategy.shouldProcess(secondIntervalEnd - 1, originator)).isFalse();
assertThat(strategy.shouldPersist(secondIntervalEnd, originator)).isFalse(); assertThat(strategy.shouldProcess(secondIntervalEnd, originator)).isFalse();
} }
@Test @Test
@ -238,19 +238,19 @@ class DeduplicatePersistenceStrategyTest {
long baseTs = 0L; long baseTs = 0L;
// First interval for both originators // First interval for both originators
assertThat(strategy.shouldPersist(baseTs, originator1)).isTrue(); assertThat(strategy.shouldProcess(baseTs, originator1)).isTrue();
assertThat(strategy.shouldPersist(baseTs, originator2)).isTrue(); assertThat(strategy.shouldProcess(baseTs, originator2)).isTrue();
// Move to the next interval // Move to the next interval
long nextIntervalTs = baseTs + Duration.ofSeconds(10).toMillis(); long nextIntervalTs = baseTs + Duration.ofSeconds(10).toMillis();
// Each originator should be allowed again in the new interval // Each originator should be allowed again in the new interval
assertThat(strategy.shouldPersist(nextIntervalTs, originator1)).isTrue(); assertThat(strategy.shouldProcess(nextIntervalTs, originator1)).isTrue();
assertThat(strategy.shouldPersist(nextIntervalTs, originator2)).isTrue(); assertThat(strategy.shouldProcess(nextIntervalTs, originator2)).isTrue();
// Subsequent calls in the same new interval should return false // Subsequent calls in the same new interval should return false
assertThat(strategy.shouldPersist(nextIntervalTs + 500L, originator1)).isFalse(); assertThat(strategy.shouldProcess(nextIntervalTs + 500L, originator1)).isFalse();
assertThat(strategy.shouldPersist(nextIntervalTs + 500L, originator2)).isFalse(); assertThat(strategy.shouldProcess(nextIntervalTs + 500L, originator2)).isFalse();
} }
private static long calculateMiddle(long start, long end) { private static long calculateMiddle(long start, long end) {

6
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessagePersistenceStrategyTest.java → rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/OnEveryMessageProcessingStrategyTest.java
View File

@ -24,13 +24,13 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
class OnEveryMessagePersistenceStrategyTest { class OnEveryMessageProcessingStrategyTest {
@ParameterizedTest @ParameterizedTest
@MethodSource("edgeCaseProvider") @MethodSource("edgeCaseProvider")
void shouldAlwaysReturnTrueForAnyInput(long timestamp, UUID originator) { void shouldAlwaysReturnTrueForAnyInput(long timestamp, UUID originator) {
var onEveryMessage = OnEveryMessagePersistenceStrategy.getInstance(); var onEveryMessage = OnEveryMessageProcessingStrategy.getInstance();
assertThat(onEveryMessage.shouldPersist(timestamp, originator)).isTrue(); assertThat(onEveryMessage.shouldProcess(timestamp, originator)).isTrue();
} }
private static Stream<Arguments> edgeCaseProvider() { private static Stream<Arguments> edgeCaseProvider() {

14
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/PersistenceStrategyTest.java → rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/ProcessingStrategyTest.java
View File

@ -22,23 +22,23 @@ import java.time.Duration;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
class PersistenceStrategyTest { class ProcessingStrategyTest {
@Test @Test
void testOnEveryMessageReturnsCorrectInstance() { void testOnEveryMessageReturnsCorrectInstance() {
PersistenceStrategy strategy = PersistenceStrategy.onEveryMessage(); ProcessingStrategy strategy = ProcessingStrategy.onEveryMessage();
assertThat(strategy) assertThat(strategy)
.isNotNull() .isNotNull()
.isInstanceOf(OnEveryMessagePersistenceStrategy.class); .isInstanceOf(OnEveryMessageProcessingStrategy.class);
} }
@Test @Test
void testDeduplicateReturnsCorrectInstance() { void testDeduplicateReturnsCorrectInstance() {
int validDeduplicationIntervalSecs = 5; int validDeduplicationIntervalSecs = 5;
PersistenceStrategy strategy = PersistenceStrategy.deduplicate(validDeduplicationIntervalSecs); ProcessingStrategy strategy = ProcessingStrategy.deduplicate(validDeduplicationIntervalSecs);
assertThat(strategy) assertThat(strategy)
.isNotNull() .isNotNull()
.isInstanceOf(DeduplicatePersistenceStrategy.class); .isInstanceOf(DeduplicateProcessingStrategy.class);
long actualDeduplicationIntervalMillis = (long) ReflectionTestUtils.getField(strategy, "deduplicationIntervalMillis"); long actualDeduplicationIntervalMillis = (long) ReflectionTestUtils.getField(strategy, "deduplicationIntervalMillis");
assertThat(actualDeduplicationIntervalMillis).isEqualTo(Duration.ofSeconds(validDeduplicationIntervalSecs).toMillis()); assertThat(actualDeduplicationIntervalMillis).isEqualTo(Duration.ofSeconds(validDeduplicationIntervalSecs).toMillis());
@ -46,10 +46,10 @@ class PersistenceStrategyTest {
@Test @Test
void testSkipReturnsCorrectInstance() { void testSkipReturnsCorrectInstance() {
PersistenceStrategy strategy = PersistenceStrategy.skip(); ProcessingStrategy strategy = ProcessingStrategy.skip();
assertThat(strategy) assertThat(strategy)
.isNotNull() .isNotNull()
.isInstanceOf(SkipPersistenceStrategy.class); .isInstanceOf(SkipProcessingStrategy.class);
} }
} }

6
rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/SkipPersistenceStrategyTest.java → rule-engine/rule-engine-components/src/test/java/org/thingsboard/rule/engine/telemetry/strategy/SkipProcessingStrategyTest.java
View File

@ -24,13 +24,13 @@ import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThat;
class SkipPersistenceStrategyTest { class SkipProcessingStrategyTest {
@ParameterizedTest @ParameterizedTest
@MethodSource("edgeCaseProvider") @MethodSource("edgeCaseProvider")
void shouldAlwaysReturnFalseForAnyInput(long timestamp, UUID originator) { void shouldAlwaysReturnFalseForAnyInput(long timestamp, UUID originator) {
var skipStrategy = SkipPersistenceStrategy.getInstance(); var skipStrategy = SkipProcessingStrategy.getInstance();
assertThat(skipStrategy.shouldPersist(timestamp, originator)).isFalse(); assertThat(skipStrategy.shouldProcess(timestamp, originator)).isFalse();
} }
private static Stream<Arguments> edgeCaseProvider() { private static Stream<Arguments> edgeCaseProvider() {
Write Preview
Loading…
Cancel
Save