YevhenBondarenko
5 years ago
49 changed files with 1801 additions and 39 deletions
@ -0,0 +1,48 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.queue; |
|||
|
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface QueueService { |
|||
|
|||
Queue saveQueue(Queue queue); |
|||
|
|||
void deleteQueue(TenantId tenantId, QueueId queueId); |
|||
|
|||
List<Queue> findQueuesByTenantId(TenantId tenantId); |
|||
|
|||
PageData<Queue> findQueuesByTenantId(TenantId tenantId, PageLink pageLink); |
|||
|
|||
List<Queue> findAllMainQueues(); |
|||
|
|||
List<Queue> findAllQueues(); |
|||
|
|||
Queue findQueueById(TenantId tenantId, QueueId queueId); |
|||
|
|||
Queue findQueueByTenantIdAndName(TenantId tenantId, String name); |
|||
|
|||
Queue createDefaultMainQueue(TenantProfile tenantProfile, TenantId tenantId); |
|||
|
|||
void deleteQueuesByTenantId(TenantId tenantId); |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.id; |
|||
|
|||
import com.fasterxml.jackson.annotation.JsonCreator; |
|||
import com.fasterxml.jackson.annotation.JsonIgnore; |
|||
import com.fasterxml.jackson.annotation.JsonProperty; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
public class QueueId extends UUIDBased implements EntityId { |
|||
|
|||
private static final long serialVersionUID = 1L; |
|||
|
|||
@JsonCreator |
|||
public QueueId(@JsonProperty("id") UUID id) { |
|||
super(id); |
|||
} |
|||
|
|||
public static QueueId fromString(String queueId) { |
|||
return new QueueId(UUID.fromString(queueId)); |
|||
} |
|||
|
|||
@JsonIgnore |
|||
@Override |
|||
public EntityType getEntityType() { |
|||
return EntityType.QUEUE; |
|||
} |
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.queue; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class ProcessingStrategy { |
|||
private ProcessingStrategyType type; |
|||
private int retries; |
|||
private double failurePercentage; |
|||
private long pauseBetweenRetries; |
|||
private long maxPauseBetweenRetries; |
|||
} |
|||
@ -0,0 +1,36 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
|
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.queue; |
|||
|
|||
public enum ProcessingStrategyType { |
|||
SKIP_ALL_FAILURES, SKIP_ALL_FAILURES_AND_TIMED_OUT, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT |
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.queue; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.BaseData; |
|||
import org.thingsboard.server.common.data.HasName; |
|||
import org.thingsboard.server.common.data.HasTenantId; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
@Data |
|||
public class Queue extends BaseData<QueueId> implements HasName, HasTenantId { |
|||
private TenantId tenantId; |
|||
private String name; |
|||
private String topic; |
|||
private int pollInterval; |
|||
private int partitions; |
|||
private long packProcessingTimeout; |
|||
private SubmitStrategy submitStrategy; |
|||
private ProcessingStrategy processingStrategy; |
|||
|
|||
public Queue() { |
|||
} |
|||
|
|||
public Queue(QueueId id) { |
|||
super(id); |
|||
} |
|||
} |
|||
@ -0,0 +1,24 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.queue; |
|||
|
|||
import lombok.Data; |
|||
|
|||
@Data |
|||
public class SubmitStrategy { |
|||
private SubmitStrategyType type; |
|||
private int batchSize; |
|||
} |
|||
@ -0,0 +1,20 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.common.data.queue; |
|||
|
|||
public enum SubmitStrategyType { |
|||
BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL |
|||
} |
|||
@ -0,0 +1,105 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.model.sql; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import lombok.Data; |
|||
import lombok.EqualsAndHashCode; |
|||
import org.hibernate.annotations.Type; |
|||
import org.hibernate.annotations.TypeDef; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.queue.ProcessingStrategy; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategy; |
|||
import org.thingsboard.server.dao.DaoUtil; |
|||
import org.thingsboard.server.dao.model.BaseSqlEntity; |
|||
import org.thingsboard.server.dao.model.ModelConstants; |
|||
import org.thingsboard.server.dao.util.mapping.JsonStringType; |
|||
|
|||
import javax.persistence.Column; |
|||
import javax.persistence.Entity; |
|||
import javax.persistence.Table; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@Entity |
|||
@TypeDef(name = "json", typeClass = JsonStringType.class) |
|||
@Table(name = ModelConstants.QUEUE_COLUMN_FAMILY_NAME) |
|||
public class QueueEntity extends BaseSqlEntity<Queue> { |
|||
|
|||
private static final ObjectMapper mapper = new ObjectMapper(); |
|||
|
|||
@Column(name = ModelConstants.QUEUE_TENANT_ID_PROPERTY) |
|||
private UUID tenantId; |
|||
|
|||
@Column(name = ModelConstants.QUEUE_NAME_PROPERTY) |
|||
private String name; |
|||
|
|||
@Column(name = ModelConstants.QUEUE_TOPIC_PROPERTY) |
|||
private String topic; |
|||
@Column(name = ModelConstants.QUEUE_POLL_INTERVAL_PROPERTY) |
|||
private int pollInterval; |
|||
|
|||
@Column(name = ModelConstants.QUEUE_PARTITIONS_PROPERTY) |
|||
private int partitions; |
|||
|
|||
@Column(name = ModelConstants.QUEUE_PACK_PROCESSING_TIMEOUT_PROPERTY) |
|||
private long packProcessingTimeout; |
|||
|
|||
@Type(type = "json") |
|||
@Column(name = ModelConstants.QUEUE_SUBMIT_STRATEGY_PROPERTY) |
|||
private JsonNode submitStrategy; |
|||
|
|||
@Type(type = "json") |
|||
@Column(name = ModelConstants.QUEUE_PROCESSING_STRATEGY_PROPERTY) |
|||
private JsonNode processingStrategy; |
|||
|
|||
public QueueEntity() { |
|||
} |
|||
|
|||
public QueueEntity(Queue queue) { |
|||
if (queue.getId() != null) { |
|||
this.setId(queue.getId().getId()); |
|||
} |
|||
this.createdTime = queue.getCreatedTime(); |
|||
this.tenantId = DaoUtil.getId(queue.getTenantId()); |
|||
this.name = queue.getName(); |
|||
this.topic = queue.getTopic(); |
|||
this.pollInterval = queue.getPollInterval(); |
|||
this.partitions = queue.getPartitions(); |
|||
this.packProcessingTimeout = queue.getPackProcessingTimeout(); |
|||
this.submitStrategy = mapper.valueToTree(queue.getSubmitStrategy()); |
|||
this.processingStrategy = mapper.valueToTree(queue.getProcessingStrategy()); |
|||
} |
|||
|
|||
@Override |
|||
public Queue toData() { |
|||
Queue queue = new Queue(new QueueId(getUuid())); |
|||
queue.setCreatedTime(createdTime); |
|||
queue.setTenantId(new TenantId(tenantId)); |
|||
queue.setName(name); |
|||
queue.setTopic(topic); |
|||
queue.setPollInterval(pollInterval); |
|||
queue.setPartitions(partitions); |
|||
queue.setPackProcessingTimeout(packProcessingTimeout); |
|||
queue.setSubmitStrategy(mapper.convertValue(this.submitStrategy, SubmitStrategy.class)); |
|||
queue.setProcessingStrategy(mapper.convertValue(this.processingStrategy, ProcessingStrategy.class)); |
|||
return queue; |
|||
} |
|||
} |
|||
@ -0,0 +1,333 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.queue; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.commons.lang3.StringUtils; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.stereotype.Service; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
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.ProcessingStrategyType; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategy; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategyType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.dao.entity.AbstractEntityService; |
|||
import org.thingsboard.server.dao.exception.DataValidationException; |
|||
import org.thingsboard.server.dao.service.DataValidator; |
|||
import org.thingsboard.server.dao.service.PaginatedRemover; |
|||
import org.thingsboard.server.dao.service.Validator; |
|||
import org.thingsboard.server.dao.tenant.TbTenantProfileCache; |
|||
import org.thingsboard.server.dao.tenant.TenantDao; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
public class BaseQueueService extends AbstractEntityService implements QueueService { |
|||
|
|||
@Autowired |
|||
private QueueDao queueDao; |
|||
|
|||
@Autowired |
|||
private TenantDao tenantDao; |
|||
|
|||
@Autowired |
|||
private TbTenantProfileCache tenantProfileCache; |
|||
|
|||
@Autowired(required = false) |
|||
private TbQueueAdmin tbQueueAdmin; |
|||
|
|||
// @Autowired(required = false)
|
|||
// private TbQueueClusterService queueClusterService;
|
|||
|
|||
// @Autowired
|
|||
// private QueueStatsService queueStatsService;
|
|||
|
|||
@Override |
|||
@Transactional |
|||
public Queue saveQueue(Queue queue) { |
|||
log.trace("Executing createOrUpdateQueue [{}]", queue); |
|||
queueValidator.validate(queue, Queue::getTenantId); |
|||
Queue savedQueue; |
|||
if (queue.getId() == null) { |
|||
savedQueue = createQueue(queue); |
|||
} else { |
|||
savedQueue = updateQueue(queue); |
|||
} |
|||
|
|||
// if (queueClusterService != null) {
|
|||
// queueClusterService.onQueueChange(savedQueue, null);
|
|||
// }
|
|||
return savedQueue; |
|||
} |
|||
|
|||
private Queue createQueue(Queue queue) { |
|||
Queue createdQueue = queueDao.save(queue.getTenantId(), queue); |
|||
if (tbQueueAdmin != null) { |
|||
for (int i = 0; i < queue.getPartitions(); i++) { |
|||
tbQueueAdmin.createTopicIfNotExists(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); |
|||
} |
|||
} |
|||
return createdQueue; |
|||
} |
|||
|
|||
private Queue updateQueue(Queue queue) { |
|||
Queue oldQueue = queueDao.findById(queue.getTenantId(), queue.getUuidId()); |
|||
Queue updatedQueue = queueDao.save(queue.getTenantId(), queue); |
|||
|
|||
int oldPartitions = oldQueue.getPartitions(); |
|||
int currentPartitions = queue.getPartitions(); |
|||
|
|||
//TODO: 3.2 remove if partitions can't be deleted.
|
|||
// if (currentPartitions != oldPartitions && tbQueueAdmin != null) {
|
|||
// queueClusterService.onQueueDelete(queue, null);
|
|||
// if (currentPartitions > oldPartitions) {
|
|||
// log.info("Added [{}] new partitions to [{}] queue", currentPartitions - oldPartitions, queue.getName());
|
|||
// for (int i = oldPartitions; i < currentPartitions; i++) {
|
|||
// tbQueueAdmin.createTopicIfNotExists(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName());
|
|||
// }
|
|||
// } else {
|
|||
// log.info("Removed [{}] partitions from [{}] queue", oldPartitions - currentPartitions, queue.getName());
|
|||
// for (int i = currentPartitions; i < oldPartitions; i++) {
|
|||
// tbQueueAdmin.deleteTopic(new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName());
|
|||
// }
|
|||
// }
|
|||
// }
|
|||
|
|||
return updatedQueue; |
|||
} |
|||
|
|||
@Override |
|||
@Transactional |
|||
public void deleteQueue(TenantId tenantId, QueueId queueId) { |
|||
log.trace("Executing deleteQueue, queueId: [{}]", queueId); |
|||
Queue queue = findQueueById(tenantId, queueId); |
|||
// if (queueClusterService != null) {
|
|||
// queueClusterService.onQueueDelete(queue, null);
|
|||
// }
|
|||
// queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId);
|
|||
boolean result = queueDao.removeById(tenantId, queueId.getId()); |
|||
if (result && tbQueueAdmin != null) { |
|||
for (int i = 0; i < queue.getPartitions(); i++) { |
|||
String fullTopicName = new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(); |
|||
log.debug("Deleting queue [{}]", fullTopicName); |
|||
try { |
|||
tbQueueAdmin.deleteTopic(fullTopicName); |
|||
} catch (Exception e) { |
|||
log.error("Failed to delete queue [{}]", fullTopicName); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findQueuesByTenantId(TenantId tenantId) { |
|||
log.trace("Executing findQueues, tenantId: [{}]", tenantId); |
|||
return queueDao.findAllByTenantId(getSystemOrIsolatedTenantId(tenantId)); |
|||
} |
|||
|
|||
@Override |
|||
public PageData<Queue> findQueuesByTenantId(TenantId tenantId, PageLink pageLink) { |
|||
log.trace("Executing findQueues pageLink [{}]", pageLink); |
|||
Validator.validatePageLink(pageLink); |
|||
return queueDao.findQueuesByTenantId(getSystemOrIsolatedTenantId(tenantId), pageLink); |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findAllMainQueues() { |
|||
log.trace("Executing findAllMainQueues"); |
|||
return queueDao.findAllMainQueues(); |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findAllQueues() { |
|||
log.trace("Executing findAllQueues"); |
|||
return queueDao.findAllQueues(); |
|||
} |
|||
|
|||
@Override |
|||
public Queue findQueueById(TenantId tenantId, QueueId queueId) { |
|||
log.trace("Executing findQueueById, queueId: [{}]", queueId); |
|||
return queueDao.findById(tenantId, queueId.getId()); |
|||
} |
|||
|
|||
@Override |
|||
public Queue findQueueByTenantIdAndName(TenantId tenantId, String queueName) { |
|||
log.trace("Executing findQueueByTenantIdAndName, tenantId: [{}] queueName: [{}]", tenantId, queueName); |
|||
return queueDao.findQueueByTenantIdAndName(getSystemOrIsolatedTenantId(tenantId), queueName); |
|||
} |
|||
|
|||
@Override |
|||
public void deleteQueuesByTenantId(TenantId tenantId) { |
|||
Validator.validateId(tenantId, "Incorrect tenant id for delete queues request."); |
|||
tenantQueuesRemover.removeEntities(tenantId, tenantId); |
|||
} |
|||
|
|||
@Override |
|||
@Transactional |
|||
public Queue createDefaultMainQueue(TenantProfile tenantProfile, TenantId tenantId) { |
|||
Queue mainQueue = new Queue(); |
|||
mainQueue.setTenantId(tenantId); |
|||
mainQueue.setName("Main"); |
|||
mainQueue.setTopic("tb_rule_engine.main"); |
|||
mainQueue.setPollInterval(25); |
|||
mainQueue.setPartitions(Math.max(tenantProfile.getMaxNumberOfPartitionsPerQueue(), 1)); |
|||
mainQueue.setPackProcessingTimeout(60000); |
|||
SubmitStrategy mainQueueSubmitStrategy = new SubmitStrategy(); |
|||
mainQueueSubmitStrategy.setType(SubmitStrategyType.BURST); |
|||
mainQueueSubmitStrategy.setBatchSize(1000); |
|||
mainQueue.setSubmitStrategy(mainQueueSubmitStrategy); |
|||
ProcessingStrategy mainQueueProcessingStrategy = new ProcessingStrategy(); |
|||
mainQueueProcessingStrategy.setType(ProcessingStrategyType.SKIP_ALL_FAILURES); |
|||
mainQueueProcessingStrategy.setRetries(3); |
|||
mainQueueProcessingStrategy.setFailurePercentage(0); |
|||
mainQueueProcessingStrategy.setPauseBetweenRetries(3); |
|||
mainQueueProcessingStrategy.setMaxPauseBetweenRetries(3); |
|||
mainQueue.setProcessingStrategy(mainQueueProcessingStrategy); |
|||
return saveQueue(mainQueue); |
|||
} |
|||
|
|||
private DataValidator<Queue> queueValidator = |
|||
new DataValidator<>() { |
|||
|
|||
@Override |
|||
protected void validateCreate(TenantId tenantId, Queue queue) { |
|||
if (queueDao.findQueueByTenantIdAndTopic(tenantId, queue.getTopic()) != null) { |
|||
throw new DataValidationException(String.format("Queue with topic: %s already exists!", queue.getTopic())); |
|||
} |
|||
if (queueDao.findQueueByTenantIdAndName(tenantId, queue.getName()) != null) { |
|||
throw new DataValidationException(String.format("Queue with name: %s already exists!", queue.getName())); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected void validateUpdate(TenantId tenantId, Queue queue) { |
|||
Queue foundQueue = queueDao.findById(tenantId, queue.getUuidId()); |
|||
if (queueDao.findById(tenantId, queue.getUuidId()) == null) { |
|||
throw new DataValidationException(String.format("Queue with id: %s does not exists!", queue.getId())); |
|||
} |
|||
if (!foundQueue.getName().equals(queue.getName())) { |
|||
throw new DataValidationException("Queue name can't be changed!"); |
|||
} |
|||
if (!foundQueue.getTopic().equals(queue.getTopic())) { |
|||
throw new DataValidationException("Queue topic can't be changed!"); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
protected void validateDataImpl(TenantId tenantId, Queue queue) { |
|||
if (!tenantId.equals(TenantId.SYS_TENANT_ID)) { |
|||
TenantProfile tenantProfile = tenantProfileCache.get(tenantId); |
|||
|
|||
if (!tenantProfile.isIsolatedTbRuleEngine()) { |
|||
throw new DataValidationException("Tenant should be isolated!"); |
|||
} |
|||
|
|||
if (queue.getId() == null) { |
|||
List<Queue> existingQueues = findQueuesByTenantId(tenantId); |
|||
if (existingQueues.size() >= tenantProfile.getMaxNumberOfQueues()) { |
|||
throw new DataValidationException("The limit for creating new queue has been exceeded!"); |
|||
} |
|||
} |
|||
|
|||
if (queue.getPartitions() > tenantProfile.getMaxNumberOfPartitionsPerQueue()) { |
|||
throw new DataValidationException(String.format("Queue partitions can't be more then %d", tenantProfile.getMaxNumberOfPartitionsPerQueue())); |
|||
} |
|||
} |
|||
|
|||
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<TenantId, Queue> tenantQueuesRemover = |
|||
new PaginatedRemover<>() { |
|||
|
|||
@Override |
|||
protected PageData<Queue> findEntities(TenantId tenantId, TenantId id, PageLink pageLink) { |
|||
return queueDao.findQueuesByTenantId(id, pageLink); |
|||
} |
|||
|
|||
@Override |
|||
protected void removeEntity(TenantId tenantId, Queue entity) { |
|||
deleteQueue(tenantId, entity.getId()); |
|||
} |
|||
}; |
|||
|
|||
private TenantId getSystemOrIsolatedTenantId(TenantId tenantId) { |
|||
if (!tenantId.equals(TenantId.SYS_TENANT_ID)) { |
|||
TenantProfile tenantProfile = tenantProfileCache.get(tenantId); |
|||
if (tenantProfile.isIsolatedTbRuleEngine()) { |
|||
return tenantId; |
|||
} |
|||
} |
|||
|
|||
return TenantId.SYS_TENANT_ID; |
|||
} |
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.queue; |
|||
|
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.dao.Dao; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface QueueDao extends Dao<Queue> { |
|||
Queue findQueueByTenantIdAndTopic(TenantId tenantId, String topic); |
|||
|
|||
Queue findQueueByTenantIdAndName(TenantId tenantId, String name); |
|||
|
|||
List<Queue> findAllMainQueues(); |
|||
|
|||
List<Queue> findAllQueues(); |
|||
|
|||
List<Queue> findAllByTenantId(TenantId tenantId); |
|||
|
|||
PageData<Queue> findQueuesByTenantId(TenantId tenantId, PageLink pageLink); |
|||
} |
|||
@ -0,0 +1,86 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sql.queue; |
|||
|
|||
import com.google.common.collect.Lists; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.repository.CrudRepository; |
|||
import org.springframework.stereotype.Component; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.page.PageData; |
|||
import org.thingsboard.server.common.data.page.PageLink; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.dao.DaoUtil; |
|||
import org.thingsboard.server.dao.model.sql.QueueEntity; |
|||
import org.thingsboard.server.dao.queue.QueueDao; |
|||
import org.thingsboard.server.dao.sql.JpaAbstractDao; |
|||
|
|||
import java.util.List; |
|||
import java.util.Objects; |
|||
import java.util.UUID; |
|||
|
|||
@Slf4j |
|||
@Component |
|||
public class JpaQueueDao extends JpaAbstractDao<QueueEntity, Queue> implements QueueDao { |
|||
|
|||
@Autowired |
|||
private QueueRepository queueRepository; |
|||
|
|||
@Override |
|||
protected Class<QueueEntity> getEntityClass() { |
|||
return QueueEntity.class; |
|||
} |
|||
|
|||
@Override |
|||
protected CrudRepository<QueueEntity, UUID> getCrudRepository() { |
|||
return queueRepository; |
|||
} |
|||
|
|||
@Override |
|||
public Queue findQueueByTenantIdAndTopic(TenantId tenantId, String topic) { |
|||
return DaoUtil.getData(queueRepository.findByTenantIdAndTopic(tenantId.getId(), topic)); |
|||
} |
|||
|
|||
@Override |
|||
public Queue findQueueByTenantIdAndName(TenantId tenantId, String name) { |
|||
return DaoUtil.getData(queueRepository.findByTenantIdAndName(tenantId.getId(), name)); |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findAllByTenantId(TenantId tenantId) { |
|||
List<QueueEntity> entities = queueRepository.findByTenantId(tenantId.getId()); |
|||
return DaoUtil.convertDataList(entities); |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findAllMainQueues() { |
|||
List<QueueEntity> entities = Lists.newArrayList(queueRepository.findAllByName("Main")); |
|||
return DaoUtil.convertDataList(entities); |
|||
} |
|||
|
|||
@Override |
|||
public List<Queue> findAllQueues() { |
|||
List<QueueEntity> entities = Lists.newArrayList(queueRepository.findAll()); |
|||
return DaoUtil.convertDataList(entities); |
|||
} |
|||
|
|||
@Override |
|||
public PageData<Queue> findQueuesByTenantId(TenantId tenantId, PageLink pageLink) { |
|||
return DaoUtil.toPageData(queueRepository |
|||
.findByTenantId(tenantId.getId(), Objects.toString(pageLink.getTextSearch(), ""), DaoUtil.toPageable(pageLink))); |
|||
} |
|||
} |
|||
@ -0,0 +1,42 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sql.queue; |
|||
|
|||
import org.springframework.data.domain.Page; |
|||
import org.springframework.data.domain.Pageable; |
|||
import org.springframework.data.jpa.repository.Query; |
|||
import org.springframework.data.repository.CrudRepository; |
|||
import org.springframework.data.repository.query.Param; |
|||
import org.thingsboard.server.dao.model.sql.QueueEntity; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
|
|||
public interface QueueRepository extends CrudRepository<QueueEntity, UUID> { |
|||
QueueEntity findByTenantIdAndTopic(UUID tenantId, String topic); |
|||
|
|||
QueueEntity findByTenantIdAndName(UUID tenantId, String name); |
|||
|
|||
List<QueueEntity> findByTenantId(UUID tenantId); |
|||
|
|||
@Query("SELECT q FROM QueueEntity q WHERE q.tenantId = :tenantId " + |
|||
"AND LOWER(q.name) LIKE LOWER(CONCAT(:textSearch, '%'))") |
|||
Page<QueueEntity> findByTenantId(@Param("tenantId") UUID tenantId, |
|||
@Param("textSearch") String textSearch, |
|||
Pageable pageable); |
|||
|
|||
List<QueueEntity> findAllByName(String name); |
|||
} |
|||
@ -0,0 +1,182 @@ |
|||
<!-- |
|||
|
|||
Copyright © 2016-2022 The Thingsboard Authors |
|||
|
|||
Licensed under the Apache License, Version 2.0 (the "License"); |
|||
you may not use this file except in compliance with the License. |
|||
You may obtain a copy of the License at |
|||
|
|||
http://www.apache.org/licenses/LICENSE-2.0 |
|||
|
|||
Unless required by applicable law or agreed to in writing, software |
|||
distributed under the License is distributed on an "AS IS" BASIS, |
|||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
See the License for the specific language governing permissions and |
|||
limitations under the License. |
|||
|
|||
--> |
|||
<div class="tb-details-buttons" fxLayout.xs="column"> |
|||
<button mat-raised-button color="primary" fxFlex.xs |
|||
[disabled]="(isLoading$ | async)" |
|||
(click)="onEntityAction($event, 'delete')" |
|||
[fxShow]="!hideDelete() && !isEdit"> |
|||
{{'queue.delete' | translate }} |
|||
</button> |
|||
</div> |
|||
|
|||
<div class="mat-padding" fxLayout="column"> |
|||
<form [formGroup]="entityForm"> |
|||
<fieldset [disabled]="(isLoading$ | async) || !isEdit"> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>admin.queue-name</mat-label> |
|||
<input matInput formControlName="name" required> |
|||
<mat-error *ngIf="entityForm.get('name').hasError('required')"> |
|||
{{ 'queue.name-required' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.poll-interval</mat-label> |
|||
<input type="number" matInput formControlName="pollInterval" required> |
|||
<mat-error *ngIf="entityForm.get('pollInterval').hasError('required')"> |
|||
{{ 'queue.poll-interval-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('pollInterval').hasError('min') && |
|||
!entityForm.get('pollInterval').hasError('required')"> |
|||
{{ 'queue.poll-interval-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.partitions</mat-label> |
|||
<input type="number" matInput formControlName="partitions" required> |
|||
<mat-error *ngIf="entityForm.get('partitions').hasError('required')"> |
|||
{{ 'queue.partitions-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('partitions').hasError('min') && |
|||
!entityForm.get('partitions').hasError('required')"> |
|||
{{ 'queue.partitions-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.processing-timeout</mat-label> |
|||
<input type="number" matInput formControlName="packProcessingTimeout" required> |
|||
<mat-error *ngIf="entityForm.get('packProcessingTimeout').hasError('required')"> |
|||
{{ 'queue.pack-processing-timeout-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('packProcessingTimeout').hasError('min') && |
|||
!entityForm.get('packProcessingTimeout').hasError('required')"> |
|||
{{ 'queue.pack-processing-timeout-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
|
|||
<div class="mat-accordion-container"> |
|||
|
|||
<mat-accordion [multi]="true"> |
|||
<mat-expansion-panel #panel1 hideToggle> |
|||
<mat-expansion-panel-header> |
|||
<mat-panel-title> |
|||
<mat-label translate>queue.submit-strategy</mat-label> |
|||
<mat-icon>{{panel1.expanded ? 'keyboard_arrow_up' : 'keyboard_arrow_down' }}</mat-icon> |
|||
</mat-panel-title> |
|||
</mat-expansion-panel-header> |
|||
<div formGroupName="submitStrategy"> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.submit-strategy</mat-label> |
|||
<mat-select formControlName="type" required> |
|||
<mat-option *ngFor="let strategy of submitStrategies" [value]="strategy"> |
|||
{{ strategy }} |
|||
</mat-option> |
|||
</mat-select> |
|||
<mat-error *ngIf="entityForm.get('submitStrategy.type').hasError('required')"> |
|||
{{ 'queue.submit-strategy-type-required' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block" *ngIf="hideBatchSize"> |
|||
<mat-label translate>queue.batch-size</mat-label> |
|||
<input type="number" matInput formControlName="batchSize" required> |
|||
<mat-error *ngIf="entityForm.get('submitStrategy.batchSize').hasError('required')"> |
|||
{{ 'queue.batch-size-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('submitStrategy.batchSize').hasError('min') && |
|||
!entityForm.get('submitStrategy.batchSize').hasError('required')"> |
|||
{{ 'queue.batch-size-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
</div> |
|||
</mat-expansion-panel> |
|||
<mat-expansion-panel #panel2 hideToggle> |
|||
<mat-expansion-panel-header> |
|||
<mat-panel-title> |
|||
<mat-label translate>queue.processing-strategy</mat-label> |
|||
<mat-icon>{{panel2.expanded ? 'keyboard_arrow_up' : 'keyboard_arrow_down' }}</mat-icon> |
|||
</mat-panel-title> |
|||
</mat-expansion-panel-header> |
|||
|
|||
<div formGroupName="processingStrategy"> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.processing-strategy</mat-label> |
|||
<mat-select formControlName="type" required> |
|||
<mat-option *ngFor="let strategy of processingStrategies" [value]="strategy"> |
|||
{{ strategy }} |
|||
</mat-option> |
|||
</mat-select> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.type').hasError('required')"> |
|||
{{ 'queue.processing-strategy-type-required' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.retries</mat-label> |
|||
<input type="number" matInput formControlName="retries" required> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.retries').hasError('required')"> |
|||
{{ 'queue.retries-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.retries').hasError('min') && |
|||
!entityForm.get('processingStrategy.retries').hasError('required')"> |
|||
{{ 'queue.retries-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.failure-percentage</mat-label> |
|||
<input type="number" matInput formControlName="failurePercentage" required> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.failurePercentage').hasError('required')"> |
|||
{{ 'queue.failure-percentage-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.failurePercentage').hasError('min') && |
|||
!entityForm.get('processingStrategy.failurePercentage').hasError('required') && |
|||
!entityForm.get('processingStrategy.failurePercentage').hasError('max')"> |
|||
{{ 'queue.failure-percentage-min-value' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.failurePercentage').hasError('max') && |
|||
!entityForm.get('processingStrategy.failurePercentage').hasError('required') && |
|||
!entityForm.get('processingStrategy.failurePercentage').hasError('min')"> |
|||
{{ 'queue.failure-percentage-max-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.pause-between-retries</mat-label> |
|||
<input type="number" matInput formControlName="pauseBetweenRetries" required> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.pauseBetweenRetries').hasError('required')"> |
|||
{{ 'queue.pause-between-retries-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.pauseBetweenRetries').hasError('min') && |
|||
!entityForm.get('processingStrategy.pauseBetweenRetries').hasError('required')"> |
|||
{{ 'queue.pause-between-retries-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
<mat-form-field class="mat-block"> |
|||
<mat-label translate>queue.max-pause-between-retries</mat-label> |
|||
<input type="number" matInput formControlName="maxPauseBetweenRetries" required> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.maxPauseBetweenRetries').hasError('required')"> |
|||
{{ 'queue.max-pause-between-retries-required' | translate }} |
|||
</mat-error> |
|||
<mat-error *ngIf="entityForm.get('processingStrategy.maxPauseBetweenRetries').hasError('min') && |
|||
!entityForm.get('processingStrategy.maxPauseBetweenRetries').hasError('required')"> |
|||
{{ 'queue.max-pause-between-retries-min-value' | translate }} |
|||
</mat-error> |
|||
</mat-form-field> |
|||
</div> |
|||
</mat-expansion-panel> |
|||
</mat-accordion> |
|||
</div> |
|||
</fieldset> |
|||
</form> |
|||
</div> |
|||
@ -0,0 +1,59 @@ |
|||
/** |
|||
* Copyright © 2016-2022 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0 |
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
:host ::ng-deep { |
|||
|
|||
.mat-expansion-panel:not([class*='mat-elevation-z']) { |
|||
box-shadow: none; |
|||
} |
|||
|
|||
.mat-accordion-container { |
|||
margin-bottom: 16px; |
|||
} |
|||
|
|||
.mat-expansion-panel { |
|||
|
|||
&:hover { |
|||
box-shadow: 0 3px 14px 2px rgba(0, 0, 0, 0.12), 0 8px 10px 1px rgba(0, 0, 0, 0.14), 0 5px 5px -3px rgba(0, 0, 0, 0.2); |
|||
} |
|||
|
|||
&-header { |
|||
height: 56px; |
|||
border: 1px solid #f2f2f2; |
|||
|
|||
&-title { |
|||
align-items: center; |
|||
justify-content: space-between; |
|||
margin-right: 0; |
|||
} |
|||
} |
|||
|
|||
&.mat-expanded { |
|||
border: none; |
|||
box-shadow: 0 3px 14px 2px rgba(0, 0, 0, 0.12), 0 8px 10px 1px rgba(0, 0, 0, 0.14), 0 5px 5px -3px rgba(0, 0, 0, 0.2); |
|||
|
|||
.mat-expansion-panel { |
|||
|
|||
&-content { |
|||
margin-top: 20px; |
|||
} |
|||
|
|||
&-body { |
|||
padding-bottom: 10px; |
|||
} |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,154 @@ |
|||
///
|
|||
/// Copyright © 2016-2022 The Thingsboard Authors
|
|||
///
|
|||
/// Licensed under the Apache License, Version 2.0 (the "License");
|
|||
/// you may not use this file except in compliance with the License.
|
|||
/// You may obtain a copy of the License at
|
|||
///
|
|||
/// http://www.apache.org/licenses/LICENSE-2.0
|
|||
///
|
|||
/// Unless required by applicable law or agreed to in writing, software
|
|||
/// distributed under the License is distributed on an "AS IS" BASIS,
|
|||
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|||
/// See the License for the specific language governing permissions and
|
|||
/// limitations under the License.
|
|||
///
|
|||
|
|||
import { ChangeDetectorRef, Component, Inject } from '@angular/core'; |
|||
import { EntityType } from '@shared/models/entity-type.models'; |
|||
import { FormBuilder, FormGroup, Validators } from '@angular/forms'; |
|||
import { EntityComponent } from '@home/components/entity/entity.component'; |
|||
import { QueueInfo, QueueProcessingStrategyTypes, QueueSubmitStrategyTypes } from '@shared/models/queue.models'; |
|||
import { Store } from '@ngrx/store'; |
|||
import { AppState } from '@core/core.state'; |
|||
import { TranslateService } from '@ngx-translate/core'; |
|||
import { EntityTableConfig } from '@home/models/entity/entities-table-config.models'; |
|||
import set = Reflect.set; |
|||
import { distinctUntilChanged } from 'rxjs/operators'; |
|||
|
|||
@Component({ |
|||
selector: 'tb-queue', |
|||
templateUrl: './queue.component.html', |
|||
styleUrls: ['./queue.component.scss'] |
|||
}) |
|||
export class QueueComponent extends EntityComponent<QueueInfo> { |
|||
entityForm: FormGroup; |
|||
|
|||
entityType = EntityType; |
|||
submitStrategies: string[] = []; |
|||
processingStrategies: string[] = []; |
|||
|
|||
QueueSubmitStrategyTypes = QueueSubmitStrategyTypes; |
|||
hideBatchSize: boolean = false; |
|||
|
|||
constructor(protected store: Store<AppState>, |
|||
protected translate: TranslateService, |
|||
@Inject('entity') protected entityValue: QueueInfo, |
|||
@Inject('entitiesTableConfig') protected entitiesTableConfigValue: EntityTableConfig<QueueInfo>, |
|||
protected cd: ChangeDetectorRef, |
|||
public fb: FormBuilder) { |
|||
super(store, fb, entityValue, entitiesTableConfigValue, cd); |
|||
this.submitStrategies = Object.values(QueueSubmitStrategyTypes); |
|||
this.processingStrategies = Object.values(QueueProcessingStrategyTypes); |
|||
} |
|||
|
|||
ngOnInit() { |
|||
super.ngOnInit(); |
|||
this.entityForm.get('submitStrategy').get('type').valueChanges.subscribe(() => { |
|||
this.submitStrategyTypeChanged(); |
|||
}); |
|||
} |
|||
|
|||
buildForm(entity: QueueInfo): FormGroup { |
|||
return this.fb.group( |
|||
{ |
|||
name: [entity ? entity.name : '', [Validators.required]], |
|||
pollInterval: [ |
|||
entity && entity.pollInterval ? entity.pollInterval : 25, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
partitions: [ |
|||
entity && entity.partitions ? entity.partitions : 10, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
packProcessingTimeout: [ |
|||
entity && entity.packProcessingTimeout ? entity.packProcessingTimeout : 2000, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
submitStrategy: this.fb.group({ |
|||
type: [entity ? entity.submitStrategy?.type : null, [Validators.required]], |
|||
batchSize: [ |
|||
entity && entity.submitStrategy?.batchSize ? entity.submitStrategy?.batchSize : 1000, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
}), |
|||
processingStrategy: this.fb.group({ |
|||
type: [entity ? entity.processingStrategy?.type : null, [Validators.required]], |
|||
retries: [ |
|||
entity && entity.processingStrategy?.retries ? entity.processingStrategy?.retries : 3, |
|||
[Validators.min(0), Validators.required] |
|||
], |
|||
failurePercentage: [ |
|||
entity && entity.processingStrategy?.failurePercentage ? entity.processingStrategy?.failurePercentage : 0, |
|||
[Validators.min(0), Validators.required, Validators.max(100)] |
|||
], |
|||
pauseBetweenRetries: [ |
|||
entity && entity.processingStrategy?.pauseBetweenRetries ? entity.processingStrategy?.pauseBetweenRetries : 3, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
maxPauseBetweenRetries: [ |
|||
entity && entity.processingStrategy?.maxPauseBetweenRetries ? entity.processingStrategy?.maxPauseBetweenRetries : 3, |
|||
[Validators.min(1), Validators.required] |
|||
], |
|||
}) |
|||
} |
|||
); |
|||
} |
|||
|
|||
hideDelete() { |
|||
if (this.entitiesTableConfig) { |
|||
return !this.entitiesTableConfig.deleteEnabled(this.entity); |
|||
} else { |
|||
return false; |
|||
} |
|||
} |
|||
|
|||
updateForm(entity: QueueInfo) { |
|||
this.entityForm.patchValue({ |
|||
name: entity.name, |
|||
pollInterval: entity.pollInterval, |
|||
partitions: entity.partitions, |
|||
packProcessingTimeout: entity.packProcessingTimeout, |
|||
submitStrategy: { |
|||
type: entity.submitStrategy?.type, |
|||
batchSize: entity.submitStrategy?.batchSize, |
|||
}, |
|||
processingStrategy: { |
|||
type: entity.processingStrategy?.type, |
|||
retries: entity.processingStrategy?.retries, |
|||
failurePercentage: entity.processingStrategy?.failurePercentage, |
|||
pauseBetweenRetries: entity.processingStrategy?.pauseBetweenRetries, |
|||
maxPauseBetweenRetries: entity.processingStrategy?.maxPauseBetweenRetries, |
|||
} |
|||
}, {emitEvent: true}); |
|||
|
|||
if (!this.isAdd) { |
|||
this.entityForm.get('name').disable({emitEvent: false}); |
|||
} |
|||
} |
|||
|
|||
submitStrategyTypeChanged() { |
|||
const form = this.entityForm.get("submitStrategy") as FormGroup; |
|||
const type: QueueSubmitStrategyTypes = form.get('type').value; |
|||
const batchSizeField = form.get('batchSize'); |
|||
if (type === QueueSubmitStrategyTypes.BATCH) { |
|||
batchSizeField.enable(); |
|||
batchSizeField.patchValue(1000); |
|||
this.hideBatchSize = true; |
|||
} else { |
|||
batchSizeField.patchValue(null); |
|||
batchSizeField.disable(); |
|||
this.hideBatchSize = false; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,115 @@ |
|||
///
|
|||
/// Copyright © 2016-2022 The Thingsboard Authors
|
|||
///
|
|||
/// Licensed under the Apache License, Version 2.0 (the "License");
|
|||
/// you may not use this file except in compliance with the License.
|
|||
/// You may obtain a copy of the License at
|
|||
///
|
|||
/// http://www.apache.org/licenses/LICENSE-2.0
|
|||
///
|
|||
/// Unless required by applicable law or agreed to in writing, software
|
|||
/// distributed under the License is distributed on an "AS IS" BASIS,
|
|||
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|||
/// See the License for the specific language governing permissions and
|
|||
/// limitations under the License.
|
|||
///
|
|||
|
|||
import { Injectable } from '@angular/core'; |
|||
import { ActivatedRouteSnapshot, Resolve } from '@angular/router'; |
|||
import { |
|||
EntityTableColumn, |
|||
EntityTableConfig |
|||
} from '@home/models/entity/entities-table-config.models'; |
|||
import { QueueInfo, ServiceType } from '@shared/models/queue.models'; |
|||
import { select, Store } from '@ngrx/store'; |
|||
import { AppState } from '@core/core.state'; |
|||
import { BroadcastService } from '@core/services/broadcast.service'; |
|||
import { CustomerService } from '@core/http/customer.service'; |
|||
import { DialogService } from '@core/services/dialog.service'; |
|||
import { HomeDialogsService } from '@home/dialogs/home-dialogs.service'; |
|||
import { map, mergeMap, take } from 'rxjs/operators'; |
|||
import { Observable } from 'rxjs'; |
|||
import { EntityType, entityTypeResources, entityTypeTranslations } from '@app/shared/models/entity-type.models'; |
|||
import { TranslateService } from '@ngx-translate/core'; |
|||
import { QueueComponent } from './queue.component'; |
|||
import { QueueService } from '@core/http/queue.service'; |
|||
import { selectAuthUser } from '@core/auth/auth.selectors'; |
|||
|
|||
@Injectable() |
|||
export class QueuesTableConfigResolver implements Resolve<EntityTableConfig<QueueInfo>> { |
|||
|
|||
readonly queueType = ServiceType.TB_RULE_ENGINE; |
|||
|
|||
private readonly config: EntityTableConfig<QueueInfo> = new EntityTableConfig<QueueInfo>(); |
|||
|
|||
constructor(private store: Store<AppState>, |
|||
private broadcast: BroadcastService, |
|||
private queueService: QueueService, |
|||
private customerService: CustomerService, |
|||
private dialogService: DialogService, |
|||
private homeDialogs: HomeDialogsService, |
|||
private translate: TranslateService) { |
|||
|
|||
this.config.entityType = EntityType.QUEUE; |
|||
this.config.entityComponent = QueueComponent; |
|||
this.config.entityTranslations = entityTypeTranslations.get(EntityType.QUEUE); |
|||
this.config.entityResources = entityTypeResources.get(EntityType.QUEUE); |
|||
|
|||
this.config.deleteEntityTitle = queue => this.translate.instant('queue.delete-queue-title', {queueName: queue.name}); |
|||
this.config.deleteEntityContent = () => this.translate.instant('queue.delete-queue-text'); |
|||
this.config.deleteEntitiesTitle = count => this.translate.instant('queue.delete-queues-title', {count}); |
|||
this.config.deleteEntitiesContent = () => this.translate.instant('queue.delete-queues-text'); |
|||
} |
|||
|
|||
resolve(route: ActivatedRouteSnapshot): Observable<EntityTableConfig<QueueInfo>> { |
|||
this.config.componentsData = { |
|||
queueType: this.queueType |
|||
}; |
|||
|
|||
return this.store.pipe(select(selectAuthUser), take(1)).pipe( |
|||
map(() => { |
|||
this.config.tableTitle = this.translate.instant('admin.queues'); |
|||
this.config.columns = this.configureColumns(); |
|||
this.configureEntityFunctions(); |
|||
return this.config; |
|||
}) |
|||
); |
|||
} |
|||
|
|||
configureColumns(): Array<EntityTableColumn<QueueInfo>> { |
|||
return [ |
|||
new EntityTableColumn<QueueInfo>('name', 'admin.queue-name', '25%'), |
|||
new EntityTableColumn<QueueInfo>('partitions', 'admin.queue-partitions', '25%'), |
|||
new EntityTableColumn<QueueInfo>('submitStrategy', 'admin.queue-submit-strategy', '25%', |
|||
(entity: QueueInfo) => { |
|||
return entity.submitStrategy.type; |
|||
}, |
|||
() => ({}), |
|||
false |
|||
), |
|||
new EntityTableColumn<QueueInfo>('processingStrategy', 'admin.queue-processing-strategy', '25%', |
|||
(entity: QueueInfo) => { |
|||
return entity.processingStrategy.type; |
|||
}, |
|||
() => ({}), |
|||
false |
|||
) |
|||
]; |
|||
} |
|||
|
|||
configureEntityFunctions(): void { |
|||
this.config.entitiesFetchFunction = pageLink => this.queueService.getTenantQueuesByServiceType(pageLink, this.queueType); |
|||
this.config.loadEntity = id => this.queueService.getQueueById(id.id); |
|||
this.config.saveEntity = queue => this.queueService.saveQueue(this.addTopicForQueue(queue), this.queueType).pipe( |
|||
mergeMap((savedQueue) => this.queueService.getQueueById(savedQueue.id.id) |
|||
)); |
|||
this.config.deleteEntity = id => this.queueService.deleteQueue(id.id); |
|||
this.config.deleteEnabled = (queue) => queue && queue.name !== 'Main'; |
|||
} |
|||
|
|||
private addTopicForQueue(queue: QueueInfo): QueueInfo { |
|||
const modifiedQueue = Object.assign({}, queue); |
|||
modifiedQueue.topic = `tb_rule_engine.${queue.name}`; |
|||
return modifiedQueue; |
|||
} |
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
///
|
|||
/// Copyright © 2016-2022 The Thingsboard Authors
|
|||
///
|
|||
/// Licensed under the Apache License, Version 2.0 (the "License");
|
|||
/// you may not use this file except in compliance with the License.
|
|||
/// You may obtain a copy of the License at
|
|||
///
|
|||
/// http://www.apache.org/licenses/LICENSE-2.0
|
|||
///
|
|||
/// Unless required by applicable law or agreed to in writing, software
|
|||
/// distributed under the License is distributed on an "AS IS" BASIS,
|
|||
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|||
/// See the License for the specific language governing permissions and
|
|||
/// limitations under the License.
|
|||
///
|
|||
|
|||
import { EntityId } from './entity-id'; |
|||
import { EntityType } from '@shared/models/entity-type.models'; |
|||
|
|||
export class QueueId implements EntityId { |
|||
entityType = EntityType.QUEUE; |
|||
id: string; |
|||
constructor(id: string) { |
|||
this.id = id; |
|||
} |
|||
} |
|||
Loading…
Reference in new issue