|
|
|
@ -29,8 +29,12 @@ import org.thingsboard.server.common.data.id.TenantId; |
|
|
|
import org.thingsboard.server.common.data.id.TenantProfileId; |
|
|
|
import org.thingsboard.server.common.data.page.PageData; |
|
|
|
import org.thingsboard.server.common.data.page.PageLink; |
|
|
|
import org.thingsboard.server.common.data.queue.ProcessingStrategy; |
|
|
|
import org.thingsboard.server.common.data.queue.SubmitStrategy; |
|
|
|
import org.thingsboard.server.common.data.queue.SubmitStrategyType; |
|
|
|
import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; |
|
|
|
import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; |
|
|
|
import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; |
|
|
|
import org.thingsboard.server.dao.entity.AbstractEntityService; |
|
|
|
import org.thingsboard.server.dao.exception.DataValidationException; |
|
|
|
import org.thingsboard.server.dao.service.DataValidator; |
|
|
|
@ -39,6 +43,10 @@ import org.thingsboard.server.dao.service.Validator; |
|
|
|
|
|
|
|
import java.util.Arrays; |
|
|
|
import java.util.Collections; |
|
|
|
import java.util.HashSet; |
|
|
|
import java.util.List; |
|
|
|
import java.util.Optional; |
|
|
|
import java.util.Set; |
|
|
|
|
|
|
|
import static org.thingsboard.server.common.data.CacheConstants.TENANT_PROFILE_CACHE; |
|
|
|
import static org.thingsboard.server.dao.service.Validator.validateId; |
|
|
|
@ -228,6 +236,35 @@ public class TenantProfileServiceImpl extends AbstractEntityService implements T |
|
|
|
throw new DataValidationException("Another default tenant profile is present!"); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
if (tenantProfile.isIsolatedTbRuleEngine()) { |
|
|
|
List<TenantProfileQueueConfiguration> queueConfiguration = tenantProfile.getProfileData().getQueueConfiguration(); |
|
|
|
if (queueConfiguration == null) { |
|
|
|
throw new DataValidationException("Tenant profile data queue configuration should be specified!"); |
|
|
|
} |
|
|
|
|
|
|
|
Optional<TenantProfileQueueConfiguration> mainQueueConfig = |
|
|
|
queueConfiguration |
|
|
|
.stream() |
|
|
|
.filter(q -> q.getName().equals("Main")) |
|
|
|
.findAny(); |
|
|
|
if (mainQueueConfig.isEmpty()) { |
|
|
|
throw new DataValidationException("Main queue configuration should be specified!"); |
|
|
|
} |
|
|
|
|
|
|
|
queueConfiguration.forEach(this::validateQueueConfiguration); |
|
|
|
|
|
|
|
Set<String> queueNames = new HashSet<>(queueConfiguration.size()); |
|
|
|
|
|
|
|
queueConfiguration.forEach(q -> { |
|
|
|
String name = q.getName(); |
|
|
|
if (queueNames.contains(name)) { |
|
|
|
throw new DataValidationException(String.format("Queue configuration name '%s' already present!", name)); |
|
|
|
} else { |
|
|
|
queueNames.add(name); |
|
|
|
} |
|
|
|
}); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
@Override |
|
|
|
@ -241,6 +278,54 @@ public class TenantProfileServiceImpl extends AbstractEntityService implements T |
|
|
|
throw new DataValidationException("Can't update isolatedTbCore property!"); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void validateQueueConfiguration(TenantProfileQueueConfiguration queue) { |
|
|
|
if (StringUtils.isEmpty(queue.getName())) { |
|
|
|
throw new DataValidationException("Queue name should be specified!"); |
|
|
|
} |
|
|
|
if (StringUtils.isBlank(queue.getTopic())) { |
|
|
|
throw new DataValidationException("Queue topic should be non empty and without spaces!"); |
|
|
|
} |
|
|
|
if (queue.getPollInterval() < 1) { |
|
|
|
throw new DataValidationException("Queue poll interval should be more then 0!"); |
|
|
|
} |
|
|
|
if (queue.getPartitions() < 1) { |
|
|
|
throw new DataValidationException("Queue partitions should be more then 0!"); |
|
|
|
} |
|
|
|
if (queue.getPackProcessingTimeout() < 1) { |
|
|
|
throw new DataValidationException("Queue pack processing timeout should be more then 0!"); |
|
|
|
} |
|
|
|
|
|
|
|
SubmitStrategy submitStrategy = queue.getSubmitStrategy(); |
|
|
|
if (submitStrategy == null) { |
|
|
|
throw new DataValidationException("Queue submit strategy can't be null!"); |
|
|
|
} |
|
|
|
if (submitStrategy.getType() == null) { |
|
|
|
throw new DataValidationException("Queue submit strategy type can't be null!"); |
|
|
|
} |
|
|
|
if (submitStrategy.getType() == SubmitStrategyType.BATCH && submitStrategy.getBatchSize() < 1) { |
|
|
|
throw new DataValidationException("Queue submit strategy batch size should be more then 0!"); |
|
|
|
} |
|
|
|
ProcessingStrategy processingStrategy = queue.getProcessingStrategy(); |
|
|
|
if (processingStrategy == null) { |
|
|
|
throw new DataValidationException("Queue processing strategy can't be null!"); |
|
|
|
} |
|
|
|
if (processingStrategy.getType() == null) { |
|
|
|
throw new DataValidationException("Queue processing strategy type can't be null!"); |
|
|
|
} |
|
|
|
if (processingStrategy.getRetries() < 0) { |
|
|
|
throw new DataValidationException("Queue processing strategy retries can't be less then 0!"); |
|
|
|
} |
|
|
|
if (processingStrategy.getFailurePercentage() < 0 || processingStrategy.getFailurePercentage() > 100) { |
|
|
|
throw new DataValidationException("Queue processing strategy failure percentage should be in a range from 0 to 100!"); |
|
|
|
} |
|
|
|
if (processingStrategy.getPauseBetweenRetries() < 0) { |
|
|
|
throw new DataValidationException("Queue processing strategy pause between retries can't be less then 0!"); |
|
|
|
} |
|
|
|
if (processingStrategy.getMaxPauseBetweenRetries() < processingStrategy.getPauseBetweenRetries()) { |
|
|
|
throw new DataValidationException("Queue processing strategy MAX pause between retries can't be less then pause between retries!"); |
|
|
|
} |
|
|
|
} |
|
|
|
}; |
|
|
|
|
|
|
|
private PaginatedRemover<String, TenantProfile> tenantProfilesRemover = |
|
|
|
|
YevhenBondarenko
5 years ago