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

DefaultNotificationCenter - less async operations

pull/8622/head
ViacheslavKlimov 4 years ago
parent
ce2caee2c8
commit
e558725ffa
6 changed files with 56 additions and 95 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. 102
      application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java
  2. 20
      application/src/main/java/org/thingsboard/server/service/notification/channels/EmailNotificationChannel.java
  3. 3
      application/src/main/java/org/thingsboard/server/service/notification/channels/NotificationChannel.java
  4. 10
      application/src/main/java/org/thingsboard/server/service/notification/channels/SlackNotificationChannel.java
  5. 13
      application/src/main/java/org/thingsboard/server/service/notification/channels/SmsNotificationChannel.java
  6. 3
      common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStats.java

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

@ -15,13 +15,10 @@
*/
package org.thingsboard.server.service.notification;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.DonAsynchron;
import org.thingsboard.rule.engine.api.NotificationCenter;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.User;
@ -31,6 +28,7 @@ import org.thingsboard.server.common.data.id.NotificationRuleId;
import org.thingsboard.server.common.data.id.NotificationTargetId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.id.UserId;
import org.thingsboard.server.common.data.limit.LimitedApi;
import org.thingsboard.server.common.data.notification.AlreadySentException;
import org.thingsboard.server.common.data.notification.Notification;
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
@ -59,14 +57,11 @@ import org.thingsboard.server.dao.notification.NotificationService;
import org.thingsboard.server.dao.notification.NotificationSettingsService;
import org.thingsboard.server.dao.notification.NotificationTargetService;
import org.thingsboard.server.dao.notification.NotificationTemplateService;
import org.thingsboard.server.dao.user.UserService;
import org.thingsboard.server.dao.util.limits.RateLimitService;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.queue.discovery.NotificationsTopicService;
import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
import org.thingsboard.server.common.data.limit.LimitedApi;
import org.thingsboard.server.dao.util.limits.RateLimitService;
import org.thingsboard.server.service.executors.DbCallbackExecutorService;
import org.thingsboard.server.service.executors.NotificationExecutorService;
import org.thingsboard.server.service.notification.channels.NotificationChannel;
import org.thingsboard.server.service.subscription.TbSubscriptionUtils;
@ -74,7 +69,6 @@ import org.thingsboard.server.service.telemetry.AbstractSubscriptionService;
import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate;
import org.thingsboard.server.service.ws.notification.sub.NotificationUpdate;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@ -87,7 +81,7 @@ import java.util.stream.Collectors;
@Service
@Slf4j
@RequiredArgsConstructor
@SuppressWarnings({"UnstableApiUsage", "rawtypes"})
@SuppressWarnings({"rawtypes"})
public class DefaultNotificationCenter extends AbstractSubscriptionService implements NotificationCenter, NotificationChannel<User, WebDeliveryMethodNotificationTemplate> {
private final NotificationTargetService notificationTargetService;
@ -97,7 +91,6 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
private final NotificationSettingsService notificationSettingsService;
private final UserService userService;
private final NotificationExecutorService notificationExecutor;
private final DbCallbackExecutorService dbCallbackExecutorService;
private final NotificationsTopicService notificationsTopicService;
private final TbQueueProducerProvider producerProvider;
private final RateLimitService rateLimitService;
@ -172,37 +165,32 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
.build();
notificationExecutor.submit(() -> {
List<ListenableFuture<Void>> results = new ArrayList<>();
for (NotificationTarget target : targets) {
List<ListenableFuture<Void>> result = processForTarget(target, ctx);
results.addAll(result);
processForTarget(target, ctx);
}
NotificationRequestId requestId = ctx.getRequest().getId();
log.debug("[{}] Notification request processing is finished", requestId);
NotificationRequestStats stats = ctx.getStats();
try {
notificationRequestService.updateNotificationRequest(tenantId, requestId, NotificationRequestStatus.SENT, stats);
} catch (Exception e) {
log.error("[{}] Failed to update stats for notification request", requestId, e);
}
Futures.whenAllComplete(results).run(() -> {
NotificationRequestId requestId = ctx.getRequest().getId();
log.debug("[{}] Notification request processing is finished", requestId);
NotificationRequestStats stats = ctx.getStats();
if (callback != null) {
try {
notificationRequestService.updateNotificationRequest(tenantId, requestId, NotificationRequestStatus.SENT, stats);
callback.accept(stats);
} catch (Exception e) {
log.error("[{}] Failed to update stats for notification request", requestId, e);
}
if (callback != null) {
try {
callback.accept(stats);
} catch (Exception e) {
log.error("Failed to process callback for notification request {}", requestId, e);
}
log.error("Failed to process callback for notification request {}", requestId, e);
}
}, dbCallbackExecutorService);
}
});
return request;
}
private List<ListenableFuture<Void>> processForTarget(NotificationTarget target, NotificationProcessingContext ctx) {
private void processForTarget(NotificationTarget target, NotificationProcessingContext ctx) {
Iterable<? extends NotificationRecipient> recipients;
switch (target.getConfiguration().getType()) {
case PLATFORM_USERS: {
@ -231,43 +219,35 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
Set<NotificationDeliveryMethod> deliveryMethods = new HashSet<>(ctx.getDeliveryMethods());
deliveryMethods.removeIf(deliveryMethod -> !target.getConfiguration().getType().getSupportedDeliveryMethods().contains(deliveryMethod));
log.debug("[{}] Processing notification request for {} target ({}) for delivery methods {}", ctx.getRequest().getId(), target.getConfiguration().getType(), target.getId(), deliveryMethods);
if (deliveryMethods.isEmpty()) {
return;
}
List<ListenableFuture<Void>> results = new ArrayList<>();
if (!deliveryMethods.isEmpty()) {
for (NotificationRecipient recipient : recipients) {
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
ListenableFuture<Void> resultFuture = processForRecipient(deliveryMethod, recipient, ctx);
DonAsynchron.withCallback(resultFuture, result -> {
ctx.getStats().reportSent(deliveryMethod, recipient);
}, error -> {
ctx.getStats().reportError(deliveryMethod, error, recipient);
});
results.add(resultFuture);
for (NotificationRecipient recipient : recipients) {
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
try {
processForRecipient(deliveryMethod, recipient, ctx);
ctx.getStats().reportSent(deliveryMethod, recipient);
} catch (Exception error) {
ctx.getStats().reportError(deliveryMethod, error, recipient);
}
}
}
return results;
}
private ListenableFuture<Void> processForRecipient(NotificationDeliveryMethod deliveryMethod, NotificationRecipient recipient, NotificationProcessingContext ctx) {
private void processForRecipient(NotificationDeliveryMethod deliveryMethod, NotificationRecipient recipient, NotificationProcessingContext ctx) throws Exception {
if (ctx.getStats().contains(deliveryMethod, recipient.getId())) {
return Futures.immediateFailedFuture(new AlreadySentException());
throw new AlreadySentException();
}
DeliveryMethodNotificationTemplate processedTemplate;
try {
processedTemplate = ctx.getProcessedTemplate(deliveryMethod, recipient);
} catch (Exception e) {
return Futures.immediateFailedFuture(e);
}
NotificationChannel notificationChannel = channels.get(deliveryMethod);
DeliveryMethodNotificationTemplate processedTemplate = ctx.getProcessedTemplate(deliveryMethod, recipient);
log.trace("[{}] Sending {} notification for recipient {}", ctx.getRequest().getId(), deliveryMethod, recipient);
return notificationChannel.sendNotification(recipient, processedTemplate, ctx);
notificationChannel.sendNotification(recipient, processedTemplate, ctx);
}
@Override
public ListenableFuture<Void> sendNotification(User recipient, WebDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
public void sendNotification(User recipient, WebDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
NotificationRequest request = ctx.getRequest();
Notification notification = Notification.builder()
.requestId(request.getId())
@ -283,14 +263,14 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
notification = notificationService.saveNotification(recipient.getTenantId(), notification);
} catch (Exception e) {
log.error("Failed to create notification for recipient {}", recipient.getId(), e);
return Futures.immediateFailedFuture(e);
throw e;
}
NotificationUpdate update = NotificationUpdate.builder()
.created(true)
.notification(notification)
.build();
return onNotificationUpdate(recipient.getTenantId(), recipient.getId(), update);
onNotificationUpdate(recipient.getTenantId(), recipient.getId(), update);
}
@Override
@ -384,13 +364,11 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
clusterService.pushMsgToCore(tenantId, notificationRequestId, toCoreMsg, null);
}
private ListenableFuture<Void> onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate update) {
private void onNotificationUpdate(TenantId tenantId, UserId recipientId, NotificationUpdate update) {
log.trace("Submitting notification update for recipient {}: {}", recipientId, update);
return Futures.submit(() -> {
forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> {
subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, TbCallback.EMPTY);
}, () -> TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, update));
}, wsCallBackExecutor);
forwardToSubscriptionManagerService(tenantId, recipientId, subscriptionManagerService -> {
subscriptionManagerService.onNotificationUpdate(tenantId, recipientId, update, TbCallback.EMPTY);
}, () -> TbSubscriptionUtils.notificationUpdateToProto(tenantId, recipientId, update));
}
private void onNotificationRequestUpdate(TenantId tenantId, NotificationRequestUpdate update) {

20
application/src/main/java/org/thingsboard/server/service/notification/channels/EmailNotificationChannel.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.service.notification.channels;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import org.thingsboard.rule.engine.api.MailService;
@ -24,7 +23,6 @@ import org.thingsboard.server.common.data.User;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
import org.thingsboard.server.common.data.notification.template.EmailDeliveryMethodNotificationTemplate;
import org.thingsboard.server.service.mail.MailExecutorService;
import org.thingsboard.server.service.notification.NotificationProcessingContext;
@Component
@ -32,19 +30,15 @@ import org.thingsboard.server.service.notification.NotificationProcessingContext
public class EmailNotificationChannel implements NotificationChannel<User, EmailDeliveryMethodNotificationTemplate> {
private final MailService mailService;
private final MailExecutorService executor;
@Override
public ListenableFuture<Void> sendNotification(User recipient, EmailDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
return executor.submit(() -> {
mailService.send(recipient.getTenantId(), null, TbEmail.builder()
.to(recipient.getEmail())
.subject(processedTemplate.getSubject())
.body(processedTemplate.getBody())
.html(true)
.build());
return null;
});
public void sendNotification(User recipient, EmailDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
mailService.send(recipient.getTenantId(), null, TbEmail.builder()
.to(recipient.getEmail())
.subject(processedTemplate.getSubject())
.body(processedTemplate.getBody())
.html(true)
.build());
}
@Override

3
application/src/main/java/org/thingsboard/server/service/notification/channels/NotificationChannel.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.service.notification.channels;
import com.google.common.util.concurrent.ListenableFuture;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
import org.thingsboard.server.common.data.notification.targets.NotificationRecipient;
@ -24,7 +23,7 @@ import org.thingsboard.server.service.notification.NotificationProcessingContext
public interface NotificationChannel<R extends NotificationRecipient, T extends DeliveryMethodNotificationTemplate> {
ListenableFuture<Void> sendNotification(R recipient, T processedTemplate, NotificationProcessingContext ctx);
void sendNotification(R recipient, T processedTemplate, NotificationProcessingContext ctx) throws Exception;
void check(TenantId tenantId) throws Exception;

10
application/src/main/java/org/thingsboard/server/service/notification/channels/SlackNotificationChannel.java
View File

@ -15,7 +15,6 @@
*/
package org.thingsboard.server.service.notification.channels;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import org.thingsboard.rule.engine.api.slack.SlackService;
@ -26,7 +25,6 @@ import org.thingsboard.server.common.data.notification.settings.SlackNotificatio
import org.thingsboard.server.common.data.notification.targets.slack.SlackConversation;
import org.thingsboard.server.common.data.notification.template.SlackDeliveryMethodNotificationTemplate;
import org.thingsboard.server.dao.notification.NotificationSettingsService;
import org.thingsboard.server.service.executors.ExternalCallExecutorService;
import org.thingsboard.server.service.notification.NotificationProcessingContext;
@Component
@ -35,15 +33,11 @@ public class SlackNotificationChannel implements NotificationChannel<SlackConver
private final SlackService slackService;
private final NotificationSettingsService notificationSettingsService;
private final ExternalCallExecutorService executor;
@Override
public ListenableFuture<Void> sendNotification(SlackConversation conversation, SlackDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
public void sendNotification(SlackConversation conversation, SlackDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
SlackNotificationDeliveryMethodConfig config = ctx.getDeliveryMethodConfig(NotificationDeliveryMethod.SLACK);
return executor.submit(() -> {
slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), processedTemplate.getBody());
return null;
});
slackService.sendMessage(ctx.getTenantId(), config.getBotToken(), conversation.getId(), processedTemplate.getBody());
}
@Override

13
application/src/main/java/org/thingsboard/server/service/notification/channels/SmsNotificationChannel.java
View File

@ -15,8 +15,6 @@
*/
package org.thingsboard.server.service.notification.channels;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
@ -26,26 +24,21 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
import org.thingsboard.server.common.data.notification.template.SmsDeliveryMethodNotificationTemplate;
import org.thingsboard.server.service.notification.NotificationProcessingContext;
import org.thingsboard.server.service.sms.SmsExecutorService;
@Component
@RequiredArgsConstructor
public class SmsNotificationChannel implements NotificationChannel<User, SmsDeliveryMethodNotificationTemplate> {
private final SmsService smsService;
private final SmsExecutorService executor;
@Override
public ListenableFuture<Void> sendNotification(User recipient, SmsDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) {
public void sendNotification(User recipient, SmsDeliveryMethodNotificationTemplate processedTemplate, NotificationProcessingContext ctx) throws Exception {
String phone = recipient.getPhone();
if (StringUtils.isBlank(phone)) {
return Futures.immediateFailedFuture(new RuntimeException("User does not have phone number"));
throw new RuntimeException("User does not have phone number");
}
return executor.submit(() -> {
smsService.sendSms(recipient.getTenantId(), recipient.getCustomerId(), new String[]{phone}, processedTemplate.getBody());
return null;
});
smsService.sendSms(recipient.getTenantId(), recipient.getCustomerId(), new String[]{phone}, processedTemplate.getBody());
}
@Override

3
common/data/src/main/java/org/thingsboard/server/common/data/notification/NotificationRequestStats.java
View File

@ -62,6 +62,9 @@ public class NotificationRequestStats {
return;
}
String errorMessage = error.getMessage();
if (errorMessage == null) {
errorMessage = error.getClass().getSimpleName();
}
errors.computeIfAbsent(deliveryMethod, k -> new ConcurrentHashMap<>()).put(recipient.getTitle(), errorMessage);
}

Write Preview
Loading…
Cancel
Save