diff --git a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java index c2e8fd048a..542a82f160 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/DefaultActorService.java @@ -66,10 +66,7 @@ public class DefaultActorService implements ActorService { public static final String APP_DISPATCHER_NAME = "app-dispatcher"; public static final String CORE_DISPATCHER_NAME = "core-dispatcher"; public static final String SYSTEM_RULE_DISPATCHER_NAME = "system-rule-dispatcher"; - public static final String SYSTEM_PLUGIN_DISPATCHER_NAME = "system-plugin-dispatcher"; public static final String TENANT_RULE_DISPATCHER_NAME = "rule-dispatcher"; - public static final String TENANT_PLUGIN_DISPATCHER_NAME = "plugin-dispatcher"; - public static final String SESSION_DISPATCHER_NAME = "session-dispatcher"; public static final String RPC_DISPATCHER_NAME = "rpc-dispatcher"; @Autowired @@ -186,8 +183,8 @@ public class DefaultActorService implements ActorService { @Override public void onReceivedMsg(ServerAddress source, ClusterAPIProtos.ClusterMessage msg) { ServerAddress serverAddress = new ServerAddress(source.getHost(), source.getPort(), source.getServerType()); - log.info("Received msg [{}] from [{}]", msg.getMessageType().name(), serverAddress); if (log.isDebugEnabled()) { + log.info("Received msg [{}] from [{}]", msg.getMessageType().name(), serverAddress); log.info("MSG: ", msg); } switch (msg.getMessageType()) { diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java index 3b0deea78a..0b70162215 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -1,12 +1,12 @@ /** * 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 - * + *

+ * 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. @@ -166,7 +166,7 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio private SubscriptionState getUpdatedSubscriptionState(EntityId entityId, SubscriptionState sub, EntityView entityView) { Map keyStates; - if(sub.isAllKeys()) { + if (sub.isAllKeys()) { keyStates = entityView.getKeys().getTimeseries().stream().collect(Collectors.toMap(k -> k, k -> 0L)); } else { keyStates = sub.getKeyStates().entrySet() @@ -618,7 +618,9 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio builder.setEntityId(sub.getEntityId().getId().toString()); builder.setType(sub.getType().name()); builder.setAllKeys(sub.isAllKeys()); - builder.setScope(sub.getScope()); + if (sub.getScope() != null) { + builder.setScope(sub.getScope()); + } sub.getKeyStates().entrySet().forEach(e -> builder.addKeyStates( ClusterAPIProtos.SubscriptionKetStateProto.newBuilder().setKey(e.getKey()).setTs(e.getValue()).build())); rpcService.tell(address, ClusterAPIProtos.MessageType.CLUSTER_TELEMETRY_SUBSCRIPTION_CREATE_MESSAGE, builder.build().toByteArray()); diff --git a/application/src/main/resources/actor-system.conf b/application/src/main/resources/actor-system.conf index 3c687759e2..763319e52c 100644 --- a/application/src/main/resources/actor-system.conf +++ b/application/src/main/resources/actor-system.conf @@ -92,7 +92,7 @@ core-dispatcher { throughput = 5 } -# This dispatcher is used for system rule actors +# This dispatcher is used for system rule chains and rule node actors system-rule-dispatcher { type = Dispatcher executor = "fork-join-executor" @@ -115,30 +115,7 @@ system-rule-dispatcher { throughput = 5 } -# This dispatcher is used for system plugin actors -system-plugin-dispatcher { - type = Dispatcher - executor = "fork-join-executor" - fork-join-executor { - # Min number of threads to cap factor-based parallelism number to - parallelism-min = 2 - # Max number of threads to cap factor-based parallelism number to - parallelism-max = 12 - - # The parallelism factor is used to determine thread pool size using the - # following formula: ceil(available processors * factor). Resulting size - # is then bounded by the parallelism-min and parallelism-max values. - parallelism-factor = 0.25 - } - # How long time the dispatcher will wait for new actors until it shuts down - shutdown-timeout = 1s - - # Throughput defines the number of messages that are processed in a batch - # before the thread is returned to the pool. Set to 1 for as fair as possible. - throughput = 5 -} - -# This dispatcher is used for tenant rule actors +# This dispatcher is used for tenant rule chains and rule node actors rule-dispatcher { type = Dispatcher executor = "fork-join-executor" @@ -160,50 +137,3 @@ rule-dispatcher { # before the thread is returned to the pool. Set to 1 for as fair as possible. throughput = 5 } - -# This dispatcher is used for tenant plugin actors -plugin-dispatcher { - type = Dispatcher - executor = "fork-join-executor" - fork-join-executor { - # Min number of threads to cap factor-based parallelism number to - parallelism-min = 2 - # Max number of threads to cap factor-based parallelism number to - parallelism-max = 12 - - # The parallelism factor is used to determine thread pool size using the - # following formula: ceil(available processors * factor). Resulting size - # is then bounded by the parallelism-min and parallelism-max values. - parallelism-factor = 0.25 - } - # How long time the dispatcher will wait for new actors until it shuts down - shutdown-timeout = 1s - - # Throughput defines the number of messages that are processed in a batch - # before the thread is returned to the pool. Set to 1 for as fair as possible. - throughput = 5 -} - - -# This dispatcher is used for rule actors -session-dispatcher { - type = Dispatcher - executor = "fork-join-executor" - fork-join-executor { - # Min number of threads to cap factor-based parallelism number to - parallelism-min = 2 - # Max number of threads to cap factor-based parallelism number to - parallelism-max = 12 - - # The parallelism factor is used to determine thread pool size using the - # following formula: ceil(available processors * factor). Resulting size - # is then bounded by the parallelism-min and parallelism-max values. - parallelism-factor = 0.25 - } - # How long time the dispatcher will wait for new actors until it shuts down - shutdown-timeout = 1s - - # Throughput defines the number of messages that are processed in a batch - # before the thread is returned to the pool. Set to 1 for as fair as possible. - throughput = 5 -} \ No newline at end of file diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNodeConfiguration.java index c568e3dd94..e3766baf7b 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/debug/TbMsgGeneratorNodeConfiguration.java @@ -37,7 +37,7 @@ public class TbMsgGeneratorNodeConfiguration implements NodeConfiguration - generateReport +