tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge with hotfix branch

pull/9427/head
Andrii Shvaika 3 years ago
parent
ce8eba2137 ee4fdde56f
commit
7b0a3c282c
8 changed files with 171 additions and 110 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 161
      application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
  2. 31
      application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java
  3. 73
      application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
  4. 1
      application/src/main/resources/thingsboard.yml
  5. 3
      application/src/test/resources/application-test.properties
  6. 4
      common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java
  7. 4
      dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java
  8. 4
      ui-ngx/src/app/modules/home/components/widget/lib/gateway/gateway-service-rpc.component.ts

161
application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View File

@ -16,16 +16,13 @@
package org.thingsboard.server.service.edge; package org.thingsboard.server.service.edge;
import com.fasterxml.jackson.databind.node.ObjectNode; import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardThreadFactory; import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.audit.ActionType;
import org.thingsboard.server.common.data.edge.Edge; import org.thingsboard.server.common.data.edge.Edge;
import org.thingsboard.server.common.data.edge.EdgeEventType; import org.thingsboard.server.common.data.edge.EdgeEventType;
@ -59,7 +56,7 @@ import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit;
@Service @Service
@TbCoreComponent @TbCoreComponent
@ -128,17 +125,20 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
@Autowired @Autowired
protected ApplicationEventPublisher eventPublisher; protected ApplicationEventPublisher eventPublisher;
private ExecutorService dbCallBackExecutor; @Value("${actors.system.edge_dispatcher_pool_size:4}")
private int edgeDispatcherSize;
private ExecutorService executor;
@PostConstruct @PostConstruct
public void initExecutor() { public void initExecutor() {
dbCallBackExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("edge-notifications")); executor = ThingsBoardExecutors.newWorkStealingPool(edgeDispatcherSize, "edge-notifications");
} }
@PreDestroy @PreDestroy
public void shutdownExecutor() { public void shutdownExecutor() {
if (dbCallBackExecutor != null) { if (executor != null) {
dbCallBackExecutor.shutdownNow(); executor.shutdownNow();
} }
} }
@ -157,79 +157,78 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) { public void pushNotificationToEdge(TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg, TbCallback callback) {
TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB())); TenantId tenantId = TenantId.fromUUID(new UUID(edgeNotificationMsg.getTenantIdMSB(), edgeNotificationMsg.getTenantIdLSB()));
log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg); log.debug("[{}] Pushing notification to edge {}", tenantId, edgeNotificationMsg);
final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
try { try {
EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); executor.submit(() -> {
ListenableFuture<Void> future; try {
switch (type) { if (deadline < System.nanoTime()) {
case EDGE: log.warn("[{}] Skipping notification message because deadline reached {}", tenantId, edgeNotificationMsg);
future = edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg); return;
break; }
case ASSET: EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType());
future = assetProcessor.processEntityNotification(tenantId, edgeNotificationMsg); switch (type) {
break; case EDGE:
case DEVICE: edgeProcessor.processEdgeNotification(tenantId, edgeNotificationMsg);
future = deviceProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case ASSET:
case ENTITY_VIEW: assetProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = entityViewProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case DEVICE:
case DASHBOARD: deviceProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = dashboardProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case ENTITY_VIEW:
case RULE_CHAIN: entityViewProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = ruleChainProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case DASHBOARD:
case USER: dashboardProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = userProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case RULE_CHAIN:
case CUSTOMER: ruleChainProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg); break;
break; case USER:
case DEVICE_PROFILE: userProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = deviceProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case CUSTOMER:
case ASSET_PROFILE: customerProcessor.processCustomerNotification(tenantId, edgeNotificationMsg);
future = assetProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case DEVICE_PROFILE:
case OTA_PACKAGE: deviceProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = otaPackageProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case ASSET_PROFILE:
case WIDGETS_BUNDLE: assetProfileProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = widgetBundleProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case OTA_PACKAGE:
case WIDGET_TYPE: otaPackageProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = widgetTypeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case WIDGETS_BUNDLE:
case QUEUE: widgetBundleProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = queueProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case WIDGET_TYPE:
case ALARM: widgetTypeProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg); break;
break; case QUEUE:
case RELATION: queueProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
future = relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg); break;
break; case ALARM:
case TENANT: alarmProcessor.processAlarmNotification(tenantId, edgeNotificationMsg);
future = tenantEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case RELATION:
case TENANT_PROFILE: relationProcessor.processRelationNotification(tenantId, edgeNotificationMsg);
future = tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break;
break; case TENANT:
default: tenantEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); break;
future = Futures.immediateFuture(null); case TENANT_PROFILE:
} tenantProfileEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg);
Futures.addCallback(future, new FutureCallback<>() { break;
@Override default:
public void onSuccess(@Nullable Void unused) { log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type);
callback.onSuccess(); }
} } catch (Exception e) {
callBackFailure(tenantId, edgeNotificationMsg, callback, e);
@Override
public void onFailure(Throwable throwable) {
callBackFailure(tenantId, edgeNotificationMsg, callback, throwable);
} }
}, dbCallBackExecutor); });
callback.onSuccess();
} catch (Exception e) { } catch (Exception e) {
callBackFailure(tenantId, edgeNotificationMsg, callback, e); callBackFailure(tenantId, edgeNotificationMsg, callback, e);
} }

31
application/src/main/java/org/thingsboard/server/service/notification/rule/DefaultNotificationRuleProcessor.java
View File

@ -77,18 +77,17 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
public void process(NotificationRuleTrigger trigger) { public void process(NotificationRuleTrigger trigger) {
NotificationRuleTriggerType triggerType = trigger.getType(); NotificationRuleTriggerType triggerType = trigger.getType();
TenantId tenantId = triggerType.isTenantLevel() ? trigger.getTenantId() : TenantId.SYS_TENANT_ID; TenantId tenantId = triggerType.isTenantLevel() ? trigger.getTenantId() : TenantId.SYS_TENANT_ID;
notificationExecutor.submit(() -> {
try { try {
List<NotificationRule> enabledRules = notificationRulesCache.getEnabled(tenantId, triggerType); List<NotificationRule> enabledRules = notificationRulesCache.getEnabled(tenantId, triggerType);
if (enabledRules.isEmpty()) { if (enabledRules.isEmpty()) {
return; return;
} }
if (trigger.deduplicate()) { if (trigger.deduplicate()) {
enabledRules = new ArrayList<>(enabledRules); enabledRules = new ArrayList<>(enabledRules);
enabledRules.removeIf(rule -> deduplicationService.alreadyProcessed(trigger, rule)); enabledRules.removeIf(rule -> deduplicationService.alreadyProcessed(trigger, rule));
} }
final List<NotificationRule> rules = enabledRules; final List<NotificationRule> rules = enabledRules;
notificationExecutor.submit(() -> {
for (NotificationRule rule : rules) { for (NotificationRule rule : rules) {
try { try {
processNotificationRule(rule, trigger); processNotificationRule(rule, trigger);
@ -96,10 +95,10 @@ public class DefaultNotificationRuleProcessor implements NotificationRuleProcess
log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e); log.error("Failed to process notification rule {} for trigger type {} with trigger object {}", rule.getId(), rule.getTriggerType(), trigger, e);
} }
} }
}); } catch (Throwable e) {
} catch (Throwable e) { log.error("Failed to process notification rules for trigger: {}", trigger, e);
log.error("Failed to process notification rules for trigger: {}", trigger, e); }
} });
} }
private void processNotificationRule(NotificationRule rule, NotificationRuleTrigger trigger) { private void processNotificationRule(NotificationRule rule, NotificationRuleTrigger trigger) {

73
application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java
View File

@ -36,10 +36,22 @@ public class TbCoreConsumerStats {
public static final String DEVICE_CLAIMS = "claimDevice"; public static final String DEVICE_CLAIMS = "claimDevice";
public static final String DEVICE_STATES = "deviceState"; public static final String DEVICE_STATES = "deviceState";
public static final String SUBSCRIPTION_MSGS = "subMsgs"; public static final String SUBSCRIPTION_MSGS = "subMsgs";
public static final String TO_CORE_NOTIFICATIONS = "coreNfs";
public static final String EDGE_NOTIFICATIONS = "edgeNfs"; public static final String EDGE_NOTIFICATIONS = "edgeNfs";
public static final String DEVICE_ACTIVITIES = "deviceActivity"; public static final String DEVICE_ACTIVITIES = "deviceActivity";
public static final String TO_CORE_NF_OTHER = "coreNfOther"; // normally, there is no messages when codebase is fine
public static final String TO_CORE_NF_COMPONENT_LIFECYCLE = "coreNfCompLfcl";
public static final String TO_CORE_NF_DEVICE_RPC_RESPONSE = "coreNfDevRpcRsp";
public static final String TO_CORE_NF_EDGE_EVENT_UPDATE = "coreNfEdgeUpd";
public static final String TO_CORE_NF_EDGE_SYNC_REQUEST = "coreNfEdgeSyncReq";
public static final String TO_CORE_NF_EDGE_SYNC_RESPONSE = "coreNfEdgeSyncResp";
public static final String TO_CORE_NF_NOTIFICATION_RULE_PROCESSOR = "coreNfNfRlProc";
public static final String TO_CORE_NF_QUEUE_UPDATE = "coreNfQueueUpd";
public static final String TO_CORE_NF_QUEUE_DELETE = "coreNfQueueDel";
public static final String TO_CORE_NF_SUBSCRIPTION_SERVICE = "coreNfSubSvc";
public static final String TO_CORE_NF_SUBSCRIPTION_MANAGER = "coreNfSubMgr";
public static final String TO_CORE_NF_VC_RESPONSE = "coreNfVCRsp";
private final StatsCounter totalCounter; private final StatsCounter totalCounter;
private final StatsCounter sessionEventCounter; private final StatsCounter sessionEventCounter;
private final StatsCounter getAttributesCounter; private final StatsCounter getAttributesCounter;
@ -48,14 +60,25 @@ public class TbCoreConsumerStats {
private final StatsCounter toDeviceRPCCallResponseCounter; private final StatsCounter toDeviceRPCCallResponseCounter;
private final StatsCounter subscriptionInfoCounter; private final StatsCounter subscriptionInfoCounter;
private final StatsCounter claimDeviceCounter; private final StatsCounter claimDeviceCounter;
private final StatsCounter deviceStateCounter; private final StatsCounter deviceStateCounter;
private final StatsCounter subscriptionMsgCounter; private final StatsCounter subscriptionMsgCounter;
private final StatsCounter toCoreNotificationsCounter;
private final StatsCounter edgeNotificationsCounter; private final StatsCounter edgeNotificationsCounter;
private final StatsCounter deviceActivitiesCounter; private final StatsCounter deviceActivitiesCounter;
private final List<StatsCounter> counters = new ArrayList<>(); private final StatsCounter toCoreNfOtherCounter;
private final StatsCounter toCoreNfComponentLifecycleCounter;
private final StatsCounter toCoreNfDeviceRpcResponseCounter;
private final StatsCounter toCoreNfEdgeEventUpdateCounter;
private final StatsCounter toCoreNfEdgeSyncRequestCounter;
private final StatsCounter toCoreNfEdgeSyncResponseCounter;
private final StatsCounter toCoreNfNotificationRuleProcessorCounter;
private final StatsCounter toCoreNfQueueUpdateCounter;
private final StatsCounter toCoreNfQueueDeleteCounter;
private final StatsCounter toCoreNfSubscriptionServiceCounter;
private final StatsCounter toCoreNfSubscriptionManagerCounter;
private final StatsCounter toCoreNfVersionControlResponseCounter;
private final List<StatsCounter> counters = new ArrayList<>(24);
public TbCoreConsumerStats(StatsFactory statsFactory) { public TbCoreConsumerStats(StatsFactory statsFactory) {
String statsKey = StatsType.CORE.getName(); String statsKey = StatsType.CORE.getName();
@ -70,9 +93,23 @@ public class TbCoreConsumerStats {
this.claimDeviceCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_CLAIMS)); this.claimDeviceCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_CLAIMS));
this.deviceStateCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_STATES)); this.deviceStateCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_STATES));
this.subscriptionMsgCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS)); this.subscriptionMsgCounter = register(statsFactory.createStatsCounter(statsKey, SUBSCRIPTION_MSGS));
this.toCoreNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NOTIFICATIONS));
this.edgeNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, EDGE_NOTIFICATIONS)); this.edgeNotificationsCounter = register(statsFactory.createStatsCounter(statsKey, EDGE_NOTIFICATIONS));
this.deviceActivitiesCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_ACTIVITIES)); this.deviceActivitiesCounter = register(statsFactory.createStatsCounter(statsKey, DEVICE_ACTIVITIES));
// Core notification counters
this.toCoreNfOtherCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_OTHER));
this.toCoreNfComponentLifecycleCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_COMPONENT_LIFECYCLE));
this.toCoreNfDeviceRpcResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_DEVICE_RPC_RESPONSE));
this.toCoreNfEdgeEventUpdateCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_EVENT_UPDATE));
this.toCoreNfEdgeSyncRequestCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_REQUEST));
this.toCoreNfEdgeSyncResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_EDGE_SYNC_RESPONSE));
this.toCoreNfNotificationRuleProcessorCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_NOTIFICATION_RULE_PROCESSOR));
this.toCoreNfQueueUpdateCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_QUEUE_UPDATE));
this.toCoreNfQueueDeleteCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_QUEUE_DELETE));
this.toCoreNfSubscriptionServiceCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_SUBSCRIPTION_SERVICE));
this.toCoreNfSubscriptionManagerCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_SUBSCRIPTION_MANAGER));
this.toCoreNfVersionControlResponseCounter = register(statsFactory.createStatsCounter(statsKey, TO_CORE_NF_VC_RESPONSE));
} }
private StatsCounter register(StatsCounter counter){ private StatsCounter register(StatsCounter counter){
@ -127,7 +164,31 @@ public class TbCoreConsumerStats {
public void log(TransportProtos.ToCoreNotificationMsg msg) { public void log(TransportProtos.ToCoreNotificationMsg msg) {
totalCounter.increment(); totalCounter.increment();
toCoreNotificationsCounter.increment(); if (msg.hasToLocalSubscriptionServiceMsg()) {
toCoreNfSubscriptionServiceCounter.increment();
} else if (msg.hasFromDeviceRpcResponse()) {
toCoreNfDeviceRpcResponseCounter.increment();
} else if (!msg.getComponentLifecycleMsg().isEmpty()) {
toCoreNfComponentLifecycleCounter.increment();
} else if (!msg.getEdgeEventUpdateMsg().isEmpty()) {
toCoreNfEdgeEventUpdateCounter.increment();
} else if (!msg.getToEdgeSyncRequestMsg().isEmpty()) {
toCoreNfEdgeSyncRequestCounter.increment();
} else if (!msg.getFromEdgeSyncResponseMsg().isEmpty()) {
toCoreNfEdgeSyncResponseCounter.increment();
} else if (msg.hasQueueUpdateMsg()) {
toCoreNfQueueUpdateCounter.increment();
} else if (msg.hasQueueDeleteMsg()) {
toCoreNfQueueDeleteCounter.increment();
} else if (msg.hasVcResponseMsg()) {
toCoreNfVersionControlResponseCounter.increment();
} else if (msg.hasToSubscriptionMgrMsg()) {
toCoreNfSubscriptionManagerCounter.increment();
} else if (msg.hasNotificationRuleProcessorMsg()) {
toCoreNfNotificationRuleProcessorCounter.increment();
} else {
toCoreNfOtherCounter.increment();
}
} }
public void printStats() { public void printStats() {

1
application/src/main/resources/thingsboard.yml
View File

@ -418,6 +418,7 @@ actors:
tenant_dispatcher_pool_size: "${ACTORS_SYSTEM_TENANT_DISPATCHER_POOL_SIZE:2}" # Thread pool size for actor system dispatcher that process messages for tenant actors tenant_dispatcher_pool_size: "${ACTORS_SYSTEM_TENANT_DISPATCHER_POOL_SIZE:2}" # Thread pool size for actor system dispatcher that process messages for tenant actors
device_dispatcher_pool_size: "${ACTORS_SYSTEM_DEVICE_DISPATCHER_POOL_SIZE:4}" # Thread pool size for actor system dispatcher that process messages for device actors device_dispatcher_pool_size: "${ACTORS_SYSTEM_DEVICE_DISPATCHER_POOL_SIZE:4}" # Thread pool size for actor system dispatcher that process messages for device actors
rule_dispatcher_pool_size: "${ACTORS_SYSTEM_RULE_DISPATCHER_POOL_SIZE:8}" # Thread pool size for actor system dispatcher that process messages for rule engine (chain/node) actors rule_dispatcher_pool_size: "${ACTORS_SYSTEM_RULE_DISPATCHER_POOL_SIZE:8}" # Thread pool size for actor system dispatcher that process messages for rule engine (chain/node) actors
edge_dispatcher_pool_size: "${ACTORS_SYSTEM_EDGE_DISPATCHER_POOL_SIZE:4}" # Thread pool size for actor system dispatcher that process messages for edge actors
tenant: tenant:
create_components_on_init: "${ACTORS_TENANT_CREATE_COMPONENTS_ON_INIT:true}" # Create components in initialization create_components_on_init: "${ACTORS_TENANT_CREATE_COMPONENTS_ON_INIT:true}" # Create components in initialization
session: session:

3
application/src/test/resources/application-test.properties
View File

@ -12,7 +12,6 @@ transport.lwm2m.security.trust-credentials.keystore.store_file=lwm2m/credentials
# Edge disabled to speed up the context init. Will be enabled by @TestPropertySource in respective tests # Edge disabled to speed up the context init. Will be enabled by @TestPropertySource in respective tests
edges.enabled=false edges.enabled=false
actors.rpc.submit_strategy=BURST actors.rpc.submit_strategy=BURST
queue.rule-engine.stats.enabled=true
# Transports disabled to speed up the context init. Particular transport will be enabled with @TestPropertySource in respective tests # Transports disabled to speed up the context init. Particular transport will be enabled with @TestPropertySource in respective tests
transport.http.enabled=false transport.http.enabled=false
@ -59,7 +58,9 @@ queue.rule-engine.queues[2].processing-strategy.retries=1
queue.rule-engine.queues[2].processing-strategy.pause-between-retries=0 queue.rule-engine.queues[2].processing-strategy.pause-between-retries=0
queue.rule-engine.queues[2].processing-strategy.max-pause-between-retries=0 queue.rule-engine.queues[2].processing-strategy.max-pause-between-retries=0
queue.rule-engine.stats.enabled=true
usage.stats.report.enabled=false usage.stats.report.enabled=false
queue.core.stats.enabled=true
sql.audit_logs.partition_size=24 sql.audit_logs.partition_size=24
sql.ttl.audit_logs.ttl=2592000 sql.ttl.audit_logs.ttl=2592000

4
common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaProducerTemplate.java
View File

@ -85,8 +85,8 @@ public class TbKafkaProducerTemplate<T extends TbQueueMsg> implements TbQueuePro
if (log.isTraceEnabled()) { if (log.isTraceEnabled()) {
try { try {
StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace(); StackTraceElement[] stackTrace = Thread.currentThread().getStackTrace();
int maxlevel = Math.min(stackTrace.length, 10); int maxLevel = Math.min(stackTrace.length, 20);
for (int i = 2; i < maxlevel; i++) { // ignore two levels: getStackTrace and addAnalyticHeaders for (int i = 2; i < maxLevel; i++) { // ignore two levels: getStackTrace and addAnalyticHeaders
headers.add(new RecordHeader("_stackTrace" + i, stackTrace[i].toString().getBytes(StandardCharsets.UTF_8))); headers.add(new RecordHeader("_stackTrace" + i, stackTrace[i].toString().getBytes(StandardCharsets.UTF_8)));
} }
} catch (Throwable t) { } catch (Throwable t) {

4
dao/src/main/java/org/thingsboard/server/dao/tenant/TenantServiceImpl.java
View File

@ -197,7 +197,7 @@ public class TenantServiceImpl extends AbstractCachedEntityService<TenantId, Ten
boolean create = tenant.getId() == null; boolean create = tenant.getId() == null;
Tenant savedTenant = tenantDao.save(tenant.getId(), tenant); Tenant savedTenant = tenantDao.save(tenant.getId(), tenant);
publishEvictEvent(new TenantEvictEvent(savedTenant.getId(), create)); publishEvictEvent(new TenantEvictEvent(savedTenant.getId(), create));
eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(TenantId.SYS_TENANT_ID).entityId(savedTenant.getId()).added(create).build()); eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(savedTenant.getId()).entityId(savedTenant.getId()).added(create).build());
if (tenant.getId() == null) { if (tenant.getId() == null) {
deviceProfileService.createDefaultDeviceProfile(savedTenant.getId()); deviceProfileService.createDefaultDeviceProfile(savedTenant.getId());
assetProfileService.createDefaultAssetProfile(savedTenant.getId()); assetProfileService.createDefaultAssetProfile(savedTenant.getId());
@ -244,7 +244,7 @@ public class TenantServiceImpl extends AbstractCachedEntityService<TenantId, Ten
adminSettingsService.deleteAdminSettingsByTenantId(tenantId); adminSettingsService.deleteAdminSettingsByTenantId(tenantId);
tenantDao.removeById(tenantId, tenantId.getId()); tenantDao.removeById(tenantId, tenantId.getId());
publishEvictEvent(new TenantEvictEvent(tenantId, true)); publishEvictEvent(new TenantEvictEvent(tenantId, true));
eventPublisher.publishEvent(DeleteEntityEvent.builder().tenantId(TenantId.SYS_TENANT_ID).entityId(tenantId).build()); eventPublisher.publishEvent(DeleteEntityEvent.builder().tenantId(tenantId).entityId(tenantId).build());
relationService.deleteEntityRelations(tenantId, tenantId); relationService.deleteEntityRelations(tenantId, tenantId);
alarmService.deleteEntityAlarmRecordsByTenantId(tenantId); alarmService.deleteEntityAlarmRecordsByTenantId(tenantId);
} }

4
ui-ngx/src/app/modules/home/components/widget/lib/gateway/gateway-service-rpc.component.ts
View File

@ -61,7 +61,7 @@ export class GatewayServiceRPCComponent implements AfterViewInit {
this.commandForm = this.fb.group({ this.commandForm = this.fb.group({
command: [null,[Validators.required]], command: [null,[Validators.required]],
time: [60, [Validators.required, Validators.min(1)]], time: [60, [Validators.required, Validators.min(1)]],
params: [{}, [jsonRequired]], params: ['{}', [jsonRequired]],
result: [null] result: [null]
}); });
} }
@ -78,7 +78,7 @@ export class GatewayServiceRPCComponent implements AfterViewInit {
sendCommand() { sendCommand() {
const formValues = this.commandForm.value; const formValues = this.commandForm.value;
const commandPrefix = this.isConnector ? `${this.connectorType}_` : 'gateway_'; const commandPrefix = this.isConnector ? `${this.connectorType}_` : 'gateway_';
this.ctx.controlApi.sendTwoWayCommand(commandPrefix+formValues.command.toLowerCase(), {},formValues.time).subscribe({ this.ctx.controlApi.sendTwoWayCommand(commandPrefix+formValues.command.toLowerCase(), formValues.params,formValues.time).subscribe({
next: resp => this.commandForm.get('result').setValue(JSON.stringify(resp)), next: resp => this.commandForm.get('result').setValue(JSON.stringify(resp)),
error: error => { error: error => {
console.log(error); console.log(error);

Write Preview
Loading…
Cancel
Save