diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseEdgeControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseEdgeControllerTest.java index 9259f3253d..326d1465c1 100644 --- a/application/src/test/java/org/thingsboard/server/controller/BaseEdgeControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/BaseEdgeControllerTest.java @@ -48,15 +48,17 @@ import org.thingsboard.server.gen.edge.v1.UserUpdateMsg; import java.util.ArrayList; import java.util.Collections; import java.util.List; -import java.util.Random; +import static org.assertj.core.api.Assertions.assertThat; import static org.hamcrest.Matchers.containsString; -import static org.hamcrest.Matchers.nullValue; import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; public abstract class BaseEdgeControllerTest extends AbstractControllerTest { + public static final String EDGE_HOST = "localhost"; + public static final int EDGE_PORT = 7070; + private IdComparator idComparator = new IdComparator<>(); private Tenant savedTenant; @@ -68,7 +70,7 @@ public abstract class BaseEdgeControllerTest extends AbstractControllerTest { loginSysAdmin(); Tenant tenant = new Tenant(); - tenant.setTitle("My tenant"); + tenant.setTitle("My tenant for Edge"); savedTenant = doPost("/api/tenant", tenant, Tenant.class); tenantId = savedTenant.getId(); Assert.assertNotNull(savedTenant); @@ -667,45 +669,45 @@ public abstract class BaseEdgeControllerTest extends AbstractControllerTest { @Test public void testSyncEdge() throws Exception { - Edge edge = doPost("/api/edge", constructEdge("Test Edge", "test"), Edge.class); + Edge edge = doPost("/api/edge", constructEdge("Test Sync Edge", "test"), Edge.class); Device device = new Device(); - device.setName("Edge Device 1"); + device.setName("Test Sync Edge Device 1"); device.setType("default"); Device savedDevice = doPost("/api/device", device, Device.class); doPost("/api/edge/" + edge.getId().getId().toString() + "/device/" + savedDevice.getId().getId().toString(), Device.class); Asset asset = new Asset(); - asset.setName("Edge Asset 1"); + asset.setName("Test Sync Edge Asset 1"); asset.setType("test"); Asset savedAsset = doPost("/api/asset", asset, Asset.class); doPost("/api/edge/" + edge.getId().getId().toString() + "/asset/" + savedAsset.getId().getId().toString(), Asset.class); - EdgeImitator edgeImitator = new EdgeImitator("localhost", 7070, edge.getRoutingKey(), edge.getSecret()); + EdgeImitator edgeImitator = new EdgeImitator(EDGE_HOST, EDGE_PORT, edge.getRoutingKey(), edge.getSecret()); edgeImitator.ignoreType(UserCredentialsUpdateMsg.class); edgeImitator.expectMessageAmount(11); edgeImitator.connect(); - Assert.assertTrue(edgeImitator.waitForMessages()); + assertThat(edgeImitator.waitForMessages()).as("await for messages on first connect").isTrue(); - Assert.assertEquals(2, edgeImitator.findAllMessagesByType(RuleChainUpdateMsg.class).size()); // one msg during sync process, another from edge creation - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(DeviceProfileUpdateMsg.class).size()); // one msg during sync process for 'default' device profile - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(DeviceUpdateMsg.class).size()); // one msg once device assigned to edge - Assert.assertEquals(2, edgeImitator.findAllMessagesByType(AssetUpdateMsg.class).size()); // two msgs - one during sync process, and one more once asset assigned to edge - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(UserUpdateMsg.class).size()); // one msg during sync process for tenant admin user - Assert.assertEquals(4, edgeImitator.findAllMessagesByType(AdminSettingsUpdateMsg.class).size()); + assertThat(edgeImitator.findAllMessagesByType(RuleChainUpdateMsg.class)).as("one msg during sync process, another from edge creation").hasSize(2); + assertThat(edgeImitator.findAllMessagesByType(DeviceProfileUpdateMsg.class)).as("one msg during sync process for 'default' device profile").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(DeviceUpdateMsg.class)).as("one msg once device assigned to edge").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(AssetUpdateMsg.class)).as("two msgs - one during sync process, and one more once asset assigned to edge").hasSize(2); + assertThat(edgeImitator.findAllMessagesByType(UserUpdateMsg.class)).as("one msg during sync process for tenant admin user").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(AdminSettingsUpdateMsg.class)).as("admin setting update").hasSize(4); edgeImitator.expectMessageAmount(8); doPost("/api/edge/sync/" + edge.getId()); - Assert.assertTrue(edgeImitator.waitForMessages()); + assertThat(edgeImitator.waitForMessages()).as("await for messages after edge sync rest api call").isTrue(); - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(RuleChainUpdateMsg.class).size()); - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(DeviceProfileUpdateMsg.class).size()); - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(AssetUpdateMsg.class).size()); - Assert.assertEquals(1, edgeImitator.findAllMessagesByType(UserUpdateMsg.class).size()); - Assert.assertEquals(4, edgeImitator.findAllMessagesByType(AdminSettingsUpdateMsg.class).size()); + assertThat(edgeImitator.findAllMessagesByType(RuleChainUpdateMsg.class)).as("rule chain msg").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(DeviceProfileUpdateMsg.class)).as("device profile msg").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(AssetUpdateMsg.class)).as("asset update msg").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(UserUpdateMsg.class)).as("user update msg").hasSize(1); + assertThat(edgeImitator.findAllMessagesByType(AdminSettingsUpdateMsg.class)).as("admin setting update msg").hasSize(4); edgeImitator.allowIgnoredTypes(); try { @@ -720,4 +722,4 @@ public abstract class BaseEdgeControllerTest extends AbstractControllerTest { .andExpect(status().isOk()); } -} \ No newline at end of file +} diff --git a/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java b/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java index 2dea0d2811..c412e86ec6 100644 --- a/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java +++ b/application/src/test/java/org/thingsboard/server/edge/BaseEdgeTest.java @@ -767,7 +767,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest { edgeImitator.expectMessageAmount(1); doDelete("/api/alarm/" + savedAlarm.getId().getId().toString()) .andExpect(status().isOk()); - Assert.assertTrue(edgeImitator.waitForMessages(1)); + Assert.assertTrue(edgeImitator.waitForMessages()); latestMessage = edgeImitator.getLatestMessage(); Assert.assertTrue(latestMessage instanceof AlarmUpdateMsg); alarmUpdateMsg = (AlarmUpdateMsg) latestMessage; diff --git a/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java b/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java index e257f5c5d2..46ee981be1 100644 --- a/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java +++ b/application/src/test/java/org/thingsboard/server/edge/imitator/EdgeImitator.java @@ -65,6 +65,8 @@ import java.util.stream.Collectors; @Slf4j public class EdgeImitator { + public static final int TIMEOUT_IN_SECONDS = 30; + private String routingKey; private String routingSecret; @@ -293,7 +295,7 @@ public class EdgeImitator { } public boolean waitForMessages() throws InterruptedException { - return waitForMessages(5); + return waitForMessages(TIMEOUT_IN_SECONDS); } public boolean waitForMessages(int timeoutInSeconds) throws InterruptedException { @@ -308,7 +310,7 @@ public class EdgeImitator { } public boolean waitForResponses() throws InterruptedException { - return responsesLatch.await(5, TimeUnit.SECONDS); + return responsesLatch.await(TIMEOUT_IN_SECONDS, TimeUnit.SECONDS); } public void expectResponsesAmount(int messageAmount) { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java index 026f879c8d..f5a592e1c9 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsMonolithQueueFactory.java @@ -102,7 +102,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override @@ -112,7 +112,7 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, coreSettings.getTopic()); } @Override @@ -182,13 +182,13 @@ public class AwsSqsMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToOtaPackageStateServiceMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getOtaPackageTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java index eaa73bc108..87ebe42024 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbCoreQueueFactory.java @@ -43,6 +43,7 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; import org.thingsboard.server.queue.sqs.TbAwsSqsProducerTemplate; @@ -63,6 +64,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -77,7 +79,8 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, - TbAwsSqsQueueAttributes sqsQueueAttributes) { + TbAwsSqsQueueAttributes sqsQueueAttributes, + TbQueueTransportNotificationSettings transportNotificationSettings) { this.sqsSettings = sqsSettings; this.coreSettings = coreSettings; this.transportApiSettings = transportApiSettings; @@ -85,6 +88,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); @@ -95,7 +99,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, transportNotificationSettings.getNotificationsTopic()); } @Override @@ -105,7 +109,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override @@ -115,7 +119,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, coreSettings.getTopic()); } @Override @@ -139,7 +143,7 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getResponsesTopic()); } @Override @@ -172,13 +176,13 @@ public class AwsSqsTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueConsumer> createToUsageStatsServiceMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getUsageStatsTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToUsageStatsServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } @Override public TbQueueConsumer> createToOtaPackageStateServiceMsgConsumer() { - return new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, coreSettings.getOtaPackageTopic(), + return new TbAwsSqsConsumerTemplate<>(coreAdmin, sqsSettings, coreSettings.getOtaPackageTopic(), msg -> new TbProtoQueueMsg<>(msg.getKey(), ToOtaPackageStateServiceMsg.parseFrom(msg.getData()), msg.getHeaders())); } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java index 0e9f1812dd..05274625d3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTbRuleEngineQueueFactory.java @@ -37,6 +37,7 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; import org.thingsboard.server.queue.sqs.TbAwsSqsConsumerTemplate; @@ -57,6 +58,7 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbAwsSqsSettings sqsSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -68,13 +70,15 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory TbServiceInfoProvider serviceInfoProvider, TbAwsSqsSettings sqsSettings, TbAwsSqsQueueAttributes sqsQueueAttributes, - TbQueueRemoteJsInvokeSettings jsInvokeSettings) { + TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings) { this.partitionService = partitionService; this.coreSettings = coreSettings; this.serviceInfoProvider = serviceInfoProvider; this.ruleEngineSettings = ruleEngineSettings; this.sqsSettings = sqsSettings; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); @@ -84,17 +88,17 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override @@ -104,7 +108,7 @@ public class AwsSqsTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbAwsSqsProducerTemplate<>(coreAdmin, sqsSettings, coreSettings.getTopic()); + return new TbAwsSqsProducerTemplate<>(notificationAdmin, sqsSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java index 689f2c041e..80f9e27d1c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/AwsSqsTransportQueueFactory.java @@ -32,6 +32,7 @@ import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.sqs.TbAwsSqsAdmin; @@ -51,34 +52,39 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { private final TbAwsSqsSettings sqsSettings; private final TbQueueCoreSettings coreSettings; private final TbServiceInfoProvider serviceInfoProvider; + private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; + private final TbQueueAdmin ruleEngineAdmin; public AwsSqsTransportQueueFactory(TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbAwsSqsSettings sqsSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, - TbAwsSqsQueueAttributes sqsQueueAttributes) { + TbAwsSqsQueueAttributes sqsQueueAttributes, + TbQueueRuleEngineSettings ruleEngineSettings) { this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.sqsSettings = sqsSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; + this.ruleEngineSettings = ruleEngineSettings; this.coreAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getCoreAttributes()); this.transportApiAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getTransportApiAttributes()); this.notificationAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getNotificationsAttributes()); + this.ruleEngineAdmin = new TbAwsSqsAdmin(sqsSettings, sqsQueueAttributes.getRuleEngineAttributes()); } @Override public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { - TbAwsSqsProducerTemplate> producerTemplate = + TbQueueProducer> producerTemplate = new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic()); - TbAwsSqsConsumerTemplate> consumerTemplate = + TbQueueConsumer> consumerTemplate = new TbAwsSqsConsumerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getResponsesTopic() + "_" + serviceInfoProvider.getServiceId(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); @@ -96,7 +102,7 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbAwsSqsProducerTemplate<>(transportApiAdmin, sqsSettings, transportApiSettings.getRequestsTopic()); + return new TbAwsSqsProducerTemplate<>(ruleEngineAdmin, sqsSettings, ruleEngineSettings.getTopic()); } @Override @@ -126,5 +132,8 @@ public class AwsSqsTransportQueueFactory implements TbTransportQueueFactory { if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (ruleEngineAdmin != null) { + ruleEngineAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java index e08be485ea..b4dd38217d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java @@ -131,7 +131,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); - requestBuilder.admin(ruleEngineAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -151,7 +151,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi requestBuilder.settings(kafkaSettings); requestBuilder.clientId("monolith-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -328,5 +328,8 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (fwUpdatesAdmin != null) { + fwUpdatesAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java index bec01c7201..f560fcc5f4 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java @@ -49,6 +49,7 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -65,6 +66,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbKafkaConsumerStatsService consumerStatsService; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -80,6 +82,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbQueueTransportApiSettings transportApiSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbKafkaConsumerStatsService consumerStatsService, + TbQueueTransportNotificationSettings transportNotificationSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; @@ -89,6 +92,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { this.transportApiSettings = transportApiSettings; this.jsInvokeSettings = jsInvokeSettings; this.consumerStatsService = consumerStatsService; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -103,8 +107,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-notifications-" + serviceInfoProvider.getServiceId()); - requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.defaultTopic(transportNotificationSettings.getNotificationsTopic()); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -124,7 +128,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); - requestBuilder.admin(ruleEngineAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -144,7 +148,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -192,8 +196,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-core-transport-api-producer-" + serviceInfoProvider.getServiceId()); - requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.defaultTopic(transportApiSettings.getResponsesTopic()); + requestBuilder.admin(transportApiAdmin); return requestBuilder.build(); } @@ -294,5 +298,8 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory { if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (fwUpdatesAdmin != null) { + fwUpdatesAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java index 58bc38d3a9..dc3db14ec2 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java @@ -46,6 +46,7 @@ import org.thingsboard.server.queue.kafka.TbKafkaTopicConfigs; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; @@ -63,6 +64,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbKafkaConsumerStatsService consumerStatsService; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -77,6 +79,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbQueueRuleEngineSettings ruleEngineSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, TbKafkaConsumerStatsService consumerStatsService, + TbQueueTransportNotificationSettings transportNotificationSettings, TbKafkaTopicConfigs kafkaTopicConfigs) { this.partitionService = partitionService; this.kafkaSettings = kafkaSettings; @@ -85,6 +88,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { this.ruleEngineSettings = ruleEngineSettings; this.jsInvokeSettings = jsInvokeSettings; this.consumerStatsService = consumerStatsService; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbKafkaAdmin(kafkaSettings, kafkaTopicConfigs.getRuleEngineConfigs()); @@ -98,8 +102,8 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-transport-notifications-" + serviceInfoProvider.getServiceId()); - requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.defaultTopic(transportNotificationSettings.getNotificationsTopic()); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -108,8 +112,8 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-" + serviceInfoProvider.getServiceId()); - requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); + requestBuilder.admin(ruleEngineAdmin); return requestBuilder.build(); } @@ -119,11 +123,10 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-rule-engine-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(ruleEngineSettings.getTopic()); - requestBuilder.admin(ruleEngineAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } - @Override public TbQueueProducer> createTbCoreMsgProducer() { TbKafkaProducerTemplate.TbKafkaProducerTemplateBuilder> requestBuilder = TbKafkaProducerTemplate.builder(); @@ -150,7 +153,7 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { requestBuilder.settings(kafkaSettings); requestBuilder.clientId("tb-rule-engine-to-core-notifications-" + serviceInfoProvider.getServiceId()); requestBuilder.defaultTopic(coreSettings.getTopic()); - requestBuilder.admin(coreAdmin); + requestBuilder.admin(notificationAdmin); return requestBuilder.build(); } @@ -239,5 +242,8 @@ public class KafkaTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory { if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (fwUpdatesAdmin != null) { + fwUpdatesAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java index 1d2f36c5c2..d22839b9be 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubMonolithQueueFactory.java @@ -112,7 +112,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, ruleEngineSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override @@ -122,7 +122,7 @@ public class PubSubMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEng @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java index 65ea183d66..086c4d9420 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbCoreQueueFactory.java @@ -46,7 +46,9 @@ import org.thingsboard.server.queue.pubsub.TbPubSubSettings; import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -61,11 +63,14 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; + private final TbQueueAdmin ruleEngineAdmin; public PubSubTbCoreQueueFactory(TbPubSubSettings pubSubSettings, TbQueueCoreSettings coreSettings, @@ -73,6 +78,8 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, + TbQueueRuleEngineSettings ruleEngineSettings, TbPubSubSubscriptionSettings pubSubSubscriptionSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; @@ -80,16 +87,19 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; + this.ruleEngineSettings = ruleEngineSettings; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.jsExecutorAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getJsExecutorSettings()); this.transportApiAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getTransportApiSettings()); this.notificationAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getNotificationsSettings()); + this.ruleEngineAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); } @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic()); } @Override @@ -99,7 +109,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override @@ -109,7 +119,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic()); } @Override @@ -133,7 +143,7 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(transportApiAdmin, pubSubSettings, transportApiSettings.getResponsesTopic()); } @Override @@ -195,5 +205,8 @@ public class PubSubTbCoreQueueFactory implements TbCoreQueueFactory { if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (ruleEngineAdmin != null) { + ruleEngineAdmin.destroy(); + } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java index b02322496f..f8f190853d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/PubSubTbRuleEngineQueueFactory.java @@ -45,6 +45,7 @@ import org.thingsboard.server.queue.pubsub.TbPubSubSubscriptionSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; @@ -60,6 +61,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -72,6 +74,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, TbPubSubSubscriptionSettings pubSubSubscriptionSettings) { this.pubSubSettings = pubSubSettings; this.coreSettings = coreSettings; @@ -79,6 +82,7 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getCoreSettings()); this.ruleEngineAdmin = new TbPubSubAdmin(pubSubSettings, pubSubSubscriptionSettings.getRuleEngineSettings()); @@ -88,28 +92,27 @@ public class PubSubTbRuleEngineQueueFactory implements TbRuleEngineQueueFactory @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(ruleEngineAdmin, pubSubSettings, ruleEngineSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createTbCoreMsgProducer() { return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); - } @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbPubSubProducerTemplate<>(coreAdmin, pubSubSettings, coreSettings.getTopic()); + return new TbPubSubProducerTemplate<>(notificationAdmin, pubSubSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java index 7d48803258..e8ec6e4cff 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqMonolithQueueFactory.java @@ -66,13 +66,13 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE private final TbQueueTransportApiSettings transportApiSettings; private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbRabbitMqSettings rabbitMqSettings; + private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; private final TbQueueAdmin jsExecutorAdmin; private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; - private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; public RabbitMqMonolithQueueFactory(PartitionService partitionService, TbQueueCoreSettings coreSettings, TbQueueRuleEngineSettings ruleEngineSettings, @@ -89,13 +89,13 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.rabbitMqSettings = rabbitMqSettings; + this.jsInvokeSettings = jsInvokeSettings; this.coreAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getCoreArgs()); this.ruleEngineAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getRuleEngineArgs()); this.jsExecutorAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getJsExecutorArgs()); this.transportApiAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getTransportApiArgs()); this.notificationAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getNotificationsArgs()); - this.jsInvokeSettings = jsInvokeSettings; } @Override @@ -110,7 +110,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override @@ -120,7 +120,7 @@ public class RabbitMqMonolithQueueFactory implements TbCoreQueueFactory, TbRuleE @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java index c0abeab162..303180bad6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbCoreQueueFactory.java @@ -48,6 +48,7 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -63,6 +64,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -77,6 +79,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, TbRabbitMqQueueArguments queueArguments) { this.rabbitMqSettings = rabbitMqSettings; this.coreSettings = coreSettings; @@ -85,6 +88,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getCoreArgs()); this.ruleEngineAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getRuleEngineArgs()); @@ -95,7 +99,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, transportNotificationSettings.getNotificationsTopic()); } @Override @@ -105,7 +109,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override @@ -115,7 +119,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, coreSettings.getTopic()); } @Override @@ -139,7 +143,7 @@ public class RabbitMqTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getResponsesTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java index 2f8a3ed655..470e4a8785 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTbRuleEngineQueueFactory.java @@ -45,6 +45,7 @@ import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; @@ -60,6 +61,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbRabbitMqSettings rabbitMqSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -71,6 +73,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor TbServiceInfoProvider serviceInfoProvider, TbRabbitMqSettings rabbitMqSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, TbRabbitMqQueueArguments queueArguments) { this.partitionService = partitionService; this.coreSettings = coreSettings; @@ -78,6 +81,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor this.ruleEngineSettings = ruleEngineSettings; this.rabbitMqSettings = rabbitMqSettings; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getCoreArgs()); this.ruleEngineAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getRuleEngineArgs()); @@ -87,17 +91,17 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override @@ -107,7 +111,7 @@ public class RabbitMqTbRuleEngineQueueFactory implements TbRuleEngineQueueFactor @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbRabbitMqProducerTemplate<>(coreAdmin, rabbitMqSettings, coreSettings.getTopic()); + return new TbRabbitMqProducerTemplate<>(notificationAdmin, rabbitMqSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java index ec27a3ed04..e40816b242 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/RabbitMqTransportQueueFactory.java @@ -37,6 +37,7 @@ import org.thingsboard.server.queue.rabbitmq.TbRabbitMqProducerTemplate; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqQueueArguments; import org.thingsboard.server.queue.rabbitmq.TbRabbitMqSettings; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; @@ -51,6 +52,7 @@ public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { private final TbRabbitMqSettings rabbitMqSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -62,12 +64,14 @@ public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { TbRabbitMqSettings rabbitMqSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, + TbQueueRuleEngineSettings ruleEngineSettings, TbRabbitMqQueueArguments queueArguments) { this.transportApiSettings = transportApiSettings; this.transportNotificationSettings = transportNotificationSettings; this.rabbitMqSettings = rabbitMqSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; + this.ruleEngineSettings = ruleEngineSettings; this.coreAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getCoreArgs()); this.ruleEngineAdmin = new TbRabbitMqAdmin(rabbitMqSettings, queueArguments.getRuleEngineArgs()); @@ -77,10 +81,10 @@ public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueRequestTemplate, TbProtoQueueMsg> createTransportApiRequestTemplate() { - TbRabbitMqProducerTemplate> producerTemplate = + TbQueueProducer> producerTemplate = new TbRabbitMqProducerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getRequestsTopic()); - TbRabbitMqConsumerTemplate> consumerTemplate = + TbQueueConsumer> consumerTemplate = new TbRabbitMqConsumerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getResponsesTopic() + "." + serviceInfoProvider.getServiceId(), msg -> new TbProtoQueueMsg<>(msg.getKey(), TransportApiResponseMsg.parseFrom(msg.getData()), msg.getHeaders())); @@ -98,7 +102,7 @@ public class RabbitMqTransportQueueFactory implements TbTransportQueueFactory { @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbRabbitMqProducerTemplate<>(transportApiAdmin, rabbitMqSettings, transportApiSettings.getRequestsTopic()); + return new TbRabbitMqProducerTemplate<>(ruleEngineAdmin, rabbitMqSettings, ruleEngineSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java index 3731a56aee..39f41e8faf 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusMonolithQueueFactory.java @@ -109,7 +109,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @Override @@ -119,7 +119,7 @@ public class ServiceBusMonolithQueueFactory implements TbCoreQueueFactory, TbRul @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java index 7a9da236ed..0280012008 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbCoreQueueFactory.java @@ -48,6 +48,7 @@ import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import javax.annotation.PreDestroy; import java.nio.charset.StandardCharsets; @@ -63,6 +64,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { private final PartitionService partitionService; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -77,6 +79,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { PartitionService partitionService, TbServiceInfoProvider serviceInfoProvider, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, TbServiceBusQueueConfigs serviceBusQueueConfigs) { this.serviceBusSettings = serviceBusSettings; this.coreSettings = coreSettings; @@ -85,6 +88,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { this.partitionService = partitionService; this.serviceInfoProvider = serviceInfoProvider; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getRuleEngineConfigs()); @@ -95,7 +99,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, transportNotificationSettings.getNotificationsTopic()); } @Override @@ -105,7 +109,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @Override @@ -115,7 +119,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, coreSettings.getTopic()); } @Override @@ -139,7 +143,7 @@ public class ServiceBusTbCoreQueueFactory implements TbCoreQueueFactory { @Override public TbQueueProducer> createTransportApiResponseProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(transportApiAdmin, serviceBusSettings, transportApiSettings.getResponsesTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java index be11736ad2..d653c8aa17 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTbRuleEngineQueueFactory.java @@ -45,6 +45,7 @@ import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; import org.thingsboard.server.queue.settings.TbQueueRemoteJsInvokeSettings; import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; +import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; import org.thingsboard.server.queue.settings.TbRuleEngineQueueConfiguration; import javax.annotation.PreDestroy; @@ -60,6 +61,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbServiceBusSettings serviceBusSettings; private final TbQueueRemoteJsInvokeSettings jsInvokeSettings; + private final TbQueueTransportNotificationSettings transportNotificationSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin ruleEngineAdmin; @@ -71,6 +73,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact TbServiceInfoProvider serviceInfoProvider, TbServiceBusSettings serviceBusSettings, TbQueueRemoteJsInvokeSettings jsInvokeSettings, + TbQueueTransportNotificationSettings transportNotificationSettings, TbServiceBusQueueConfigs serviceBusQueueConfigs) { this.partitionService = partitionService; this.coreSettings = coreSettings; @@ -78,6 +81,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact this.ruleEngineSettings = ruleEngineSettings; this.serviceBusSettings = serviceBusSettings; this.jsInvokeSettings = jsInvokeSettings; + this.transportNotificationSettings = transportNotificationSettings; this.coreAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getCoreConfigs()); this.ruleEngineAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getRuleEngineConfigs()); @@ -87,17 +91,17 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact @Override public TbQueueProducer> createTransportNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, transportNotificationSettings.getNotificationsTopic()); } @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @Override public TbQueueProducer> createRuleEngineNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @Override @@ -107,7 +111,7 @@ public class ServiceBusTbRuleEngineQueueFactory implements TbRuleEngineQueueFact @Override public TbQueueProducer> createTbCoreNotificationsMsgProducer() { - return new TbServiceBusProducerTemplate<>(coreAdmin, serviceBusSettings, coreSettings.getTopic()); + return new TbServiceBusProducerTemplate<>(notificationAdmin, serviceBusSettings, coreSettings.getTopic()); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java index 3befd52729..85ecdd2643 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/provider/ServiceBusTransportQueueFactory.java @@ -37,6 +37,7 @@ import org.thingsboard.server.queue.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.settings.TbQueueCoreSettings; +import org.thingsboard.server.queue.settings.TbQueueRuleEngineSettings; import org.thingsboard.server.queue.settings.TbQueueTransportApiSettings; import org.thingsboard.server.queue.settings.TbQueueTransportNotificationSettings; @@ -51,14 +52,17 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory private final TbServiceBusSettings serviceBusSettings; private final TbServiceInfoProvider serviceInfoProvider; private final TbQueueCoreSettings coreSettings; + private final TbQueueRuleEngineSettings ruleEngineSettings; private final TbQueueAdmin coreAdmin; private final TbQueueAdmin transportApiAdmin; private final TbQueueAdmin notificationAdmin; + private final TbQueueAdmin ruleEngineAdmin; public ServiceBusTransportQueueFactory(TbQueueTransportApiSettings transportApiSettings, TbQueueTransportNotificationSettings transportNotificationSettings, TbServiceBusSettings serviceBusSettings, + TbQueueRuleEngineSettings ruleEngineSettings, TbServiceInfoProvider serviceInfoProvider, TbQueueCoreSettings coreSettings, TbServiceBusQueueConfigs serviceBusQueueConfigs) { @@ -67,10 +71,12 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory this.serviceBusSettings = serviceBusSettings; this.serviceInfoProvider = serviceInfoProvider; this.coreSettings = coreSettings; + this.ruleEngineSettings = ruleEngineSettings; this.coreAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getCoreConfigs()); this.transportApiAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getTransportApiConfigs()); this.notificationAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getNotificationsConfigs()); + this.ruleEngineAdmin = new TbServiceBusAdmin(serviceBusSettings, serviceBusQueueConfigs.getRuleEngineConfigs()); } @Override @@ -96,7 +102,7 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory @Override public TbQueueProducer> createRuleEngineMsgProducer() { - return new TbServiceBusProducerTemplate<>(transportApiAdmin, serviceBusSettings, transportApiSettings.getRequestsTopic()); + return new TbServiceBusProducerTemplate<>(ruleEngineAdmin, serviceBusSettings, ruleEngineSettings.getTopic()); } @Override @@ -127,5 +133,8 @@ public class ServiceBusTransportQueueFactory implements TbTransportQueueFactory if (notificationAdmin != null) { notificationAdmin.destroy(); } + if (ruleEngineAdmin != null) { + ruleEngineAdmin.destroy(); + } } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java index 3255bf75b4..f70d1c1ffe 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java @@ -336,12 +336,14 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe return savedDevice; } + @Transactional @Override - public void deleteDevice(TenantId tenantId, DeviceId deviceId) { + public void deleteDevice(final TenantId tenantId, final DeviceId deviceId) { log.trace("Executing deleteDevice [{}]", deviceId); validateId(deviceId, INCORRECT_DEVICE_ID + deviceId); Device device = deviceDao.findById(tenantId, deviceId.getId()); + final String deviceName = device.getName(); try { List entityViews = entityViewService.findEntityViewsByTenantIdAndEntityIdAsync(device.getTenantId(), deviceId).get(); if (entityViews != null && !entityViews.isEmpty()) { @@ -358,10 +360,10 @@ public class DeviceServiceImpl extends AbstractEntityService implements DeviceSe } deleteEntityRelations(tenantId, deviceId); - removeDeviceFromCacheByName(tenantId, device.getName()); - removeDeviceFromCacheById(tenantId, device.getId()); - deviceDao.removeById(tenantId, deviceId.getId()); + + removeDeviceFromCacheByName(tenantId, deviceName); + removeDeviceFromCacheById(tenantId, deviceId); } private void removeDeviceFromCacheByName(TenantId tenantId, String name) {