@ -15,13 +15,8 @@
* /
package org.thingsboard.rule.engine.telemetry ;
import com.fasterxml.jackson.core.type.TypeReference ;
import com.google.common.util.concurrent.FutureCallback ;
import com.google.gson.JsonParser ;
import jakarta.annotation.Nullable ;
import lombok.extern.slf4j.Slf4j ;
import org.thingsboard.common.util.JacksonUtil ;
import org.thingsboard.rule.engine.api.AttributesDeleteRequest ;
import org.thingsboard.rule.engine.api.AttributesSaveRequest ;
import org.thingsboard.rule.engine.api.EmptyNodeConfiguration ;
import org.thingsboard.rule.engine.api.RuleNode ;
@ -29,7 +24,6 @@ import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode ;
import org.thingsboard.rule.engine.api.TbNodeConfiguration ;
import org.thingsboard.rule.engine.api.TbNodeException ;
import org.thingsboard.rule.engine.api.TimeseriesDeleteRequest ;
import org.thingsboard.rule.engine.api.TimeseriesSaveRequest ;
import org.thingsboard.rule.engine.api.util.TbNodeUtils ;
import org.thingsboard.server.common.adaptor.JsonConverter ;
@ -42,23 +36,23 @@ import org.thingsboard.server.common.data.plugin.ComponentType;
import org.thingsboard.server.common.msg.TbMsg ;
import java.util.ArrayList ;
import java.util.Collections ;
import java.util.List ;
import java.util.Map ;
import java.util.Optional ;
import static org.thingsboard.server.common.data.DataConstants.SCOPE ;
@Slf4j
@RuleNode (
type = ComponentType . ACTION ,
name = "apply to calculated fields" ,
name = "calculated fields" ,
configClazz = EmptyNodeConfiguration . class ,
nodeDescription = "Processes incoming messages for calculated fields" ,
nodeDetails = "This node processes incoming messages to update telemetry or attributes for predefined calculated fields without storing the original telemetry or attributes in the database. " +
"It ensures that calculated fields receive and process the necessary data without persisting the incoming values." ,
nodeDescription = "Pushes incoming messages to calculated fields service" ,
nodeDetails = "Node enables the processing of calculated fields without persisting incoming messages to the database. " +
"By default, the processing of calculated fields is triggered by the <b>save attributes</b> and <b>save time series</b> nodes. " +
"This rule node accepts the same messages as these nodes but allows you to trigger the processing of calculated " +
"fields independently, ensuring that derived data can be computed and utilized in real time without storing the original message in the database." ,
configDirective = "tbNodeEmptyConfig" ,
icon = "call_made"
icon = "published_with_changes "
)
public class TbCalculatedFieldsNode implements TbNode {
@ -74,8 +68,6 @@ public class TbCalculatedFieldsNode implements TbNode {
switch ( msg . getInternalType ( ) ) {
case POST_TELEMETRY_REQUEST - > processPostTelemetryRequest ( ctx , msg ) ;
case POST_ATTRIBUTES_REQUEST - > processPostAttributesRequest ( ctx , msg ) ;
case TIMESERIES_DELETED - > processTimeSeriesDeleted ( ctx , msg ) ;
case ATTRIBUTES_DELETED - > processAttributesDeleted ( ctx , msg ) ;
default - > ctx . tellFailure ( msg , new IllegalArgumentException ( "Unsupported msg type: " + msg . getType ( ) ) ) ;
}
}
@ -84,7 +76,7 @@ public class TbCalculatedFieldsNode implements TbNode {
Map < Long , List < KvEntry > > tsKvMap = JsonConverter . convertToTelemetry ( JsonParser . parseString ( msg . getData ( ) ) , System . currentTimeMillis ( ) ) ;
if ( tsKvMap . isEmpty ( ) ) {
ctx . tellFailure ( msg , new IllegalArgumentException ( "Msg body is empty: " + msg . getData ( ) ) ) ;
ctx . tellSuccess ( msg ) ;
return ;
}
@ -130,76 +122,4 @@ public class TbCalculatedFieldsNode implements TbNode {
ctx . getCalculatedFieldQueueService ( ) . pushRequestToQueue ( attributesSaveRequest , attributesSaveRequest . getCallback ( ) ) ;
}
private void processTimeSeriesDeleted ( TbContext ctx , TbMsg msg ) {
List < String > keysToDelete = Optional . ofNullable (
JacksonUtil . convertValue ( JacksonUtil . toJsonNode ( msg . getData ( ) ) . get ( "timeseries" ) , new TypeReference < List < String > > ( ) {
} )
) . orElse ( Collections . emptyList ( ) ) ;
if ( keysToDelete . isEmpty ( ) ) {
ctx . tellSuccess ( msg ) ;
return ;
}
TimeseriesDeleteRequest timeseriesDeleteRequest = TimeseriesDeleteRequest . builder ( )
. tenantId ( ctx . getTenantId ( ) )
. entityId ( msg . getOriginator ( ) )
. keys ( keysToDelete )
. previousCalculatedFieldIds ( msg . getPreviousCalculatedFieldIds ( ) )
. tbMsgId ( msg . getId ( ) )
. tbMsgType ( msg . getInternalType ( ) )
. callback ( new FutureCallback < List < String > > ( ) {
@Override
public void onSuccess ( @Nullable List < String > tmp ) {
ctx . tellSuccess ( msg ) ;
}
@Override
public void onFailure ( Throwable t ) {
ctx . tellFailure ( msg , t ) ;
}
} )
. build ( ) ;
ctx . getCalculatedFieldQueueService ( ) . pushRequestToQueue ( timeseriesDeleteRequest , keysToDelete , getCalculatedFieldCallback ( timeseriesDeleteRequest . getCallback ( ) , keysToDelete ) ) ;
}
private void processAttributesDeleted ( TbContext ctx , TbMsg msg ) {
List < String > keysToDelete = Optional . ofNullable (
JacksonUtil . convertValue ( JacksonUtil . toJsonNode ( msg . getData ( ) ) . get ( "attributes" ) , new TypeReference < List < String > > ( ) {
} )
) . orElse ( Collections . emptyList ( ) ) ;
if ( keysToDelete . isEmpty ( ) ) {
ctx . tellSuccess ( msg ) ;
return ;
}
AttributesDeleteRequest attributesDeleteRequest = AttributesDeleteRequest . builder ( )
. tenantId ( ctx . getTenantId ( ) )
. entityId ( msg . getOriginator ( ) )
. scope ( AttributeScope . valueOf ( msg . getMetaData ( ) . getValue ( SCOPE ) ) )
. keys ( keysToDelete )
. previousCalculatedFieldIds ( msg . getPreviousCalculatedFieldIds ( ) )
. tbMsgId ( msg . getId ( ) )
. tbMsgType ( msg . getInternalType ( ) )
. callback ( new TelemetryNodeCallback ( ctx , msg ) )
. build ( ) ;
ctx . getCalculatedFieldQueueService ( ) . pushRequestToQueue ( attributesDeleteRequest , keysToDelete , attributesDeleteRequest . getCallback ( ) ) ;
}
private FutureCallback < Void > getCalculatedFieldCallback ( FutureCallback < List < String > > originalCallback , List < String > keys ) {
return new FutureCallback < Void > ( ) {
@Override
public void onSuccess ( Void unused ) {
originalCallback . onSuccess ( keys ) ;
}
@Override
public void onFailure ( Throwable t ) {
originalCallback . onFailure ( t ) ;
}
} ;
}
}
IrynaMatveieva
1 year ago