Browse Source

Actor system implementation draft

pull/12586/head
Andrii Shvaika 2 years ago
parent
commit
3d42a4ca04
  1. 8
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java
  2. 6
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java
  3. 1
      application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java
  4. 10
      common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
  5. 2
      dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java

8
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@ -78,6 +78,10 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
this.ctx = ctx;
}
public void process(CalculatedFieldStateRestoreMsg msg) {
states.put(msg.getId().cfId(), msg.getState());
}
public void process(EntityCalculatedFieldTelemetryMsg msg) {
var proto = msg.getProto();
var numberOfCallbacks = CALLBACKS_PER_CF * (msg.getEntityIdFields().size() + msg.getProfileIdFields().size());
@ -192,8 +196,4 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
}
return cfIds;
}
public void process(CalculatedFieldStateRestoreMsg msg) {
states.put(msg.getId().cfId(), msg.getState());
}
}

6
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldExecutionService.java

@ -837,8 +837,10 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
telemetryMsg.setEntityIdMSB(entityId.getId().getMostSignificantBits());
telemetryMsg.setEntityIdLSB(entityId.getId().getLeastSignificantBits());
for (CalculatedFieldId cfId : calculatedFieldIds) {
telemetryMsg.addPreviousCalculatedFields(toProto(cfId));
if(calculatedFieldIds != null) {
for (CalculatedFieldId cfId : calculatedFieldIds) {
telemetryMsg.addPreviousCalculatedFields(toProto(cfId));
}
}
return telemetryMsg;

1
application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldState.java

@ -44,5 +44,6 @@ public interface CalculatedFieldState {
ListenableFuture<CalculatedFieldResult> performCalculation(CalculatedFieldCtx ctx);
@JsonIgnore
boolean isReady();
}

10
common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java

@ -644,11 +644,13 @@ public class ProtoUtils {
}
public static TransportProtos.TsKvProto toTsKvProto(TsKvEntry tsKvEntry) {
return TransportProtos.TsKvProto.newBuilder()
var builder = TransportProtos.TsKvProto.newBuilder()
.setTs(tsKvEntry.getTs())
.setKv(toKeyValueProto(tsKvEntry))
.setVersion(tsKvEntry.getVersion())
.build();
.setKv(toKeyValueProto(tsKvEntry));
if (tsKvEntry.getVersion() != null) {
builder.setVersion(tsKvEntry.getVersion());
}
return builder.build();
}
public static TransportProtos.KeyValueProto toKeyValueProto(KvEntry kvEntry) {

2
dao/src/main/java/org/thingsboard/server/dao/sql/cf/DefaultNativeCalculatedFieldRepository.java

@ -77,7 +77,7 @@ public class DefaultNativeCalculatedFieldRepository implements NativeCalculatedF
String name = (String) row.get("name");
int configurationVersion = (int) row.get("configuration_version");
JsonNode configuration = JacksonUtil.toJsonNode((String) row.get("configuration"));
long version = (long) row.get("version");
long version = row.get("version") != null ? (long) row.get("version") : 0;
Object externalIdObj = row.get("external_id");
CalculatedField calculatedField = new CalculatedField();

Loading…
Cancel
Save