dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

use separate dbPool for async DB Tasks

Browse Source
This commit is contained in:
vparomskiy 2018-05-24 13:13:56 +03:00
parent e90a708185
commit 76e852bff0
4 changed files with 5 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractAlarmNode.java
View File

@ -62,7 +62,7 @@ public abstract class TbAbstractAlarmNode<C extends TbAbstractAlarmNodeConfigura
ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Cleared"); ctx.tellNext(toAlarmMsg(ctx, alarmResult, msg), "Cleared");
} }
}, },
t -> ctx.tellFailure(msg, t)); t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} }
protected abstract ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg); protected abstract ListenableFuture<AlarmResult> processAlarm(TbContext ctx, TbMsg msg);

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbAbstractGetAttributesNode.java
View File

@ -69,7 +69,7 @@ public abstract class TbAbstractGetAttributesNode<C extends TbGetAttributesNodeC
putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, config.getSharedAttributeNames(), "shared_"), putAttrAsync(ctx, entityId, msg, SHARED_SCOPE, config.getSharedAttributeNames(), "shared_"),
putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, config.getServerAttributeNames(), "ss_") putAttrAsync(ctx, entityId, msg, SERVER_SCOPE, config.getServerAttributeNames(), "ss_")
); );
withCallback(allFutures, i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t)); withCallback(allFutures, i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} }
private ListenableFuture<Void> putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List<String> keys, String prefix) { private ListenableFuture<Void> putAttrAsync(TbContext ctx, EntityId entityId, TbMsg msg, String scope, List<String> keys, String prefix) {

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/metadata/TbEntityGetAttrNode.java
View File

@ -54,7 +54,7 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode
withCallback( withCallback(
findEntityAsync(ctx, msg.getOriginator()), findEntityAsync(ctx, msg.getOriginator()),
entityId -> safeGetAttributes(ctx, msg, entityId), entityId -> safeGetAttributes(ctx, msg, entityId),
t -> ctx.tellFailure(msg, t)); t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} catch (Throwable th) { } catch (Throwable th) {
ctx.tellFailure(msg, th); ctx.tellFailure(msg, th);
} }
@ -68,7 +68,7 @@ public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode
withCallback(config.isTelemetry() ? getLatestTelemetry(ctx, entityId) : getAttributesAsync(ctx, entityId), withCallback(config.isTelemetry() ? getLatestTelemetry(ctx, entityId) : getAttributesAsync(ctx, entityId),
attributes -> putAttributesAndTell(ctx, msg, attributes), attributes -> putAttributesAndTell(ctx, msg, attributes),
t -> ctx.tellFailure(msg, t)); t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
} }
private ListenableFuture<List<KvEntry>> getAttributesAsync(TbContext ctx, EntityId entityId) { private ListenableFuture<List<KvEntry>> getAttributesAsync(TbContext ctx, EntityId entityId) {

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/transform/TbChangeOriginatorNode.java
View File

@ -69,7 +69,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode {
return null; return null;
} }
return ctx.transformMsg(msg, msg.getType(), n, msg.getMetaData(), msg.getData()); return ctx.transformMsg(msg, msg.getType(), n, msg.getMetaData(), msg.getData());
}); }, ctx.getDbCallbackExecutor());
} }
private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, EntityId original) { private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, EntityId original) {