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

Minor refactoring

pull/7980/head
ViacheslavKlimov 4 years ago
parent
43305ed2b4
commit
1a5f9bc16d
12 changed files with 112 additions and 89 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. 4
      application/src/main/data/upgrade/3.4.3/schema_update.sql
  2. 128
      application/src/main/java/org/thingsboard/server/service/notification/DefaultNotificationCenter.java
  3. 6
      application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java
  4. 1
      dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
  5. 4
      dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java
  6. 25
      dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java
  7. 5
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
  8. 2
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java
  9. 12
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java
  10. 8
      dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java
  11. 1
      dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java
  12. 5
      dao/src/main/resources/sql/schema-entities.sql

4
application/src/main/data/upgrade/3.4.3/schema_update.sql
View File

@ -74,13 +74,11 @@ CREATE TABLE IF NOT EXISTS notification (
type VARCHAR(32) NOT NULL,
subject VARCHAR(255),
text VARCHAR(1000) NOT NULL,
additional_config VARCHAR(500),
additional_config VARCHAR(1000),
info VARCHAR(1000),
status VARCHAR(32)
) PARTITION BY RANGE (created_time);
CREATE INDEX IF NOT EXISTS idx_notification_id_recipient_id ON notification(id, recipient_id);
CREATE INDEX IF NOT EXISTS idx_notification_recipient_id_status_created_time ON notification(recipient_id, status, created_time DESC);
ALTER TABLE alarm ADD COLUMN IF NOT EXISTS notification_rule_id UUID;
ALTER TABLE tb_user ADD COLUMN IF NOT EXISTS phone VARCHAR(255);

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

@ -58,6 +58,7 @@ 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.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;
import org.thingsboard.server.service.telemetry.AbstractSubscriptionService;
@ -85,6 +86,7 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
private final NotificationService notificationService;
private final NotificationTemplateService notificationTemplateService;
private final NotificationSettingsService notificationSettingsService;
private final NotificationExecutorService notificationExecutor;
private final DbCallbackExecutorService dbCallbackExecutorService;
private final NotificationsTopicService notificationsTopicService;
private final TbQueueProducerProvider producerProvider;
@ -124,74 +126,76 @@ public class DefaultNotificationCenter extends AbstractSubscriptionService imple
notificationRequest.setStatus(NotificationRequestStatus.PROCESSING);
NotificationRequest savedNotificationRequest = notificationRequestService.saveNotificationRequest(tenantId, notificationRequest);
NotificationProcessingContext ctx = NotificationProcessingContext.builder()
.tenantId(tenantId)
.request(savedNotificationRequest)
.settings(settings)
.template(notificationTemplate)
.build();
ctx.init();
Set<NotificationDeliveryMethod> deliveryMethods = ctx.getDeliveryMethods();
List<ListenableFuture<Void>> results = new ArrayList<>();
for (UUID targetId : notificationRequest.getTargets()) {
DaoUtil.processBatches(pageLink -> {
return notificationTargetService.findRecipientsForNotificationTarget(tenantId, ctx.getCustomerId(), new NotificationTargetId(targetId), pageLink);
}, 200, recipientsBatch -> {
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
if (deliveryMethod.isStandalone()) continue;
List<User> recipients = recipientsBatch.getData();
log.debug("Sending {} notifications for request {} to recipients batch ({})", deliveryMethod, savedNotificationRequest.getId(), recipients.size());
NotificationChannel notificationChannel = channels.get(deliveryMethod);
for (User recipient : recipients) {
ListenableFuture<Void> resultFuture = process(notificationChannel, recipient, ctx);
DonAsynchron.withCallback(resultFuture, result -> {
ctx.getStats().reportSent(deliveryMethod, recipient);
}, error -> {
ctx.getStats().reportError(deliveryMethod, error, recipient);
});
results.add(resultFuture);
notificationExecutor.submit(() -> {
NotificationProcessingContext ctx = NotificationProcessingContext.builder()
.tenantId(tenantId)
.request(savedNotificationRequest)
.settings(settings)
.template(notificationTemplate)
.build();
ctx.init();
Set<NotificationDeliveryMethod> deliveryMethods = ctx.getDeliveryMethods();
List<ListenableFuture<Void>> results = new ArrayList<>();
for (UUID targetId : notificationRequest.getTargets()) {
DaoUtil.processBatches(pageLink -> {
return notificationTargetService.findRecipientsForNotificationTarget(tenantId, ctx.getCustomerId(), new NotificationTargetId(targetId), pageLink);
}, 200, recipientsBatch -> {
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
if (deliveryMethod.isStandalone()) continue;
List<User> recipients = recipientsBatch.getData();
log.debug("Sending {} notifications for request {} to recipients batch ({})", deliveryMethod, savedNotificationRequest.getId(), recipients.size());
NotificationChannel notificationChannel = channels.get(deliveryMethod);
for (User recipient : recipients) {
ListenableFuture<Void> resultFuture = process(notificationChannel, recipient, ctx);
DonAsynchron.withCallback(resultFuture, result -> {
ctx.getStats().reportSent(deliveryMethod, recipient);
}, error -> {
ctx.getStats().reportError(deliveryMethod, error, recipient);
});
results.add(resultFuture);
}
}
}
});
}
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
if (deliveryMethod.isStandalone()) {
NotificationChannel notificationChannel = channels.get(deliveryMethod);
ListenableFuture<Void> resultFuture = process(notificationChannel, null, ctx);
DonAsynchron.withCallback(resultFuture, result -> {
ctx.getStats().reportSent(deliveryMethod, null);
}, error -> {
ctx.getStats().reportError(deliveryMethod, error, null);
});
results.add(resultFuture);
}
}
Futures.whenAllComplete(results).run(() -> {
NotificationRequestStats stats = ctx.getStats();
try {
notificationRequestService.updateNotificationRequest(tenantId, savedNotificationRequest.getId(),
NotificationRequestStatus.SENT, stats);
} catch (Exception e) {
log.error("Failed to update stats for notification request {}", savedNotificationRequest.getId(), e);
for (NotificationDeliveryMethod deliveryMethod : deliveryMethods) {
if (deliveryMethod.isStandalone()) {
NotificationChannel notificationChannel = channels.get(deliveryMethod);
ListenableFuture<Void> resultFuture = process(notificationChannel, null, ctx);
DonAsynchron.withCallback(resultFuture, result -> {
ctx.getStats().reportSent(deliveryMethod, null);
}, error -> {
ctx.getStats().reportError(deliveryMethod, error, null);
});
results.add(resultFuture);
}
}
UserId senderId = notificationRequest.getSenderId();
if (senderId != null) {
if (stats.getErrors().isEmpty()) {
int sent = stats.getSent().values().stream().mapToInt(AtomicInteger::get).sum();
sendBasicNotification(tenantId, senderId, "Notifications sent",
"All notifications were successfully sent (" + sent + ")");
} else {
int failures = stats.getErrors().values().stream().mapToInt(Map::size).sum();
sendBasicNotification(tenantId, senderId, "Notification failure",
"Some notifications were not sent (" + failures + ")"); // TODO: 'Go to request' button
Futures.whenAllComplete(results).run(() -> {
NotificationRequestStats stats = ctx.getStats();
try {
notificationRequestService.updateNotificationRequest(tenantId, savedNotificationRequest.getId(),
NotificationRequestStatus.SENT, stats);
} catch (Exception e) {
log.error("Failed to update stats for notification request {}", savedNotificationRequest.getId(), e);
}
}
}, dbCallbackExecutorService);
UserId senderId = notificationRequest.getSenderId();
if (senderId != null) {
if (stats.getErrors().isEmpty()) {
int sent = stats.getSent().values().stream().mapToInt(AtomicInteger::get).sum();
sendBasicNotification(tenantId, senderId, "Notifications sent",
"All notifications were successfully sent (" + sent + ")");
} else {
int failures = stats.getErrors().values().stream().mapToInt(Map::size).sum();
sendBasicNotification(tenantId, senderId, "Notification failure",
"Some notifications were not sent (" + failures + ")"); // TODO: 'Go to request' button
}
}
}, dbCallbackExecutorService);
});
return savedNotificationRequest;
}

6
application/src/main/java/org/thingsboard/server/service/ws/notification/DefaultNotificationCommandsHandler.java
View File

@ -200,7 +200,11 @@ public class DefaultNotificationCommandsHandler implements NotificationCommandsH
}
case UPDATED: {
if (update.getUpdatedStatus() == NotificationStatus.READ) {
subscription.getUnreadCounter().decrementAndGet();
if (update.isAllNotifications()) {
fetchUnreadNotificationsCount(subscription);
} else {
subscription.getUnreadCounter().decrementAndGet();
}
sendUpdate(subscription.getSessionId(), subscription.createUpdate());
}
break;

1
dao/src/main/java/org/thingsboard/server/dao/model/ModelConstants.java
View File

@ -295,7 +295,6 @@ public class ModelConstants {
public static final String ALARM_PROPAGATE_TO_OWNER_PROPERTY = "propagate_to_owner";
public static final String ALARM_PROPAGATE_TO_TENANT_PROPERTY = "propagate_to_tenant";
public static final String ALARM_PROPAGATE_RELATION_TYPES = "propagate_relation_types";
public static final String ALARM_NOTIFICATION_RULE_ID = "notification_rule_id";
public static final String ALARM_BY_ID_VIEW_NAME = "alarm_by_id";

4
dao/src/main/java/org/thingsboard/server/dao/model/sql/AbstractAlarmEntity.java
View File

@ -48,7 +48,6 @@ import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ACK_TS_PROPE
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_CLEAR_TS_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_CUSTOMER_ID_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_END_TS_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_NOTIFICATION_RULE_ID;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ORIGINATOR_ID_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_ORIGINATOR_TYPE_PROPERTY;
import static org.thingsboard.server.dao.model.ModelConstants.ALARM_PROPAGATE_PROPERTY;
@ -118,9 +117,6 @@ public abstract class AbstractAlarmEntity<T extends Alarm> extends BaseSqlEntity
@Column(name = ALARM_PROPAGATE_RELATION_TYPES)
private String propagateRelationTypes;
@Column(name = ALARM_NOTIFICATION_RULE_ID)
private UUID notificationRuleId;
public AbstractAlarmEntity() {
super();
}

25
dao/src/main/java/org/thingsboard/server/dao/model/sql/NotificationRequestInfoEntity.java
View File

@ -19,11 +19,16 @@ import com.fasterxml.jackson.databind.JsonNode;
import lombok.Data;
import lombok.EqualsAndHashCode;
import org.thingsboard.server.common.data.notification.NotificationDeliveryMethod;
import org.thingsboard.server.common.data.notification.NotificationRequest;
import org.thingsboard.server.common.data.notification.NotificationRequestInfo;
import org.thingsboard.server.common.data.notification.template.NotificationTemplateConfig;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@Data
@ -41,11 +46,21 @@ public class NotificationRequestInfoEntity extends NotificationRequestEntity {
@Override
public NotificationRequestInfo toData() {
List<NotificationDeliveryMethod> deliveryMethods = fromJson(templateConfig, NotificationTemplateConfig.class)
.getDeliveryMethodsTemplates().entrySet().stream()
.filter(entry -> entry.getValue().isEnabled())
.map(Map.Entry::getKey).collect(Collectors.toList());
return new NotificationRequestInfo(super.toData(), templateName, deliveryMethods);
NotificationRequest request = super.toData();
List<NotificationDeliveryMethod> deliveryMethods;
if (templateConfig != null) {
deliveryMethods = fromJson(templateConfig, NotificationTemplateConfig.class)
.getDeliveryMethodsTemplates().entrySet().stream()
.filter(entry -> entry.getValue().isEnabled())
.map(Map.Entry::getKey).collect(Collectors.toList());
} else if (request.getStats() != null) {
Set<NotificationDeliveryMethod> methods = new HashSet<>(request.getStats().getSent().keySet());
methods.addAll(request.getStats().getErrors().keySet());
deliveryMethods = new ArrayList<>(methods);
} else {
deliveryMethods = Collections.emptyList();
}
return new NotificationRequestInfo(request, templateName, deliveryMethods);
}
}

5
dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRequestDao.java
View File

@ -15,6 +15,7 @@
*/
package org.thingsboard.server.dao.sql.notification;
import com.google.common.base.Strings;
import lombok.RequiredArgsConstructor;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Component;
@ -60,8 +61,8 @@ public class JpaNotificationRequestDao extends JpaAbstractDao<NotificationReques
@Override
public PageData<NotificationRequestInfo> findInfosByTenantIdAndOriginatorTypeAndPageLink(TenantId tenantId, EntityType originatorType, PageLink pageLink) {
return DaoUtil.pageToPageData(notificationRequestRepository.findInfosByTenantIdAndOriginatorEntityType(getId(tenantId, true),
originatorType, DaoUtil.toPageable(pageLink))).mapData(NotificationRequestInfoEntity::toData);
return DaoUtil.pageToPageData(notificationRequestRepository.findInfosByTenantIdAndOriginatorEntityTypeAndSearchText(getId(tenantId, true),
originatorType, Strings.nullToEmpty(pageLink.getTextSearch()), DaoUtil.toPageable(pageLink))).mapData(NotificationRequestInfoEntity::toData);
}
@Override

2
dao/src/main/java/org/thingsboard/server/dao/sql/notification/JpaNotificationRuleDao.java
View File

@ -46,7 +46,7 @@ public class JpaNotificationRuleDao extends JpaAbstractDao<NotificationRuleEntit
@Override
public PageData<NotificationRule> findByTenantIdAndPageLink(TenantId tenantId, PageLink pageLink) {
return DaoUtil.toPageData(notificationRuleRepository.findByTenantIdAndNameContainingIgnoreCase(getId(tenantId, true),
return DaoUtil.toPageData(notificationRuleRepository.findByTenantIdAndSearchText(getId(tenantId, true),
Strings.nullToEmpty(pageLink.getTextSearch()), DaoUtil.toPageable(pageLink)));
}

12
dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRequestRepository.java
View File

@ -36,14 +36,16 @@ import java.util.UUID;
public interface NotificationRequestRepository extends JpaRepository<NotificationRequestEntity, UUID> {
String REQUEST_INFO_QUERY = "SELECT new org.thingsboard.server.dao.model.sql.NotificationRequestInfoEntity(r, t.name, t.configuration) " +
"FROM NotificationRequestEntity r INNER JOIN NotificationTemplateEntity t ON r.templateId = t.id";
"FROM NotificationRequestEntity r LEFT JOIN NotificationTemplateEntity t ON r.templateId = t.id";
Page<NotificationRequestEntity> findByTenantIdAndOriginatorEntityType(UUID tenantId, EntityType originatorType, Pageable pageable);
@Query(REQUEST_INFO_QUERY + " WHERE r.tenantId = :tenantId AND r.originatorEntityType = :originatorType")
Page<NotificationRequestInfoEntity> findInfosByTenantIdAndOriginatorEntityType(@Param("tenantId") UUID tenantId,
@Param("originatorType") EntityType originatorType,
Pageable pageable);
@Query(REQUEST_INFO_QUERY + " WHERE r.tenantId = :tenantId AND r.originatorEntityType = :originatorType " +
"AND (:searchText = '' OR (t.name IS NOT NULL AND lower(t.name) LIKE lower(concat('%', :searchText, '%'))))")
Page<NotificationRequestInfoEntity> findInfosByTenantIdAndOriginatorEntityTypeAndSearchText(@Param("tenantId") UUID tenantId,
@Param("originatorType") EntityType originatorType,
@Param("searchText") String searchText,
Pageable pageable);
@Query(REQUEST_INFO_QUERY + " WHERE r.id = :id")
NotificationRequestInfoEntity findInfoById(@Param("id") UUID id);

8
dao/src/main/java/org/thingsboard/server/dao/sql/notification/NotificationRuleRepository.java
View File

@ -18,6 +18,8 @@ package org.thingsboard.server.dao.sql.notification;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.data.repository.query.Param;
import org.springframework.stereotype.Repository;
import org.thingsboard.server.common.data.notification.rule.trigger.NotificationRuleTriggerType;
import org.thingsboard.server.dao.model.sql.NotificationRuleEntity;
@ -28,7 +30,11 @@ import java.util.UUID;
@Repository
public interface NotificationRuleRepository extends JpaRepository<NotificationRuleEntity, UUID> {
Page<NotificationRuleEntity> findByTenantIdAndNameContainingIgnoreCase(UUID tenantId, String searchText, Pageable pageable);
@Query("SELECT r FROM NotificationRuleEntity r WHERE r.tenantId = :tenantId " +
"AND lower(r.name) LIKE lower(concat('%', :searchText, '%')) ")
Page<NotificationRuleEntity> findByTenantIdAndSearchText(@Param("tenantId") UUID tenantId,
@Param("searchText") String searchText,
Pageable pageable);
boolean existsByRecipientsConfigContaining(String string);

1
dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultAlarmQueryRepository.java
View File

@ -104,7 +104,6 @@ public class DefaultAlarmQueryRepository implements AlarmQueryRepository {
" a.tenant_id as tenant_id, " +
" a.customer_id as customer_id, " +
" a.propagate_relation_types as propagate_relation_types, " +
" a.notification_rule_id as notification_rule_id, " +
" a.type as type," + SELECT_ORIGINATOR_NAME + ", ";
private static final String JOIN_ENTITY_ALARMS = "inner join entity_alarm ea on a.id = ea.alarm_id";

5
dao/src/main/resources/sql/schema-entities.sql
View File

@ -58,8 +58,7 @@ CREATE TABLE IF NOT EXISTS alarm (
propagate_relation_types varchar,
type varchar(255),
propagate_to_owner boolean,
propagate_to_tenant boolean,
notification_rule_id uuid
propagate_to_tenant boolean
);
CREATE TABLE IF NOT EXISTS entity_alarm (
@ -834,7 +833,7 @@ CREATE TABLE IF NOT EXISTS notification (
type VARCHAR(32) NOT NULL,
subject VARCHAR(255),
text VARCHAR(1000) NOT NULL,
additional_config VARCHAR(500),
additional_config VARCHAR(1000),
info VARCHAR(1000),
status VARCHAR(32)
) PARTITION BY RANGE (created_time);

Write Preview
Loading…
Cancel
Save