21 changed files with 470 additions and 306 deletions
@ -0,0 +1,117 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2018 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.rule.engine.action; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.fasterxml.jackson.databind.ObjectMapper; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.thingsboard.rule.engine.api.*; |
||||
|
import org.thingsboard.server.common.data.alarm.Alarm; |
||||
|
import org.thingsboard.server.common.msg.TbMsg; |
||||
|
import org.thingsboard.server.common.msg.TbMsgMetaData; |
||||
|
|
||||
|
import static org.thingsboard.rule.engine.DonAsynchron.withCallback; |
||||
|
|
||||
|
@Slf4j |
||||
|
public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfiguration> implements TbNode { |
||||
|
|
||||
|
static final String PREV_ALARM_DETAILS = "prevAlarmDetails"; |
||||
|
|
||||
|
static final String IS_NEW_ALARM = "isNewAlarm"; |
||||
|
static final String IS_EXISTING_ALARM = "isExistingAlarm"; |
||||
|
static final String IS_CLEARED_ALARM = "isClearedAlarm"; |
||||
|
|
||||
|
private final ObjectMapper mapper = new ObjectMapper(); |
||||
|
|
||||
|
protected C config; |
||||
|
private ScriptEngine buildDetailsJsEngine; |
||||
|
|
||||
|
@Override |
||||
|
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
||||
|
this.config = loadAlarmNodeConfig(configuration); |
||||
|
this.buildDetailsJsEngine = ctx.createJsScriptEngine(config.getAlarmDetailsBuildJs(), "Details"); |
||||
|
} |
||||
|
|
||||
|
protected abstract C loadAlarmNodeConfig(TbNodeConfiguration configuration) throws TbNodeException; |
||||
|
|
||||
|
@Override |
||||
|
public void onMsg(TbContext ctx, TbMsg msg) { |
||||
|
withCallback(processAlarm(ctx, msg), |
||||
|
alarmResult -> { |
||||
|
if (alarmResult.alarm == null) { |
||||
|
ctx.tellNext(msg, "False"); |
||||
|
} else if (alarmResult.isCreated) { |
||||
|
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Created"); |
||||
|
} else if (alarmResult.isUpdated) { |
||||
|
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Updated"); |
||||
|
} else if (alarmResult.isCleared) { |
||||
|
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Cleared"); |
||||
|
} |
||||
|
}, |
||||
|
t -> ctx.tellError(msg, t)); |
||||
|
} |
||||
|
|
||||
|
protected abstract ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg); |
||||
|
|
||||
|
protected ListenableFuture<JsonNode> buildAlarmDetails(TbContext ctx, TbMsg msg, JsonNode previousDetails) { |
||||
|
return ctx.getJsExecutor().executeAsync(() -> { |
||||
|
TbMsg theMsg = msg; |
||||
|
if (previousDetails != null) { |
||||
|
TbMsgMetaData metaData = msg.getMetaData().copy(); |
||||
|
metaData.putValue(PREV_ALARM_DETAILS, mapper.writeValueAsString(previousDetails)); |
||||
|
theMsg = ctx.newMsg(msg.getType(), msg.getOriginator(), metaData, msg.getData()); |
||||
|
} |
||||
|
return buildDetailsJsEngine.executeJson(theMsg); |
||||
|
}); |
||||
|
} |
||||
|
|
||||
|
private TbMsg toAlarmMsg(TbContext ctx, AlarmResult alarmResult, TbMsg originalMsg) { |
||||
|
JsonNode jsonNodes = mapper.valueToTree(alarmResult.alarm); |
||||
|
String data = jsonNodes.toString(); |
||||
|
TbMsgMetaData metaData = originalMsg.getMetaData().copy(); |
||||
|
if (alarmResult.isCreated) { |
||||
|
metaData.putValue(IS_NEW_ALARM, Boolean.TRUE.toString()); |
||||
|
} else if (alarmResult.isUpdated) { |
||||
|
metaData.putValue(IS_EXISTING_ALARM, Boolean.TRUE.toString()); |
||||
|
} else if (alarmResult.isCleared) { |
||||
|
metaData.putValue(IS_CLEARED_ALARM, Boolean.TRUE.toString()); |
||||
|
} |
||||
|
return ctx.transformMsg(originalMsg, "ALARM", originalMsg.getOriginator(), metaData, data); |
||||
|
} |
||||
|
|
||||
|
|
||||
|
@Override |
||||
|
public void destroy() { |
||||
|
if (buildDetailsJsEngine != null) { |
||||
|
buildDetailsJsEngine.destroy(); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
protected static class AlarmResult { |
||||
|
boolean isCreated; |
||||
|
boolean isUpdated; |
||||
|
boolean isCleared; |
||||
|
Alarm alarm; |
||||
|
|
||||
|
AlarmResult(boolean isCreated, boolean isUpdated, boolean isCleared, Alarm alarm) { |
||||
|
this.isCreated = isCreated; |
||||
|
this.isUpdated = isUpdated; |
||||
|
this.isCleared = isCleared; |
||||
|
this.alarm = alarm; |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,26 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2018 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.rule.engine.action; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
|
||||
|
@Data |
||||
|
public abstract class TbAbstractAlarmNodeConfiguration { |
||||
|
|
||||
|
private String alarmType; |
||||
|
private String alarmDetailsBuildJs; |
||||
|
|
||||
|
} |
||||
@ -1,219 +0,0 @@ |
|||||
/** |
|
||||
* Copyright © 2016-2018 The Thingsboard Authors |
|
||||
* |
|
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|
||||
* you may not use this file except in compliance with the License. |
|
||||
* You may obtain a copy of the License at |
|
||||
* |
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
|
||||
* |
|
||||
* Unless required by applicable law or agreed to in writing, software |
|
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|
||||
* See the License for the specific language governing permissions and |
|
||||
* limitations under the License. |
|
||||
*/ |
|
||||
package org.thingsboard.rule.engine.action; |
|
||||
|
|
||||
import com.fasterxml.jackson.databind.JsonNode; |
|
||||
import com.fasterxml.jackson.databind.ObjectMapper; |
|
||||
import com.google.common.base.Function; |
|
||||
import com.google.common.util.concurrent.AsyncFunction; |
|
||||
import com.google.common.util.concurrent.Futures; |
|
||||
import com.google.common.util.concurrent.ListenableFuture; |
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.thingsboard.rule.engine.TbNodeUtils; |
|
||||
import org.thingsboard.rule.engine.api.*; |
|
||||
import org.thingsboard.server.common.data.alarm.Alarm; |
|
||||
import org.thingsboard.server.common.data.alarm.AlarmStatus; |
|
||||
import org.thingsboard.server.common.data.id.TenantId; |
|
||||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|
||||
import org.thingsboard.server.common.msg.TbMsg; |
|
||||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|
||||
|
|
||||
import static org.thingsboard.rule.engine.DonAsynchron.withCallback; |
|
||||
|
|
||||
@Slf4j |
|
||||
@RuleNode( |
|
||||
type = ComponentType.ACTION, |
|
||||
name = "alarm", relationTypes = {"Created", "Updated", "Cleared", "False"}, |
|
||||
configClazz = TbAlarmNodeConfiguration.class, |
|
||||
nodeDescription = "Create/Update/Clear Alarm", |
|
||||
nodeDetails = "isAlarm - JS function that verifies if Alarm should be CREATED for incoming message.\n" + |
|
||||
"isCleared - JS function that verifies if Alarm should be CLEARED for incoming message.\n" + |
|
||||
"Details - JS function that creates JSON object based on incoming message. This object will be added into Alarm.details field.\n" + |
|
||||
"Node output:\n" + |
|
||||
"If alarm was not created, original message is returned. Otherwise new Message returned with type 'ALARM', Alarm object in 'msg' property and 'matadata' will contains one of those properties 'isNewAlarm/isExistingAlarm/isClearedAlarm' " + |
|
||||
"Message payload can be accessed via <code>msg</code> property. For example <code>'temperature = ' + msg.temperature ;</code>" + |
|
||||
"Message metadata can be accessed via <code>metadata</code> property. For example <code>'name = ' + metadata.customerName;</code>", |
|
||||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|
||||
configDirective = "tbActionNodeAlarmConfig", |
|
||||
icon = "notifications_active" |
|
||||
) |
|
||||
|
|
||||
public class TbAlarmNode implements TbNode { |
|
||||
|
|
||||
static final String IS_NEW_ALARM = "isNewAlarm"; |
|
||||
static final String IS_EXISTING_ALARM = "isExistingAlarm"; |
|
||||
static final String IS_CLEARED_ALARM = "isClearedAlarm"; |
|
||||
|
|
||||
private final ObjectMapper mapper = new ObjectMapper(); |
|
||||
|
|
||||
private TbAlarmNodeConfiguration config; |
|
||||
private ScriptEngine createJsEngine; |
|
||||
private ScriptEngine clearJsEngine; |
|
||||
private ScriptEngine buildDetailsJsEngine; |
|
||||
|
|
||||
@Override |
|
||||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|
||||
this.config = TbNodeUtils.convert(configuration, TbAlarmNodeConfiguration.class); |
|
||||
this.createJsEngine = ctx.createJsScriptEngine(config.getCreateConditionJs(), "isAlarm"); |
|
||||
this.clearJsEngine = ctx.createJsScriptEngine(config.getClearConditionJs(), "isCleared"); |
|
||||
this.buildDetailsJsEngine = ctx.createJsScriptEngine(config.getAlarmDetailsBuildJs(), "Details"); |
|
||||
} |
|
||||
|
|
||||
@Override |
|
||||
public void onMsg(TbContext ctx, TbMsg msg) { |
|
||||
ListeningExecutor jsExecutor = ctx.getJsExecutor(); |
|
||||
|
|
||||
ListenableFuture<Boolean> shouldCreate = jsExecutor.executeAsync(() -> createJsEngine.executeFilter(msg)); |
|
||||
ListenableFuture<AlarmResult> transform = Futures.transform(shouldCreate, (AsyncFunction<Boolean, AlarmResult>) create -> { |
|
||||
if (create) { |
|
||||
return createOrUpdate(ctx, msg); |
|
||||
} else { |
|
||||
return checkForClearIfExist(ctx, msg); |
|
||||
} |
|
||||
}, ctx.getDbCallbackExecutor()); |
|
||||
|
|
||||
withCallback(transform, |
|
||||
alarmResult -> { |
|
||||
if (alarmResult.alarm == null) { |
|
||||
ctx.tellNext(msg, "False"); |
|
||||
} else if (alarmResult.isCreated) { |
|
||||
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Created"); |
|
||||
} else if (alarmResult.isUpdated) { |
|
||||
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Updated"); |
|
||||
} else if (alarmResult.isCleared) { |
|
||||
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Cleared"); |
|
||||
} |
|
||||
}, |
|
||||
t -> ctx.tellError(msg, t)); |
|
||||
|
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<AlarmResult> createOrUpdate(TbContext ctx, TbMsg msg) { |
|
||||
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType()); |
|
||||
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> { |
|
||||
if (a == null || a.getStatus().isCleared()) { |
|
||||
return createNewAlarm(ctx, msg); |
|
||||
} else { |
|
||||
return updateAlarm(ctx, msg, a); |
|
||||
} |
|
||||
}, ctx.getDbCallbackExecutor()); |
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<AlarmResult> checkForClearIfExist(TbContext ctx, TbMsg msg) { |
|
||||
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType()); |
|
||||
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> { |
|
||||
if (a != null && !a.getStatus().isCleared()) { |
|
||||
return clearAlarm(ctx, msg, a); |
|
||||
} |
|
||||
return Futures.immediateFuture(new AlarmResult(false, false, false, null)); |
|
||||
}, ctx.getDbCallbackExecutor()); |
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg) { |
|
||||
ListenableFuture<Alarm> asyncAlarm = Futures.transform(buildAlarmDetails(ctx, msg), |
|
||||
(Function<JsonNode, Alarm>) details -> buildAlarm(msg, details, ctx.getTenantId())); |
|
||||
ListenableFuture<Alarm> asyncCreated = Futures.transform(asyncAlarm, |
|
||||
(Function<Alarm, Alarm>) alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor()); |
|
||||
return Futures.transform(asyncCreated, (Function<Alarm, AlarmResult>) alarm -> new AlarmResult(true, false, false, alarm)); |
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { |
|
||||
ListenableFuture<Alarm> asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg), (Function<JsonNode, Alarm>) details -> { |
|
||||
alarm.setSeverity(config.getSeverity()); |
|
||||
alarm.setPropagate(config.isPropagate()); |
|
||||
alarm.setDetails(details); |
|
||||
alarm.setEndTs(System.currentTimeMillis()); |
|
||||
return ctx.getAlarmService().createOrUpdateAlarm(alarm); |
|
||||
}, ctx.getDbCallbackExecutor()); |
|
||||
|
|
||||
return Futures.transform(asyncUpdated, (Function<Alarm, AlarmResult>) a -> new AlarmResult(false, true, false, a)); |
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<AlarmResult> clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { |
|
||||
ListenableFuture<Boolean> shouldClear = ctx.getJsExecutor().executeAsync(() -> clearJsEngine.executeFilter(msg)); |
|
||||
return Futures.transform(shouldClear, (AsyncFunction<Boolean, AlarmResult>) clear -> { |
|
||||
if (clear) { |
|
||||
ListenableFuture<Boolean> clearFuture = ctx.getAlarmService().clearAlarm(alarm.getId(), System.currentTimeMillis()); |
|
||||
return Futures.transform(clearFuture, (Function<Boolean, AlarmResult>) cleared -> { |
|
||||
alarm.setStatus(alarm.getStatus().isAck() ? AlarmStatus.CLEARED_ACK : AlarmStatus.CLEARED_UNACK); |
|
||||
return new AlarmResult(false, false, true, alarm); |
|
||||
}); |
|
||||
} |
|
||||
return Futures.immediateFuture(new AlarmResult(false, false, false, null)); |
|
||||
}); |
|
||||
} |
|
||||
|
|
||||
private Alarm buildAlarm(TbMsg msg, JsonNode details, TenantId tenantId) { |
|
||||
return Alarm.builder() |
|
||||
.tenantId(tenantId) |
|
||||
.originator(msg.getOriginator()) |
|
||||
.status(AlarmStatus.ACTIVE_UNACK) |
|
||||
.severity(config.getSeverity()) |
|
||||
.propagate(config.isPropagate()) |
|
||||
.type(config.getAlarmType()) |
|
||||
//todo-vp: alarm date should be taken from Message or current Time should be used?
|
|
||||
// .startTs(System.currentTimeMillis())
|
|
||||
// .endTs(System.currentTimeMillis())
|
|
||||
.details(details) |
|
||||
.build(); |
|
||||
} |
|
||||
|
|
||||
private ListenableFuture<JsonNode> buildAlarmDetails(TbContext ctx, TbMsg msg) { |
|
||||
return ctx.getJsExecutor().executeAsync(() -> buildDetailsJsEngine.executeJson(msg)); |
|
||||
} |
|
||||
|
|
||||
private TbMsg toAlarmMsg(TbContext ctx, AlarmResult alarmResult, TbMsg originalMsg) { |
|
||||
JsonNode jsonNodes = mapper.valueToTree(alarmResult.alarm); |
|
||||
String data = jsonNodes.toString(); |
|
||||
TbMsgMetaData metaData = originalMsg.getMetaData().copy(); |
|
||||
if (alarmResult.isCreated) { |
|
||||
metaData.putValue(IS_NEW_ALARM, Boolean.TRUE.toString()); |
|
||||
} else if (alarmResult.isUpdated) { |
|
||||
metaData.putValue(IS_EXISTING_ALARM, Boolean.TRUE.toString()); |
|
||||
} else if (alarmResult.isCleared) { |
|
||||
metaData.putValue(IS_CLEARED_ALARM, Boolean.TRUE.toString()); |
|
||||
} |
|
||||
return ctx.transformMsg(originalMsg, "ALARM", originalMsg.getOriginator(), metaData, data); |
|
||||
} |
|
||||
|
|
||||
|
|
||||
@Override |
|
||||
public void destroy() { |
|
||||
if (createJsEngine != null) { |
|
||||
createJsEngine.destroy(); |
|
||||
} |
|
||||
if (clearJsEngine != null) { |
|
||||
clearJsEngine.destroy(); |
|
||||
} |
|
||||
if (buildDetailsJsEngine != null) { |
|
||||
buildDetailsJsEngine.destroy(); |
|
||||
} |
|
||||
} |
|
||||
|
|
||||
private static class AlarmResult { |
|
||||
boolean isCreated; |
|
||||
boolean isUpdated; |
|
||||
boolean isCleared; |
|
||||
Alarm alarm; |
|
||||
|
|
||||
AlarmResult(boolean isCreated, boolean isUpdated, boolean isCleared, Alarm alarm) { |
|
||||
this.isCreated = isCreated; |
|
||||
this.isUpdated = isUpdated; |
|
||||
this.isCleared = isCleared; |
|
||||
this.alarm = alarm; |
|
||||
} |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,80 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2018 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.rule.engine.action; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.google.common.util.concurrent.AsyncFunction; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.thingsboard.rule.engine.TbNodeUtils; |
||||
|
import org.thingsboard.rule.engine.api.RuleNode; |
||||
|
import org.thingsboard.rule.engine.api.TbContext; |
||||
|
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
||||
|
import org.thingsboard.rule.engine.api.TbNodeException; |
||||
|
import org.thingsboard.server.common.data.alarm.Alarm; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmStatus; |
||||
|
import org.thingsboard.server.common.data.plugin.ComponentType; |
||||
|
import org.thingsboard.server.common.msg.TbMsg; |
||||
|
|
||||
|
@Slf4j |
||||
|
@RuleNode( |
||||
|
type = ComponentType.ACTION, |
||||
|
name = "clear alarm", relationTypes = {"Cleared", "False"}, |
||||
|
configClazz = TbClearAlarmNodeConfiguration.class, |
||||
|
nodeDescription = "Clear Alarm", |
||||
|
nodeDetails = |
||||
|
"Details - JS function that creates JSON object based on incoming message. This object will be added into Alarm.details field.\n" + |
||||
|
"Node output:\n" + |
||||
|
"If alarm was not cleared, original message is returned. Otherwise new Message returned with type 'ALARM', Alarm object in 'msg' property and 'matadata' will contains 'isClearedAlarm' property " + |
||||
|
"Message payload can be accessed via <code>msg</code> property. For example <code>'temperature = ' + msg.temperature ;</code>" + |
||||
|
"Message metadata can be accessed via <code>metadata</code> property. For example <code>'name = ' + metadata.customerName;</code>", |
||||
|
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
||||
|
configDirective = "tbActionNodeClearAlarmConfig", |
||||
|
icon = "notifications_active" |
||||
|
) |
||||
|
public class TbClearAlarmNode extends TbAbstractAlarmNode<TbClearAlarmNodeConfiguration> { |
||||
|
|
||||
|
@Override |
||||
|
protected TbClearAlarmNodeConfiguration loadAlarmNodeConfig(TbNodeConfiguration configuration) throws TbNodeException { |
||||
|
return TbNodeUtils.convert(configuration, TbClearAlarmNodeConfiguration.class); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) { |
||||
|
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType()); |
||||
|
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> { |
||||
|
if (a != null && !a.getStatus().isCleared()) { |
||||
|
return clearAlarm(ctx, msg, a); |
||||
|
} |
||||
|
return Futures.immediateFuture(new AlarmResult(false, false, false, null)); |
||||
|
}, ctx.getDbCallbackExecutor()); |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<AlarmResult> clearAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { |
||||
|
ListenableFuture<JsonNode> asyncDetails = buildAlarmDetails(ctx, msg, alarm.getDetails()); |
||||
|
return Futures.transform(asyncDetails, (AsyncFunction<JsonNode, AlarmResult>) details -> { |
||||
|
ListenableFuture<Boolean> clearFuture = ctx.getAlarmService().clearAlarm(alarm.getId(), details, System.currentTimeMillis()); |
||||
|
return Futures.transform(clearFuture, (AsyncFunction<Boolean, AlarmResult>) cleared -> { |
||||
|
if (cleared && details != null) { |
||||
|
alarm.setDetails(details); |
||||
|
} |
||||
|
alarm.setStatus(alarm.getStatus().isAck() ? AlarmStatus.CLEARED_ACK : AlarmStatus.CLEARED_UNACK); |
||||
|
return Futures.immediateFuture(new AlarmResult(false, false, true, alarm)); |
||||
|
}); |
||||
|
}, ctx.getDbCallbackExecutor()); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,36 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2018 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.rule.engine.action; |
||||
|
|
||||
|
import lombok.Data; |
||||
|
import org.thingsboard.rule.engine.api.NodeConfiguration; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmSeverity; |
||||
|
|
||||
|
@Data |
||||
|
public class TbClearAlarmNodeConfiguration extends TbAbstractAlarmNodeConfiguration implements NodeConfiguration<TbClearAlarmNodeConfiguration> { |
||||
|
|
||||
|
@Override |
||||
|
public TbClearAlarmNodeConfiguration defaultConfiguration() { |
||||
|
TbClearAlarmNodeConfiguration configuration = new TbClearAlarmNodeConfiguration(); |
||||
|
configuration.setAlarmDetailsBuildJs("var details = {};\n" + |
||||
|
"if (metadata.prevAlarmDetails) {\n" + |
||||
|
" details = JSON.parse(metadata.prevAlarmDetails);\n" + |
||||
|
"}\n" + |
||||
|
"return details;"); |
||||
|
configuration.setAlarmType("General Alarm"); |
||||
|
return configuration; |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,106 @@ |
|||||
|
/** |
||||
|
* Copyright © 2016-2018 The Thingsboard Authors |
||||
|
* |
||||
|
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
|
* you may not use this file except in compliance with the License. |
||||
|
* You may obtain a copy of the License at |
||||
|
* |
||||
|
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
* |
||||
|
* Unless required by applicable law or agreed to in writing, software |
||||
|
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
|
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
|
* See the License for the specific language governing permissions and |
||||
|
* limitations under the License. |
||||
|
*/ |
||||
|
package org.thingsboard.rule.engine.action; |
||||
|
|
||||
|
import com.fasterxml.jackson.databind.JsonNode; |
||||
|
import com.google.common.base.Function; |
||||
|
import com.google.common.util.concurrent.AsyncFunction; |
||||
|
import com.google.common.util.concurrent.Futures; |
||||
|
import com.google.common.util.concurrent.ListenableFuture; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.thingsboard.rule.engine.TbNodeUtils; |
||||
|
import org.thingsboard.rule.engine.api.RuleNode; |
||||
|
import org.thingsboard.rule.engine.api.TbContext; |
||||
|
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
||||
|
import org.thingsboard.rule.engine.api.TbNodeException; |
||||
|
import org.thingsboard.server.common.data.alarm.Alarm; |
||||
|
import org.thingsboard.server.common.data.alarm.AlarmStatus; |
||||
|
import org.thingsboard.server.common.data.id.TenantId; |
||||
|
import org.thingsboard.server.common.data.plugin.ComponentType; |
||||
|
import org.thingsboard.server.common.msg.TbMsg; |
||||
|
|
||||
|
@Slf4j |
||||
|
@RuleNode( |
||||
|
type = ComponentType.ACTION, |
||||
|
name = "create alarm", relationTypes = {"Created", "Updated", "False"}, |
||||
|
configClazz = TbCreateAlarmNodeConfiguration.class, |
||||
|
nodeDescription = "Create or Update Alarm", |
||||
|
nodeDetails = |
||||
|
"Details - JS function that creates JSON object based on incoming message. This object will be added into Alarm.details field.\n" + |
||||
|
"Node output:\n" + |
||||
|
"If alarm was not created, original message is returned. Otherwise new Message returned with type 'ALARM', Alarm object in 'msg' property and 'matadata' will contains one of those properties 'isNewAlarm/isExistingAlarm' " + |
||||
|
"Message payload can be accessed via <code>msg</code> property. For example <code>'temperature = ' + msg.temperature ;</code>" + |
||||
|
"Message metadata can be accessed via <code>metadata</code> property. For example <code>'name = ' + metadata.customerName;</code>", |
||||
|
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
||||
|
configDirective = "tbActionNodeCreateAlarmConfig", |
||||
|
icon = "notifications_active" |
||||
|
) |
||||
|
public class TbCreateAlarmNode extends TbAbstractAlarmNode<TbCreateAlarmNodeConfiguration> { |
||||
|
|
||||
|
@Override |
||||
|
protected TbCreateAlarmNodeConfiguration loadAlarmNodeConfig(TbNodeConfiguration configuration) throws TbNodeException { |
||||
|
return TbNodeUtils.convert(configuration, TbCreateAlarmNodeConfiguration.class); |
||||
|
} |
||||
|
|
||||
|
@Override |
||||
|
protected ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg) { |
||||
|
ListenableFuture<Alarm> latest = ctx.getAlarmService().findLatestByOriginatorAndType(ctx.getTenantId(), msg.getOriginator(), config.getAlarmType()); |
||||
|
return Futures.transform(latest, (AsyncFunction<Alarm, AlarmResult>) a -> { |
||||
|
if (a == null || a.getStatus().isCleared()) { |
||||
|
return createNewAlarm(ctx, msg); |
||||
|
} else { |
||||
|
return updateAlarm(ctx, msg, a); |
||||
|
} |
||||
|
}, ctx.getDbCallbackExecutor()); |
||||
|
|
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<AlarmResult> createNewAlarm(TbContext ctx, TbMsg msg) { |
||||
|
ListenableFuture<Alarm> asyncAlarm = Futures.transform(buildAlarmDetails(ctx, msg, null), |
||||
|
(Function<JsonNode, Alarm>) details -> buildAlarm(msg, details, ctx.getTenantId())); |
||||
|
ListenableFuture<Alarm> asyncCreated = Futures.transform(asyncAlarm, |
||||
|
(Function<Alarm, Alarm>) alarm -> ctx.getAlarmService().createOrUpdateAlarm(alarm), ctx.getDbCallbackExecutor()); |
||||
|
return Futures.transform(asyncCreated, (Function<Alarm, AlarmResult>) alarm -> new AlarmResult(true, false, false, alarm)); |
||||
|
} |
||||
|
|
||||
|
private ListenableFuture<AlarmResult> updateAlarm(TbContext ctx, TbMsg msg, Alarm alarm) { |
||||
|
ListenableFuture<Alarm> asyncUpdated = Futures.transform(buildAlarmDetails(ctx, msg, alarm.getDetails()), (Function<JsonNode, Alarm>) details -> { |
||||
|
alarm.setSeverity(config.getSeverity()); |
||||
|
alarm.setPropagate(config.isPropagate()); |
||||
|
alarm.setDetails(details); |
||||
|
alarm.setEndTs(System.currentTimeMillis()); |
||||
|
return ctx.getAlarmService().createOrUpdateAlarm(alarm); |
||||
|
}, ctx.getDbCallbackExecutor()); |
||||
|
|
||||
|
return Futures.transform(asyncUpdated, (Function<Alarm, AlarmResult>) a -> new AlarmResult(false, true, false, a)); |
||||
|
} |
||||
|
|
||||
|
private Alarm buildAlarm(TbMsg msg, JsonNode details, TenantId tenantId) { |
||||
|
return Alarm.builder() |
||||
|
.tenantId(tenantId) |
||||
|
.originator(msg.getOriginator()) |
||||
|
.status(AlarmStatus.ACTIVE_UNACK) |
||||
|
.severity(config.getSeverity()) |
||||
|
.propagate(config.isPropagate()) |
||||
|
.type(config.getAlarmType()) |
||||
|
//todo-vp: alarm date should be taken from Message or current Time should be used?
|
||||
|
// .startTs(System.currentTimeMillis())
|
||||
|
// .endTs(System.currentTimeMillis())
|
||||
|
.details(details) |
||||
|
.build(); |
||||
|
} |
||||
|
|
||||
|
} |
||||
File diff suppressed because one or more lines are too long
Loading…
Reference in new issue