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

Fix notification request update handling in cluster mode; Rule nodes fixes

pull/8265/head
ViacheslavKlimov 4 years ago
parent
0f8a3048a7
commit
7aa215f49d
7 changed files with 37 additions and 26 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 17
      application/src/main/java/org/thingsboard/server/controller/NotificationTemplateController.java
  2. 1
      application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java
  3. 2
      application/src/main/java/org/thingsboard/server/service/slack/DefaultSlackService.java
  4. 35
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
  5. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java
  6. 4
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNodeConfiguration.java
  7. 2
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbSlackNode.java

17
application/src/main/java/org/thingsboard/server/controller/NotificationTemplateController.java
View File

@ -17,6 +17,7 @@ package org.thingsboard.server.controller;
import io.swagger.annotations.ApiOperation;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.security.core.annotation.AuthenticationPrincipal;
import org.springframework.web.bind.annotation.DeleteMapping;
@ -136,16 +137,20 @@ public class NotificationTemplateController extends BaseController {
@GetMapping("/slack/conversations")
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN')")
public List<SlackConversation> listSlackConversations(@RequestParam SlackConversationType type,
@RequestParam(required = false) String token,
@AuthenticationPrincipal SecurityUser user) {
// generic permission
NotificationSettings settings = notificationSettingsService.findNotificationSettings(user.getTenantId());
SlackNotificationDeliveryMethodConfig slackConfig = (SlackNotificationDeliveryMethodConfig)
settings.getDeliveryMethodsConfigs().get(NotificationDeliveryMethod.SLACK);
if (slackConfig == null) {
throw new IllegalArgumentException("Slack is not configured");
if (StringUtils.isEmpty(token)) {
NotificationSettings settings = notificationSettingsService.findNotificationSettings(user.getTenantId());
SlackNotificationDeliveryMethodConfig slackConfig = (SlackNotificationDeliveryMethodConfig)
settings.getDeliveryMethodsConfigs().get(NotificationDeliveryMethod.SLACK);
if (slackConfig == null) {
throw new IllegalArgumentException("Slack is not configured");
}
token = slackConfig.getBotToken();
}
return slackService.listConversations(user.getTenantId(), slackConfig.getBotToken(), type);
return slackService.listConversations(user.getTenantId(), token, type);
}
}

1
application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java
View File

@ -364,6 +364,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
.deleted(true)
.build());
} else if (notificationRequest.isScheduled()) {
// TODO: just forward to scheduler service
clusterService.broadcastEntityStateChangeEvent(tenantId, notificationRequestId, ComponentLifecycleEvent.DELETED);
}
}

2
application/src/main/java/org/thingsboard/server/service/slack/DefaultSlackService.java
View File

@ -155,7 +155,7 @@ public class DefaultSlackService implements SlackService {
String neededScope = response.getNeeded();
error = "bot token scope '" + neededScope + "' is needed";
}
throw new RuntimeException("Failed to send message via Slack: " + error);
throw new RuntimeException("Slack API error: " + error);
}
return response;

35
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
View File

@ -316,20 +316,11 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
Set<TbSubscription> subscriptions = subscriptionsByEntityId.get(recipientId);
if (subscriptions != null) {
NotificationsSubscriptionUpdate subscriptionUpdate = new NotificationsSubscriptionUpdate(notificationUpdate);
log.trace("Handling notificationUpdate for user {}: {}", recipientId, notificationUpdate);
subscriptions.stream()
.filter(subscription -> subscription.getType() == TbSubscriptionType.NOTIFICATIONS
|| subscription.getType() == TbSubscriptionType.NOTIFICATIONS_COUNT)
.forEach(subscription -> {
if (serviceId.equals(subscription.getServiceId())) {
localSubscriptionService.onSubscriptionUpdate(subscription.getSessionId(),
subscription.getSubscriptionId(), subscriptionUpdate, TbCallback.EMPTY);
} else {
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
ToCoreNotificationMsg updateProto = TbSubscriptionUtils.notificationsSubUpdateToProto(subscription, subscriptionUpdate);
TbProtoQueueMsg<ToCoreNotificationMsg> queueMsg = new TbProtoQueueMsg<>(subscription.getEntityId().getId(), updateProto);
toCoreNotificationsProducer.send(tpi, queueMsg, null);
}
});
.forEach(subscription -> onNotificationsSubUpdate(subscriptionUpdate, subscription));
}
callback.onSuccess();
}
@ -341,21 +332,37 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
if (entityId.getEntityType() != EntityType.USER) {
return;
}
log.trace("Handling notificationRequestUpdate for user {}: {}", entityId, notificationRequestUpdate);
subscriptions.forEach(subscription -> {
if (subscription.getType() != TbSubscriptionType.NOTIFICATIONS &&
subscription.getType() != TbSubscriptionType.NOTIFICATIONS_COUNT) {
return;
}
if (!subscription.getTenantId().equals(tenantId) || !subscription.getServiceId().equals(serviceId)) {
if (!subscription.getTenantId().equals(tenantId)) {
return;
}
localSubscriptionService.onSubscriptionUpdate(subscription.getSessionId(), subscription.getSubscriptionId(),
subscriptionUpdate, TbCallback.EMPTY);
onNotificationsSubUpdate(subscriptionUpdate, subscription);
});
});
callback.onSuccess();
}
private void onNotificationsSubUpdate(NotificationsSubscriptionUpdate subscriptionUpdate, TbSubscription subscription) {
if (serviceId.equals(subscription.getServiceId())) {
log.trace("[{}][{}][{}] Subscription session is managed by current service, forwarding to localSubscriptionService (update: {})",
subscription.getServiceId(), subscription.getEntityId(), subscription.getSessionId(), subscriptionUpdate);
localSubscriptionService.onSubscriptionUpdate(subscription.getSessionId(),
subscription.getSubscriptionId(), subscriptionUpdate, TbCallback.EMPTY);
} else {
log.trace("[{}][{}][{}] Subscription session is not managed by current service (update: {})",
subscription.getServiceId(), subscription.getEntityId(), subscription.getSessionId(), subscriptionUpdate);
TopicPartitionInfo tpi = notificationsTopicService.getNotificationsTopic(ServiceType.TB_CORE, subscription.getServiceId());
ToCoreNotificationMsg updateProto = TbSubscriptionUtils.notificationsSubUpdateToProto(subscription, subscriptionUpdate);
TbProtoQueueMsg<ToCoreNotificationMsg> queueMsg = new TbProtoQueueMsg<>(subscription.getEntityId().getId(), updateProto);
toCoreNotificationsProducer.send(tpi, queueMsg, null);
}
}
@Override
public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, boolean notifyDevice, TbCallback callback) {
onLocalTelemetrySubUpdate(entityId,

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNode.java
View File

@ -64,7 +64,7 @@ public class TbNotificationNode implements TbNode {
NotificationRequest notificationRequest = NotificationRequest.builder()
.tenantId(ctx.getTenantId())
.targets(config.getTargets())
.templateId(new NotificationTemplateId(config.getTemplateId()))
.templateId(config.getTemplateId())
.info(notificationInfo)
.additionalConfig(new NotificationRequestConfig())
.originatorEntityId(ctx.getSelf().getRuleChainId())

4
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbNotificationNodeConfiguration.java
View File

@ -17,9 +17,7 @@ package org.thingsboard.rule.engine.notification;
import lombok.Data;
import org.thingsboard.rule.engine.api.NodeConfiguration;
import org.thingsboard.server.common.data.id.NotificationTargetId;
import org.thingsboard.server.common.data.id.NotificationTemplateId;
import org.thingsboard.server.common.data.notification.NotificationRequestConfig;
import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;
@ -32,7 +30,7 @@ public class TbNotificationNodeConfiguration implements NodeConfiguration<TbNoti
@NotEmpty
private List<UUID> targets;
@NotNull
private UUID templateId;
private NotificationTemplateId templateId;
@Override
public TbNotificationNodeConfiguration defaultConfiguration() {

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/notification/TbSlackNode.java
View File

@ -29,7 +29,7 @@ import java.util.concurrent.ExecutionException;
@RuleNode(
type = ComponentType.EXTERNAL,
name = "send to Slack",
name = "send to slack",
configClazz = TbSlackNodeConfiguration.class,
nodeDescription = "Send message via Slack",
nodeDetails = "Sends message to a Slack channel or user",

Write Preview
Loading…
Cancel
Save