Andrew Shvayka
4 years ago
committed by
GitHub
GitHub
210 changed files with 6058 additions and 1598 deletions
@ -0,0 +1,49 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
ALTER TABLE device_profile |
|||
ADD COLUMN IF NOT EXISTS default_queue_id uuid; |
|||
|
|||
DO |
|||
$$ |
|||
BEGIN |
|||
IF EXISTS |
|||
(SELECT column_name |
|||
FROM information_schema.columns |
|||
WHERE table_name = 'device_profile' |
|||
AND column_name = 'default_queue_name' |
|||
) |
|||
THEN |
|||
UPDATE device_profile |
|||
SET default_queue_id = q.id |
|||
FROM queue as q |
|||
WHERE default_queue_name = q.name; |
|||
END IF; |
|||
END |
|||
$$; |
|||
|
|||
DO |
|||
$$ |
|||
BEGIN |
|||
IF NOT EXISTS(SELECT 1 FROM pg_constraint WHERE conname = 'fk_default_queue_device_profile') THEN |
|||
ALTER TABLE device_profile |
|||
ADD CONSTRAINT fk_default_queue_device_profile FOREIGN KEY (default_queue_id) REFERENCES queue (id); |
|||
END IF; |
|||
END; |
|||
$$; |
|||
|
|||
ALTER TABLE device_profile |
|||
DROP COLUMN IF EXISTS default_queue_name; |
|||
@ -0,0 +1,269 @@ |
|||
/** |
|||
* 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.service.entitiy.queue; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.cluster.TbClusterService; |
|||
import org.thingsboard.server.common.data.DeviceProfile; |
|||
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.PageLink; |
|||
import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.dao.device.DeviceProfileService; |
|||
import org.thingsboard.server.queue.TbQueueAdmin; |
|||
import org.thingsboard.server.queue.scheduler.SchedulerComponent; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.entitiy.AbstractTbEntityService; |
|||
|
|||
import java.util.ArrayList; |
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@TbCoreComponent |
|||
@AllArgsConstructor |
|||
public class DefaultTbQueueService extends AbstractTbEntityService implements TbQueueService { |
|||
private static final String MAIN = "Main"; |
|||
private static final long DELETE_DELAY = 30; |
|||
|
|||
private final TbClusterService tbClusterService; |
|||
private final TbQueueAdmin tbQueueAdmin; |
|||
private final DeviceProfileService deviceProfileService; |
|||
private final SchedulerComponent scheduler; |
|||
|
|||
@Override |
|||
public Queue saveQueue(Queue queue) { |
|||
boolean create = queue.getId() == null; |
|||
Queue oldQueue; |
|||
|
|||
if (create) { |
|||
oldQueue = null; |
|||
} else { |
|||
oldQueue = queueService.findQueueById(queue.getTenantId(), queue.getId()); |
|||
} |
|||
|
|||
//TODO: add checkNotNull
|
|||
Queue savedQueue = queueService.saveQueue(queue); |
|||
|
|||
if (create) { |
|||
onQueueCreated(savedQueue); |
|||
} else { |
|||
onQueueUpdated(savedQueue, oldQueue); |
|||
} |
|||
|
|||
return savedQueue; |
|||
} |
|||
|
|||
@Override |
|||
public void deleteQueue(TenantId tenantId, QueueId queueId) { |
|||
Queue queue = queueService.findQueueById(tenantId, queueId); |
|||
queueService.deleteQueue(tenantId, queueId); |
|||
onQueueDeleted(queue); |
|||
} |
|||
|
|||
@Override |
|||
public void deleteQueueByQueueName(TenantId tenantId, String queueName) { |
|||
Queue queue = queueService.findQueueByTenantIdAndNameInternal(tenantId, queueName); |
|||
queueService.deleteQueue(tenantId, queue.getId()); |
|||
onQueueDeleted(queue); |
|||
} |
|||
|
|||
private void onQueueCreated(Queue queue) { |
|||
for (int i = 0; i < queue.getPartitions(); i++) { |
|||
tbQueueAdmin.createTopicIfNotExists( |
|||
new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName()); |
|||
} |
|||
|
|||
tbClusterService.onQueueChange(queue); |
|||
} |
|||
|
|||
private void onQueueUpdated(Queue queue, Queue oldQueue) { |
|||
int oldPartitions = oldQueue.getPartitions(); |
|||
int currentPartitions = queue.getPartitions(); |
|||
|
|||
if (currentPartitions != oldPartitions) { |
|||
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()); |
|||
} |
|||
tbClusterService.onQueueChange(queue); |
|||
} else { |
|||
log.info("Removed [{}] partitions from [{}] queue", oldPartitions - currentPartitions, queue.getName()); |
|||
tbClusterService.onQueueChange(queue); |
|||
|
|||
scheduler.schedule(() -> { |
|||
for (int i = currentPartitions; i < oldPartitions; i++) { |
|||
String fullTopicName = new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(); |
|||
log.info("Removed partition [{}]", fullTopicName); |
|||
tbQueueAdmin.deleteTopic( |
|||
fullTopicName); |
|||
} |
|||
}, DELETE_DELAY, TimeUnit.SECONDS); |
|||
} |
|||
} else if (!oldQueue.equals(queue)) { |
|||
tbClusterService.onQueueChange(queue); |
|||
} |
|||
} |
|||
|
|||
private void onQueueDeleted(Queue queue) { |
|||
tbClusterService.onQueueDelete(queue); |
|||
|
|||
// queueStatsService.deleteQueueStatsByQueueId(tenantId, queueId);
|
|||
|
|||
scheduler.schedule(() -> { |
|||
for (int i = 0; i < queue.getPartitions(); i++) { |
|||
String fullTopicName = new TopicPartitionInfo(queue.getTopic(), queue.getTenantId(), i, false).getFullTopicName(); |
|||
log.info("Deleting queue [{}]", fullTopicName); |
|||
try { |
|||
tbQueueAdmin.deleteTopic(fullTopicName); |
|||
} catch (Exception e) { |
|||
log.error("Failed to delete queue [{}]", fullTopicName); |
|||
} |
|||
} |
|||
}, DELETE_DELAY, TimeUnit.SECONDS); |
|||
} |
|||
|
|||
@Override |
|||
public void updateQueuesByTenants(List<TenantId> tenantIds, TenantProfile newTenantProfile, TenantProfile |
|||
oldTenantProfile) { |
|||
boolean oldIsolated = oldTenantProfile != null && oldTenantProfile.isIsolatedTbRuleEngine(); |
|||
boolean newIsolated = newTenantProfile.isIsolatedTbRuleEngine(); |
|||
|
|||
if (!oldIsolated && !newIsolated) { |
|||
return; |
|||
} |
|||
|
|||
if (newTenantProfile.equals(oldTenantProfile)) { |
|||
return; |
|||
} |
|||
|
|||
Map<String, TenantProfileQueueConfiguration> oldQueues; |
|||
Map<String, TenantProfileQueueConfiguration> newQueues; |
|||
|
|||
if (oldIsolated) { |
|||
oldQueues = oldTenantProfile.getProfileData().getQueueConfiguration().stream() |
|||
.collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); |
|||
} else { |
|||
oldQueues = Collections.emptyMap(); |
|||
} |
|||
|
|||
if (newIsolated) { |
|||
newQueues = newTenantProfile.getProfileData().getQueueConfiguration().stream() |
|||
.collect(Collectors.toMap(TenantProfileQueueConfiguration::getName, q -> q)); |
|||
} else { |
|||
newQueues = Collections.emptyMap(); |
|||
} |
|||
|
|||
List<String> toRemove = new ArrayList<>(); |
|||
List<String> toCreate = new ArrayList<>(); |
|||
List<String> toUpdate = new ArrayList<>(); |
|||
|
|||
for (String oldQueue : oldQueues.keySet()) { |
|||
if (!newQueues.containsKey(oldQueue)) { |
|||
toRemove.add(oldQueue); |
|||
} |
|||
} |
|||
|
|||
for (String newQueue : newQueues.keySet()) { |
|||
if (oldQueues.containsKey(newQueue)) { |
|||
toUpdate.add(newQueue); |
|||
} else { |
|||
toCreate.add(newQueue); |
|||
} |
|||
} |
|||
|
|||
tenantIds.forEach(tenantId -> { |
|||
Map<QueueId, List<DeviceProfile>> deviceProfileQueues; |
|||
|
|||
if (oldTenantProfile != null && !newTenantProfile.getId().equals(oldTenantProfile.getId()) || !toRemove.isEmpty()) { |
|||
List<DeviceProfile> deviceProfiles = deviceProfileService.findDeviceProfiles(tenantId, new PageLink(Integer.MAX_VALUE)).getData(); |
|||
deviceProfileQueues = deviceProfiles.stream() |
|||
.filter(dp -> dp.getDefaultQueueId() != null) |
|||
.collect(Collectors.groupingBy(DeviceProfile::getDefaultQueueId)); |
|||
} else { |
|||
deviceProfileQueues = Collections.emptyMap(); |
|||
} |
|||
|
|||
Map<String, QueueId> createdQueues = toCreate.stream() |
|||
.map(key -> saveQueue(new Queue(tenantId, newQueues.get(key)))) |
|||
.collect(Collectors.toMap(Queue::getName, Queue::getId)); |
|||
|
|||
// assigning created queues to device profiles instead of system queues
|
|||
if (oldTenantProfile != null && !oldTenantProfile.isIsolatedTbRuleEngine()) { |
|||
deviceProfileQueues.forEach((queueId, list) -> { |
|||
Queue queue = queueService.findQueueById(TenantId.SYS_TENANT_ID, queueId); |
|||
QueueId queueIdToAssign = createdQueues.get(queue.getName()); |
|||
if (queueIdToAssign == null) { |
|||
queueIdToAssign = createdQueues.get(MAIN); |
|||
} |
|||
for (DeviceProfile deviceProfile : list) { |
|||
deviceProfile.setDefaultQueueId(queueIdToAssign); |
|||
saveDeviceProfile(deviceProfile); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
toUpdate.forEach(key -> { |
|||
Queue queueToUpdate = new Queue(tenantId, newQueues.get(key)); |
|||
Queue foundQueue = queueService.findQueueByTenantIdAndName(tenantId, key); |
|||
queueToUpdate.setId(foundQueue.getId()); |
|||
queueToUpdate.setCreatedTime(foundQueue.getCreatedTime()); |
|||
|
|||
if (queueToUpdate.equals(foundQueue)) { |
|||
//Queue not changed
|
|||
} else { |
|||
saveQueue(queueToUpdate); |
|||
} |
|||
}); |
|||
|
|||
toRemove.forEach(q -> { |
|||
Queue queue = queueService.findQueueByTenantIdAndNameInternal(tenantId, q); |
|||
QueueId queueIdForRemove = queue.getId(); |
|||
if (deviceProfileQueues.containsKey(queueIdForRemove)) { |
|||
Queue foundQueue = queueService.findQueueByTenantIdAndName(tenantId, q); |
|||
if (foundQueue == null || queue.equals(foundQueue)) { |
|||
foundQueue = queueService.findQueueByTenantIdAndName(tenantId, MAIN); |
|||
} |
|||
QueueId newQueueId = foundQueue.getId(); |
|||
deviceProfileQueues.get(queueIdForRemove).stream() |
|||
.peek(dp -> dp.setDefaultQueueId(newQueueId)) |
|||
.forEach(this::saveDeviceProfile); |
|||
} |
|||
deleteQueue(tenantId, queueIdForRemove); |
|||
}); |
|||
}); |
|||
} |
|||
|
|||
//TODO: remove after implementing TbDeviceProfileService
|
|||
private void saveDeviceProfile(DeviceProfile deviceProfile) { |
|||
DeviceProfile savedDeviceProfile = deviceProfileService.saveDeviceProfile(deviceProfile); |
|||
tbClusterService.onDeviceProfileChange(savedDeviceProfile, null); |
|||
tbClusterService.broadcastEntityStateChangeEvent(deviceProfile.getTenantId(), savedDeviceProfile.getId(), ComponentLifecycleEvent.UPDATED); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* 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.service.entitiy.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.queue.Queue; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface TbQueueService { |
|||
|
|||
Queue saveQueue(Queue queue); |
|||
|
|||
void deleteQueue(TenantId tenantId, QueueId queueId); |
|||
|
|||
void deleteQueueByQueueName(TenantId tenantId, String queueName); |
|||
|
|||
void updateQueuesByTenants(List<TenantId> tenantIds, TenantProfile newTenantProfile, TenantProfile oldTenantProfile); |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* 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.service.entitiy.tenant_profile; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.dao.tenant.TenantProfileService; |
|||
import org.thingsboard.server.dao.tenant.TenantService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
import org.thingsboard.server.service.entitiy.queue.TbQueueService; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@TbCoreComponent |
|||
@AllArgsConstructor |
|||
public class DefaultTbTenantProfileService implements TbTenantProfileService { |
|||
private final TbQueueService tbQueueService; |
|||
private final TenantProfileService tenantProfileService; |
|||
private final TenantService tenantService; |
|||
|
|||
@Override |
|||
public TenantProfile saveTenantProfile(TenantId tenantId, TenantProfile tenantProfile, TenantProfile oldTenantProfile) { |
|||
TenantProfile savedTenantProfile = tenantProfileService.saveTenantProfile(tenantId, tenantProfile); |
|||
|
|||
if (oldTenantProfile != null && savedTenantProfile.isIsolatedTbRuleEngine()) { |
|||
List<TenantId> tenantIds = tenantService.findTenantIdsByTenantProfileId(savedTenantProfile.getId()); |
|||
tbQueueService.updateQueuesByTenants(tenantIds, savedTenantProfile, oldTenantProfile); |
|||
} |
|||
|
|||
return savedTenantProfile; |
|||
} |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* 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.service.entitiy.tenant_profile; |
|||
|
|||
import org.thingsboard.server.common.data.TenantProfile; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
public interface TbTenantProfileService { |
|||
TenantProfile saveTenantProfile(TenantId tenantId, TenantProfile tenantProfile, TenantProfile oldTenantProfile); |
|||
} |
|||
@ -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.service.install; |
|||
|
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration; |
|||
import org.springframework.boot.context.properties.ConfigurationProperties; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.springframework.context.annotation.Profile; |
|||
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.List; |
|||
|
|||
@Slf4j |
|||
@Data |
|||
@EnableAutoConfiguration |
|||
@Configuration |
|||
@ConfigurationProperties(prefix = "queue.rule-engine") |
|||
@Profile("install") |
|||
public class TbRuleEngineQueueConfigService { |
|||
|
|||
private String topic; |
|||
private List<TbRuleEngineQueueConfiguration> queues; |
|||
|
|||
@PostConstruct |
|||
public void validate() { |
|||
queues.stream().filter(queue -> queue.getName().equals("Main")).findFirst().orElseThrow(() -> { |
|||
log.error("Main queue is not configured in thingsboard.yml"); |
|||
return new RuntimeException("No \"Main\" queue configured!"); |
|||
}); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* 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.service.queue; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.dao.queue.QueueService; |
|||
import org.thingsboard.server.queue.discovery.QueueRoutingInfo; |
|||
import org.thingsboard.server.queue.discovery.QueueRoutingInfoService; |
|||
|
|||
import java.util.List; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Slf4j |
|||
@Service |
|||
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core' || '${service.type:null}'=='tb-rule-engine'") |
|||
public class DefaultQueueRoutingInfoService implements QueueRoutingInfoService { |
|||
|
|||
private final QueueService queueService; |
|||
|
|||
public DefaultQueueRoutingInfoService(QueueService queueService) { |
|||
this.queueService = queueService; |
|||
} |
|||
|
|||
@Override |
|||
public List<QueueRoutingInfo> getAllQueuesRoutingInfo() { |
|||
return queueService.findAllQueues().stream().map(QueueRoutingInfo::new).collect(Collectors.toList()); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* 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.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> findAllQueues(); |
|||
|
|||
Queue findQueueById(TenantId tenantId, QueueId queueId); |
|||
|
|||
Queue findQueueByTenantIdAndName(TenantId tenantId, String name); |
|||
|
|||
Queue findQueueByTenantIdAndNameInternal(TenantId tenantId, String queueName); |
|||
|
|||
void deleteQueuesByTenantId(TenantId tenantId); |
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
/** |
|||
* 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.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)); |
|||
} |
|||
|
|||
@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,62 @@ |
|||
/** |
|||
* 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.HasName; |
|||
import org.thingsboard.server.common.data.HasTenantId; |
|||
import org.thingsboard.server.common.data.SearchTextBasedWithAdditionalInfo; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.tenant.profile.TenantProfileQueueConfiguration; |
|||
|
|||
@Data |
|||
public class Queue extends SearchTextBasedWithAdditionalInfo<QueueId> implements HasName, HasTenantId { |
|||
private TenantId tenantId; |
|||
private String name; |
|||
private String topic; |
|||
private int pollInterval; |
|||
private int partitions; |
|||
private boolean consumerPerPartition; |
|||
private long packProcessingTimeout; |
|||
private SubmitStrategy submitStrategy; |
|||
private ProcessingStrategy processingStrategy; |
|||
|
|||
public Queue() { |
|||
} |
|||
|
|||
public Queue(QueueId id) { |
|||
super(id); |
|||
} |
|||
|
|||
public Queue(TenantId tenantId, TenantProfileQueueConfiguration queueConfiguration) { |
|||
this.tenantId = tenantId; |
|||
this.name = queueConfiguration.getName(); |
|||
this.topic = queueConfiguration.getTopic(); |
|||
this.pollInterval = queueConfiguration.getPollInterval(); |
|||
this.partitions = queueConfiguration.getPartitions(); |
|||
this.consumerPerPartition = queueConfiguration.isConsumerPerPartition(); |
|||
this.packProcessingTimeout = queueConfiguration.getPackProcessingTimeout(); |
|||
this.submitStrategy = queueConfiguration.getSubmitStrategy(); |
|||
this.processingStrategy = queueConfiguration.getProcessingStrategy(); |
|||
setAdditionalInfo(queueConfiguration.getAdditionalInfo()); |
|||
} |
|||
|
|||
@Override |
|||
public String getSearchText() { |
|||
return getName(); |
|||
} |
|||
} |
|||
@ -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,34 @@ |
|||
/** |
|||
* 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.tenant.profile; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.queue.ProcessingStrategy; |
|||
import org.thingsboard.server.common.data.queue.SubmitStrategy; |
|||
|
|||
@Data |
|||
public class TenantProfileQueueConfiguration { |
|||
private String name; |
|||
private String topic; |
|||
private int pollInterval; |
|||
private int partitions; |
|||
private boolean consumerPerPartition; |
|||
private long packProcessingTimeout; |
|||
private SubmitStrategy submitStrategy; |
|||
private ProcessingStrategy processingStrategy; |
|||
private JsonNode additionalInfo; |
|||
} |
|||
@ -1,62 +0,0 @@ |
|||
/** |
|||
* 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.msg.queue; |
|||
|
|||
import lombok.ToString; |
|||
|
|||
import java.util.Objects; |
|||
|
|||
@ToString |
|||
public class ServiceQueue { |
|||
|
|||
public static final String MAIN = "Main"; |
|||
|
|||
private final ServiceType type; |
|||
private final String queue; |
|||
|
|||
public ServiceQueue(ServiceType type) { |
|||
this.type = type; |
|||
this.queue = MAIN; |
|||
} |
|||
|
|||
public ServiceQueue(ServiceType type, String queue) { |
|||
this.type = type; |
|||
this.queue = queue != null ? queue : MAIN; |
|||
} |
|||
|
|||
public ServiceType getType() { |
|||
return type; |
|||
} |
|||
|
|||
public String getQueue() { |
|||
return queue; |
|||
} |
|||
|
|||
@Override |
|||
public boolean equals(Object o) { |
|||
if (this == o) return true; |
|||
if (o == null || getClass() != o.getClass()) return false; |
|||
ServiceQueue that = (ServiceQueue) o; |
|||
return type == that.type && |
|||
queue.equals(that.queue); |
|||
} |
|||
|
|||
@Override |
|||
public int hashCode() { |
|||
return Objects.hash(type, queue); |
|||
} |
|||
|
|||
} |
|||
@ -1,54 +0,0 @@ |
|||
/** |
|||
* 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.msg.queue; |
|||
|
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
|
|||
import java.util.Objects; |
|||
|
|||
@ToString |
|||
public class ServiceQueueKey { |
|||
@Getter |
|||
private final ServiceQueue serviceQueue; |
|||
|
|||
@Getter |
|||
private final TenantId tenantId; |
|||
|
|||
public ServiceQueueKey(ServiceQueue serviceQueue, TenantId tenantId) { |
|||
this.serviceQueue = serviceQueue; |
|||
this.tenantId = tenantId; |
|||
} |
|||
|
|||
@Override |
|||
public boolean equals(Object o) { |
|||
if (this == o) return true; |
|||
if (o == null || getClass() != o.getClass()) return false; |
|||
ServiceQueueKey that = (ServiceQueueKey) o; |
|||
return serviceQueue.equals(that.serviceQueue) && |
|||
Objects.equals(tenantId, that.tenantId); |
|||
} |
|||
|
|||
@Override |
|||
public int hashCode() { |
|||
return Objects.hash(serviceQueue, tenantId); |
|||
} |
|||
|
|||
public ServiceType getServiceType() { |
|||
return serviceQueue.getType(); |
|||
} |
|||
} |
|||
@ -1,62 +0,0 @@ |
|||
/** |
|||
* 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.queue; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import org.springframework.stereotype.Service; |
|||
import org.springframework.util.StringUtils; |
|||
import org.thingsboard.server.common.msg.queue.ServiceQueue; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; |
|||
import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import java.util.Collections; |
|||
import java.util.LinkedHashSet; |
|||
import java.util.Set; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@Service |
|||
@RequiredArgsConstructor |
|||
public class DefaultQueueService implements QueueService { |
|||
|
|||
private final TbQueueRuleEngineSettings ruleEngineSettings; |
|||
private Set<String> ruleEngineQueues; |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
ruleEngineQueues = ruleEngineSettings.getQueues().stream() |
|||
.map(TbRuleEngineQueueConfiguration::getName).collect(Collectors.toCollection(LinkedHashSet::new)); |
|||
} |
|||
|
|||
@Override |
|||
public Set<String> getQueuesByServiceType(ServiceType type) { |
|||
if (type == ServiceType.TB_RULE_ENGINE) { |
|||
return ruleEngineQueues; |
|||
} else { |
|||
return Collections.emptySet(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public String resolve(ServiceType serviceType, String queueName) { |
|||
if (StringUtils.isEmpty(queueName) || !getQueuesByServiceType(serviceType).contains(queueName)) { |
|||
return ServiceQueue.MAIN; |
|||
} else { |
|||
return queueName; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,114 @@ |
|||
/** |
|||
* 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.queue; |
|||
|
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.context.annotation.Bean; |
|||
import org.springframework.context.annotation.Configuration; |
|||
import org.thingsboard.server.queue.azure.servicebus.TbServiceBusAdmin; |
|||
import org.thingsboard.server.queue.azure.servicebus.TbServiceBusQueueConfigs; |
|||
import org.thingsboard.server.queue.azure.servicebus.TbServiceBusSettings; |
|||
import org.thingsboard.server.queue.kafka.TbKafkaAdmin; |
|||
import org.thingsboard.server.queue.kafka.TbKafkaSettings; |
|||
import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubAdmin; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSettings; |
|||
import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; |
|||
import org.thingsboard.server.queue.rabbitmq.TbRabbitMqAdmin; |
|||
import org.thingsboard.server.queue.rabbitmq.TbRabbitMqQueueArguments; |
|||
import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; |
|||
import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; |
|||
import org.thingsboard.server.queue.sqs.TbAwsSqsQueueAttributes; |
|||
import org.thingsboard.server.queue.sqs.TbAwsSqsSettings; |
|||
|
|||
@Configuration |
|||
public class RuleEngineTbQueueAdminFactory { |
|||
|
|||
@Autowired(required = false) |
|||
private TbKafkaTopicConfigs kafkaTopicConfigs; |
|||
@Autowired(required = false) |
|||
private TbKafkaSettings kafkaSettings; |
|||
|
|||
@Autowired(required = false) |
|||
private TbAwsSqsQueueAttributes awsSqsQueueAttributes; |
|||
@Autowired(required = false) |
|||
private TbAwsSqsSettings awsSqsSettings; |
|||
|
|||
@Autowired(required = false) |
|||
private TbPubSubSubscriptionSettings pubSubSubscriptionSettings; |
|||
@Autowired(required = false) |
|||
private TbPubSubSettings pubSubSettings; |
|||
|
|||
@Autowired(required = false) |
|||
private TbRabbitMqQueueArguments rabbitMqQueueArguments; |
|||
@Autowired(required = false) |
|||
private TbRabbitMqSettings rabbitMqSettings; |
|||
|
|||
@Autowired(required = false) |
|||
private TbServiceBusQueueConfigs serviceBusQueueConfigs; |
|||
@Autowired(required = false) |
|||
private TbServiceBusSettings serviceBusSettings; |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='kafka'") |
|||
@Bean |
|||
public TbQueueAdmin createKafkaAdmin() { |
|||
return new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); |
|||
} |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='aws-sqs'") |
|||
@Bean |
|||
public TbQueueAdmin createAwsSqsAdmin() { |
|||
return new TbAwsSqsAdmin(awsSqsSettings, awsSqsQueueAttributes.getRuleEngineAttributes()); |
|||
} |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='pubsub'") |
|||
@Bean |
|||
public TbQueueAdmin createPubSubAdmin() { |
|||
return new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); |
|||
} |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='rabbitmq'") |
|||
@Bean |
|||
public TbQueueAdmin createRabbitMqAdmin() { |
|||
return new TbRabbitMqAdmin(rabbitMqSettings, rabbitMqQueueArguments.getRuleEngineArgs()); |
|||
} |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='service-bus'") |
|||
@Bean |
|||
public TbQueueAdmin createServiceBusAdmin() { |
|||
return new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getRuleEngineConfigs()); |
|||
} |
|||
|
|||
@ConditionalOnExpression("'${queue.type:null}'=='in-memory'") |
|||
@Bean |
|||
public TbQueueAdmin createInMemoryAdmin() { |
|||
return new TbQueueAdmin() { |
|||
|
|||
@Override |
|||
public void createTopicIfNotExists(String topic) { |
|||
} |
|||
|
|||
@Override |
|||
public void deleteTopic(String topic) { |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
} |
|||
}; |
|||
} |
|||
} |
|||
@ -0,0 +1,54 @@ |
|||
/** |
|||
* 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.queue.discovery; |
|||
|
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
@Service |
|||
public class NotificationsTopicService { |
|||
|
|||
private Map<String, TopicPartitionInfo> tbCoreNotificationTopics = new HashMap<>(); |
|||
private Map<String, TopicPartitionInfo> tbRuleEngineNotificationTopics = new HashMap<>(); |
|||
|
|||
/** |
|||
* Each Service should start a consumer for messages that target individual service instance based on serviceId. |
|||
* This topic is likely to have single partition, and is always assigned to the service. |
|||
* @param serviceType |
|||
* @param serviceId |
|||
* @return |
|||
*/ |
|||
public TopicPartitionInfo getNotificationsTopic(ServiceType serviceType, String serviceId) { |
|||
switch (serviceType) { |
|||
case TB_CORE: |
|||
return tbCoreNotificationTopics.computeIfAbsent(serviceId, |
|||
id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); |
|||
case TB_RULE_ENGINE: |
|||
return tbRuleEngineNotificationTopics.computeIfAbsent(serviceId, |
|||
id -> buildNotificationsTopicPartitionInfo(serviceType, serviceId)); |
|||
default: |
|||
return buildNotificationsTopicPartitionInfo(serviceType, serviceId); |
|||
} |
|||
} |
|||
|
|||
private TopicPartitionInfo buildNotificationsTopicPartitionInfo(ServiceType serviceType, String serviceId) { |
|||
return new TopicPartitionInfo(serviceType.name().toLowerCase() + ".notifications." + serviceId, null, null, false); |
|||
} |
|||
} |
|||
@ -0,0 +1,56 @@ |
|||
/** |
|||
* 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.queue.discovery; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
public class QueueKey { |
|||
private static final String MAIN = "Main"; |
|||
|
|||
private final ServiceType type; |
|||
private final String queueName; |
|||
private final TenantId tenantId; |
|||
|
|||
public QueueKey(ServiceType type, Queue queue) { |
|||
this.type = type; |
|||
this.queueName = queue.getName(); |
|||
this.tenantId = queue.getTenantId(); |
|||
} |
|||
|
|||
public QueueKey(ServiceType type, QueueRoutingInfo queueRoutingInfo) { |
|||
this.type = type; |
|||
this.queueName = queueRoutingInfo.getQueueName(); |
|||
this.tenantId = queueRoutingInfo.getTenantId(); |
|||
} |
|||
|
|||
public QueueKey(ServiceType type, TenantId tenantId) { |
|||
this.type = type; |
|||
this.queueName = MAIN; |
|||
this.tenantId = tenantId != null ? tenantId : TenantId.SYS_TENANT_ID; |
|||
} |
|||
|
|||
public QueueKey(ServiceType type) { |
|||
this.type = type; |
|||
this.queueName = MAIN; |
|||
this.tenantId = TenantId.SYS_TENANT_ID; |
|||
} |
|||
} |
|||
@ -0,0 +1,60 @@ |
|||
/** |
|||
* 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.queue.discovery; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.QueueId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.queue.Queue; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.GetQueueRoutingInfoResponseMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.QueueUpdateMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
|
|||
@Data |
|||
public class QueueRoutingInfo { |
|||
|
|||
private final TenantId tenantId; |
|||
private final QueueId queueId; |
|||
private final String queueName; |
|||
private final String queueTopic; |
|||
private final int partitions; |
|||
|
|||
public QueueRoutingInfo(Queue queue) { |
|||
this.tenantId = queue.getTenantId(); |
|||
this.queueId = queue.getId(); |
|||
this.queueName = queue.getName(); |
|||
this.queueTopic = queue.getTopic(); |
|||
this.partitions = queue.getPartitions(); |
|||
} |
|||
|
|||
public QueueRoutingInfo(GetQueueRoutingInfoResponseMsg routingInfo) { |
|||
this.tenantId = new TenantId(new UUID(routingInfo.getTenantIdMSB(), routingInfo.getTenantIdLSB())); |
|||
this.queueId = new QueueId(new UUID(routingInfo.getQueueIdMSB(), routingInfo.getQueueIdLSB())); |
|||
this.queueName = routingInfo.getQueueName(); |
|||
this.queueTopic = routingInfo.getQueueTopic(); |
|||
this.partitions = routingInfo.getPartitions(); |
|||
} |
|||
|
|||
public QueueRoutingInfo(QueueUpdateMsg queueUpdateMsg) { |
|||
this.tenantId = new TenantId(new UUID(queueUpdateMsg.getTenantIdMSB(), queueUpdateMsg.getTenantIdLSB())); |
|||
this.queueId = new QueueId(new UUID(queueUpdateMsg.getQueueIdMSB(), queueUpdateMsg.getQueueIdLSB())); |
|||
this.queueName = queueUpdateMsg.getQueueName(); |
|||
this.queueTopic = queueUpdateMsg.getQueueTopic(); |
|||
this.partitions = queueUpdateMsg.getPartitions(); |
|||
} |
|||
} |
|||
@ -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.queue.discovery; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface QueueRoutingInfoService { |
|||
|
|||
List<QueueRoutingInfo> getAllQueuesRoutingInfo(); |
|||
|
|||
} |
|||
@ -1,44 +0,0 @@ |
|||
/** |
|||
* 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.queue.discovery; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.queue.ServiceQueue; |
|||
|
|||
import java.util.Objects; |
|||
|
|||
@AllArgsConstructor |
|||
public class TopicPartitionInfoKey { |
|||
private ServiceQueue serviceQueue; |
|||
private TenantId isolatedTenantId; |
|||
private int partition; |
|||
|
|||
@Override |
|||
public boolean equals(Object o) { |
|||
if (this == o) return true; |
|||
if (o == null || getClass() != o.getClass()) return false; |
|||
TopicPartitionInfoKey that = (TopicPartitionInfoKey) o; |
|||
return partition == that.partition && |
|||
serviceQueue.equals(that.serviceQueue) && |
|||
Objects.equals(isolatedTenantId, that.isolatedTenantId); |
|||
} |
|||
|
|||
@Override |
|||
public int hashCode() { |
|||
return Objects.hash(serviceQueue, isolatedTenantId, partition); |
|||
} |
|||
} |
|||
Some files were not shown because too many files changed in this diff
Loading…
Reference in new issue