diff --git a/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java b/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java index 3032b1dfe0..48a31127f4 100644 --- a/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java +++ b/application/src/main/java/org/thingsboard/server/actors/service/ActorService.java @@ -20,7 +20,6 @@ import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; -import org.thingsboard.server.common.msg.system.ServiceToRuleEngineMsg; import org.thingsboard.server.common.transport.SessionMsgProcessor; import org.thingsboard.server.service.cluster.discovery.DiscoveryServiceListener; import org.thingsboard.server.service.cluster.rpc.RpcMsgListener; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java index 4cd5c664fd..0d0e38a1b2 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCoreConsumerService.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; import akka.actor.ActorRef; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java index 96b72843ed..413d5ee9dd 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbRuleEngineConsumerService.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; import akka.actor.ActorRef; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java index c700c18dc0..5b94db28b0 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/MsgPackCallback.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; import lombok.extern.slf4j.Slf4j; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java index 0541de7c63..1e1b352645 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerService.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; import org.thingsboard.server.gen.transport.TransportProtos; diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java index e3bb3a3c4b..f7912836c5 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbCoreConsumerStats.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java index 44dd058e1e..774377407a 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbMsgCallback.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; public interface TbMsgCallback { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java index 691a54aaab..671cd72262 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerService.java @@ -1,3 +1,18 @@ +/** + * 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.service.queue; public interface TbRuleEngineConsumerService { diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java index 80d05628f2..75711d20b2 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbRuleEngineConsumerStats.java @@ -25,13 +25,13 @@ public class TbRuleEngineConsumerStats { private final AtomicInteger totalCounter = new AtomicInteger(0); private final AtomicInteger sessionEventCounter = new AtomicInteger(0); - private final AtomicInteger postTelemetryCounter = new AtomicInteger(0); - private final AtomicInteger postAttributesCounter = new AtomicInteger(0); +// private final AtomicInteger postTelemetryCounter = new AtomicInteger(0); +// private final AtomicInteger postAttributesCounter = new AtomicInteger(0); private final AtomicInteger getAttributesCounter = new AtomicInteger(0); private final AtomicInteger subscribeToAttributesCounter = new AtomicInteger(0); private final AtomicInteger subscribeToRPCCounter = new AtomicInteger(0); private final AtomicInteger toDeviceRPCCallResponseCounter = new AtomicInteger(0); - private final AtomicInteger toServerRPCCallRequestCounter = new AtomicInteger(0); +// private final AtomicInteger toServerRPCCallRequestCounter = new AtomicInteger(0); private final AtomicInteger subscriptionInfoCounter = new AtomicInteger(0); private final AtomicInteger claimDeviceCounter = new AtomicInteger(0); @@ -40,12 +40,12 @@ public class TbRuleEngineConsumerStats { if (msg.hasSessionEvent()) { sessionEventCounter.incrementAndGet(); } - if (msg.hasPostTelemetry()) { - postTelemetryCounter.incrementAndGet(); - } - if (msg.hasPostAttributes()) { - postAttributesCounter.incrementAndGet(); - } +// if (msg.hasPostTelemetry()) { +// postTelemetryCounter.incrementAndGet(); +// } +// if (msg.hasPostAttributes()) { +// postAttributesCounter.incrementAndGet(); +// } if (msg.hasGetAttributes()) { getAttributesCounter.incrementAndGet(); } @@ -58,9 +58,9 @@ public class TbRuleEngineConsumerStats { if (msg.hasToDeviceRPCCallResponse()) { toDeviceRPCCallResponseCounter.incrementAndGet(); } - if (msg.hasToServerRPCCallRequest()) { - toServerRPCCallRequestCounter.incrementAndGet(); - } +// if (msg.hasToServerRPCCallRequest()) { +// toServerRPCCallRequestCounter.incrementAndGet(); +// } if (msg.hasSubscriptionInfo()) { subscriptionInfoCounter.incrementAndGet(); } diff --git a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java index d39465ff12..08578c3890 100644 --- a/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java +++ b/application/src/main/java/org/thingsboard/server/service/script/RemoteJsInvokeService.java @@ -20,7 +20,6 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import lombok.Getter; import lombok.extern.slf4j.Slf4j; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.scheduling.annotation.Scheduled; @@ -29,7 +28,6 @@ import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.common.TbProtoQueueMsg; import org.thingsboard.server.gen.js.JsInvokeProtos; -import org.thingsboard.server.kafka.TbNodeIdProvider; import javax.annotation.Nullable; import javax.annotation.PostConstruct; @@ -45,12 +43,6 @@ import java.util.concurrent.atomic.AtomicInteger; @Service public class RemoteJsInvokeService extends AbstractJsInvokeService { - @Autowired - private TbNodeIdProvider nodeIdProvider; - - @Autowired - private TbKafkaSettings kafkaSettings; - @Value("${js.remote.request_topic}") private String requestTopic; diff --git a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java index c1b0608d44..fd8ee406c2 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/DefaultTbCoreToTransportService.java @@ -1,3 +1,18 @@ +/** + * 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.service.transport; import lombok.extern.slf4j.Slf4j; @@ -15,6 +30,8 @@ import org.thingsboard.server.provider.TbCoreQueueProvider; import java.util.UUID; import java.util.function.Consumer; +import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; + @Slf4j @Service @ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core')") @@ -42,7 +59,7 @@ public class DefaultTbCoreToTransportService implements TbCoreToTransportService UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()); ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setToDeviceSessionMsg(msg).build(); log.trace("[{}][{}] Pushing session data to topic: {}", topic, sessionId, transportMsg); - TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(sessionId, transportMsg); + TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(NULL_UUID, transportMsg); tbTransportProducer.send(topic, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); } diff --git a/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java b/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java index 980f5e6964..eed2b8fa8b 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/RemoteRuleEngineTransportService.java @@ -41,9 +41,8 @@ import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTranspo import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; -import org.thingsboard.server.kafka.TbNodeIdProvider; +import org.thingsboard.server.provider.TbCoreQueueProvider; import org.thingsboard.server.service.cluster.routing.ClusterRoutingService; -import org.thingsboard.server.service.cluster.rpc.ClusterRpcService; import org.thingsboard.server.service.encoding.DataDecodingEncodingService; import org.thingsboard.server.service.queue.TbCoreConsumerStats; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; @@ -65,7 +64,7 @@ import java.util.function.Consumer; @Slf4j @Service @ConditionalOnProperty(prefix = "transport", value = "type", havingValue = "remote") -public class RemoteRuleEngineTransportService implements RuleEngineTransportService { +public class RemoteRuleEngineTransportService { @Value("${transport.remote.rule_engine.topic}") private String ruleEngineTopic; @@ -85,12 +84,6 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ @Value("${transport.remote.rule_engine.stats.enabled:false}") private boolean statsEnabled; -// @Autowired -// private TbKafkaSettings kafkaSettings; -// - @Autowired - private TbNodeIdProvider nodeIdProvider; - @Autowired private ActorSystemContext actorContext; @@ -102,6 +95,9 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ @Autowired private DataDecodingEncodingService encodingService; + @Autowired + private TbCoreQueueProvider coreQueueProvider; + private TbQueueConsumer> ruleEngineConsumer; private TbQueueProducer> notificationsProducer; @@ -114,26 +110,12 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ @PostConstruct public void init() { - TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder notificationsProducerBuilder = TBKafkaProducerTemplate.builder(); - notificationsProducerBuilder.settings(kafkaSettings); - notificationsProducerBuilder.clientId("producer-transport-notification-" + nodeIdProvider.getNodeId()); - notificationsProducerBuilder.encoder(new ToTransportMsgEncoder()); - - notificationsProducer = notificationsProducerBuilder.build(); + notificationsProducer = coreQueueProvider.getTransportMsgProducer(); notificationsProducer.init(); - TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder ruleEngineConsumerBuilder = TBKafkaConsumerTemplate.builder(); - ruleEngineConsumerBuilder.settings(kafkaSettings); - ruleEngineConsumerBuilder.topic(ruleEngineTopic); - ruleEngineConsumerBuilder.clientId("transport-" + nodeIdProvider.getNodeId()); - ruleEngineConsumerBuilder.groupId("tb-node"); - ruleEngineConsumerBuilder.autoCommit(true); - ruleEngineConsumerBuilder.autoCommitIntervalMs(autoCommitInterval); - ruleEngineConsumerBuilder.maxPollRecords(pollRecordsPackSize); - ruleEngineConsumerBuilder.decoder(new ToRuleEngineMsgDecoder()); - - ruleEngineConsumer = ruleEngineConsumerBuilder.build(); - ruleEngineConsumer.subscribe(); + //TODO: 2.5 +// ruleEngineConsumer = +// ruleEngineConsumer.subscribe(); } @EventListener(ApplicationReadyEvent.class) @@ -198,6 +180,7 @@ public class RemoteRuleEngineTransportService implements RuleEngineTransportServ UUID sessionId = new UUID(msg.getSessionIdMSB(), msg.getSessionIdLSB()); ToTransportMsg transportMsg = ToTransportMsg.newBuilder().setToDeviceSessionMsg(msg).build(); log.trace("[{}][{}] Pushing session data to topic: {}", topic, sessionId, transportMsg); + //TODO: 2.5 id TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(sessionId, transportMsg); notificationsProducer.send(topic, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); } diff --git a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java index 354252642f..4765737430 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java @@ -5,7 +5,7 @@ * 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 + * 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, diff --git a/application/src/main/java/org/thingsboard/server/service/transport/TbCoreToTransportService.java b/application/src/main/java/org/thingsboard/server/service/transport/TbCoreToTransportService.java index 3324f02440..5b11edb5f0 100644 --- a/application/src/main/java/org/thingsboard/server/service/transport/TbCoreToTransportService.java +++ b/application/src/main/java/org/thingsboard/server/service/transport/TbCoreToTransportService.java @@ -1,3 +1,18 @@ +/** + * 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.service.transport; import org.thingsboard.server.gen.transport.TransportProtos; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueAdmin.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueAdmin.java index 7f8cff5f22..6917c23719 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueAdmin.java @@ -1,3 +1,18 @@ +/** + * 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; import com.google.common.util.concurrent.ListenableFuture; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueCallback.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueCallback.java index 3d7d791ae4..e823d2a5fc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueCallback.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueCallback.java @@ -1,3 +1,18 @@ +/** + * 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; public interface TbQueueCallback { diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java index ca51eb4ddb..ddf9d7d9b3 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueConsumer.java @@ -1,3 +1,18 @@ +/** + * 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; import java.util.List; diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestIdExtractor.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueCoreSettings.java similarity index 64% rename from common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestIdExtractor.java rename to common/queue/src/main/java/org/thingsboard/server/TbQueueCoreSettings.java index 6e433e71a9..dc84625634 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestIdExtractor.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueCoreSettings.java @@ -13,12 +13,19 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.kafka; +package org.thingsboard.server; -import java.util.UUID; +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; -public interface TbKafkaRequestIdExtractor { +@Data +@Component +public class TbQueueCoreSettings { - UUID extractRequestId(T value); + @Value("${queue.core.topic}") + private String topic; + @Value("${queue.core.partitions}") + private int partitions; } diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsg.java index 22714af77e..ca8f3a61da 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsg.java @@ -1,3 +1,18 @@ +/** + * 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; import java.util.UUID; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgHeaders.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgHeaders.java index 9dad87588d..f95454c976 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgHeaders.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgHeaders.java @@ -1,3 +1,18 @@ +/** + * 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; import java.util.Map; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgMetadata.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgMetadata.java index 8eecae51fb..a3331999a5 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgMetadata.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueMsgMetadata.java @@ -1,3 +1,18 @@ +/** + * 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; public interface TbQueueMsgMetadata { diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java index 41d165d743..206f753168 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueProducer.java @@ -1,15 +1,28 @@ +/** + * 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; -import com.google.common.util.concurrent.ListenableFuture; - public interface TbQueueProducer { void init(); String getDefaultTopic(); - ListenableFuture send(T msg, TbQueueCallback callback); + void send(T msg, TbQueueCallback callback); - ListenableFuture send(String topic, T msg, TbQueueCallback callback); + void send(String topic, T msg, TbQueueCallback callback); } diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java index 5ba28f22c5..5182bb7260 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueRequestTemplate.java @@ -1,3 +1,18 @@ +/** + * 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; import com.google.common.util.concurrent.ListenableFuture; diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueResponseTemplate.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueResponseTemplate.java index f3b4e4ad5d..686570ca32 100644 --- a/common/queue/src/main/java/org/thingsboard/server/TbQueueResponseTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueResponseTemplate.java @@ -1,3 +1,18 @@ +/** + * 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; public interface TbQueueResponseTemplate { diff --git a/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java new file mode 100644 index 0000000000..603d787fa8 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/TbQueueTransportApiSettings.java @@ -0,0 +1,40 @@ +/** + * 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; + +import lombok.Data; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +@Data +@Component +public class TbQueueTransportApiSettings { + @Value("${queue.transport_api.requests_topic}") + private String requestsTopic; + + @Value("${queue.transport_api.responses_topic}") + private String responsesTopic; + + @Value("${queue.transport_api.max_pending_requests}") + private int maxPendingRequests; + + @Value("${queue.transport_api.max_requests_timeout}") + private int maxRequestsTimeout; + + @Value("${queue.transport_api.response_poll_interval}") + private long responsePollInterval; + +} diff --git a/common/queue/src/main/java/org/thingsboard/server/common/AbstractTbQueueTemplate.java b/common/queue/src/main/java/org/thingsboard/server/common/AbstractTbQueueTemplate.java index 2f406a9631..b70fc417c8 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/AbstractTbQueueTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/AbstractTbQueueTemplate.java @@ -1,3 +1,18 @@ +/** + * 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; import java.nio.ByteBuffer; @@ -7,6 +22,7 @@ import java.util.UUID; public class AbstractTbQueueTemplate { protected static final String REQUEST_ID_HEADER = "requestId"; protected static final String RESPONSE_TOPIC_HEADER = "responseTopic"; + protected static final String REQUEST_TIME = "requestTime"; protected byte[] uuidToBytes(UUID uuid) { ByteBuffer buf = ByteBuffer.allocate(16); @@ -29,4 +45,17 @@ public class AbstractTbQueueTemplate { protected String bytesToString(byte[] data) { return new String(data, StandardCharsets.UTF_8); } + + private static ByteBuffer longBuffer = ByteBuffer.allocate(Long.BYTES); + + protected static byte[] longToBytes(long x) { + longBuffer.putLong(0, x); + return longBuffer.array(); + } + + protected static long bytesToLong(byte[] bytes) { + longBuffer.put(bytes, 0, bytes.length); + longBuffer.flip();//need flip + return longBuffer.getLong(); + } } diff --git a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueMsgHeaders.java b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueMsgHeaders.java index e547a6c070..de0c368a65 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueMsgHeaders.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueMsgHeaders.java @@ -1,3 +1,18 @@ +/** + * 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; import org.thingsboard.server.TbQueueMsgHeaders; diff --git a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java index 49022ad2e3..d022dafadc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueRequestTemplate.java @@ -1,3 +1,18 @@ +/** + * 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; import com.google.common.util.concurrent.Futures; @@ -79,7 +94,7 @@ public class DefaultTbQueueRequestTemplate { log.trace("Received response to Kafka Template request: {}", response); byte[] requestIdHeader = response.getHeaders().get(REQUEST_ID_HEADER); - UUID requestId = null; + UUID requestId; if (requestIdHeader == null) { log.error("[{}] Missing requestId in header and body", response); } else { @@ -140,6 +155,7 @@ public class DefaultTbQueueRequestTemplate future = SettableFuture.create(); ResponseMetaData responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future); pendingRequests.putIfAbsent(requestId, responseMetaData); diff --git a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueResponseTemplate.java b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueResponseTemplate.java index 92dac0551b..3804fb5160 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueResponseTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/DefaultTbQueueResponseTemplate.java @@ -1,3 +1,18 @@ +/** + * 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; import lombok.Builder; @@ -70,41 +85,46 @@ public class DefaultTbQueueResponseTemplate requests = requestTemplate.poll(pollInterval); + requests.forEach(request -> { - byte[] requestIdHeader = request.getHeaders().get(REQUEST_ID_HEADER); - if (requestIdHeader == null) { - log.error("[{}] Missing requestId in header", request); - return; - } - byte[] responseTopicHeader = request.getHeaders().get(RESPONSE_TOPIC_HEADER); - if (responseTopicHeader == null) { - log.error("[{}] Missing response topic in header", request); - return; - } - UUID requestId = bytesToUuid(requestIdHeader); - String responseTopic = bytesToString(responseTopicHeader); - try { - pendingRequestCount.getAndIncrement(); - AsyncCallbackTemplate.withCallbackAndTimeout(handler.handle(request), - response -> { - pendingRequestCount.decrementAndGet(); - response.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId)); - responseTemplate.send(responseTopic, response, null); - }, - e -> { - pendingRequestCount.decrementAndGet(); - if (e.getCause() != null && e.getCause() instanceof TimeoutException) { - log.warn("[{}] Timeout to process the request: {}", requestId, request, e); - } else { - log.trace("[{}] Failed to process the request: {}", requestId, request, e); - } - }, - requestTimeout, - timeoutExecutor, - callbackExecutor); - } catch (Throwable e) { - pendingRequestCount.decrementAndGet(); - log.warn("[{}] Failed to process the request: {}", requestId, request, e); + long currentTime = System.currentTimeMillis(); + long requestTime = bytesToLong(request.getHeaders().get(REQUEST_TIME)); + if (requestTime + requestTimeout >= currentTime) { + byte[] requestIdHeader = request.getHeaders().get(REQUEST_ID_HEADER); + if (requestIdHeader == null) { + log.error("[{}] Missing requestId in header", request); + return; + } + byte[] responseTopicHeader = request.getHeaders().get(RESPONSE_TOPIC_HEADER); + if (responseTopicHeader == null) { + log.error("[{}] Missing response topic in header", request); + return; + } + UUID requestId = bytesToUuid(requestIdHeader); + String responseTopic = bytesToString(responseTopicHeader); + try { + pendingRequestCount.getAndIncrement(); + AsyncCallbackTemplate.withCallbackAndTimeout(handler.handle(request), + response -> { + pendingRequestCount.decrementAndGet(); + response.getHeaders().put(REQUEST_ID_HEADER, uuidToBytes(requestId)); + responseTemplate.send(responseTopic, response, null); + }, + e -> { + pendingRequestCount.decrementAndGet(); + if (e.getCause() != null && e.getCause() instanceof TimeoutException) { + log.warn("[{}] Timeout to process the request: {}", requestId, request, e); + } else { + log.trace("[{}] Failed to process the request: {}", requestId, request, e); + } + }, + requestTimeout, + timeoutExecutor, + callbackExecutor); + } catch (Throwable e) { + pendingRequestCount.decrementAndGet(); + log.warn("[{}] Failed to process the request: {}", requestId, request, e); + } } }); requestTemplate.commit(); diff --git a/common/queue/src/main/java/org/thingsboard/server/common/TbProtoQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/common/TbProtoQueueMsg.java index c072dc4d67..8dd492504c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/common/TbProtoQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/common/TbProtoQueueMsg.java @@ -1,3 +1,18 @@ +/** + * 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; import lombok.Data; diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/AbstractTbKafkaTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/AbstractTbKafkaTemplate.java deleted file mode 100644 index 0c68c8dd5d..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/AbstractTbKafkaTemplate.java +++ /dev/null @@ -1,50 +0,0 @@ -/** - * 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.kafka; - -import lombok.extern.slf4j.Slf4j; - -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; -import java.util.UUID; - -/** - * Created by ashvayka on 25.09.18. - */ -@Slf4j -public abstract class AbstractTbKafkaTemplate { - protected byte[] uuidToBytes(UUID uuid) { - ByteBuffer buf = ByteBuffer.allocate(16); - buf.putLong(uuid.getMostSignificantBits()); - buf.putLong(uuid.getLeastSignificantBits()); - return buf.array(); - } - - protected static UUID bytesToUuid(byte[] bytes) { - ByteBuffer bb = ByteBuffer.wrap(bytes); - long firstLong = bb.getLong(); - long secondLong = bb.getLong(); - return new UUID(firstLong, secondLong); - } - - protected byte[] stringToBytes(String string) { - return string.getBytes(StandardCharsets.UTF_8); - } - - protected String bytesToString(byte[] data) { - return new String(data, StandardCharsets.UTF_8); - } -} diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/AsyncCallbackTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/AsyncCallbackTemplate.java deleted file mode 100644 index 17599bfccb..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/AsyncCallbackTemplate.java +++ /dev/null @@ -1,66 +0,0 @@ -/** - * 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.kafka; - -import com.google.common.util.concurrent.FutureCallback; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; - -import java.util.concurrent.Executor; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.function.Consumer; - -/** - * Created by ashvayka on 05.10.18. - */ -public class AsyncCallbackTemplate { - - public static void withCallbackAndTimeout(ListenableFuture future, - Consumer onSuccess, - Consumer onFailure, - long timeoutInMs, - ScheduledExecutorService timeoutExecutor, - Executor callbackExecutor) { - future = Futures.withTimeout(future, timeoutInMs, TimeUnit.MILLISECONDS, timeoutExecutor); - withCallback(future, onSuccess, onFailure, callbackExecutor); - } - - public static void withCallback(ListenableFuture future, Consumer onSuccess, - Consumer onFailure, Executor executor) { - FutureCallback callback = new FutureCallback() { - @Override - public void onSuccess(T result) { - try { - onSuccess.accept(result); - } catch (Throwable th) { - onFailure(th); - } - } - - @Override - public void onFailure(Throwable t) { - onFailure.accept(t); - } - }; - if (executor != null) { - Futures.addCallback(future, callback, executor); - } else { - Futures.addCallback(future, callback); - } - } - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsg.java b/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsg.java index e44a93a70d..9f8e73bdfc 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsg.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsg.java @@ -1,3 +1,18 @@ +/** + * 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.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsgMetadata.java b/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsgMetadata.java index 4698c226d6..09cc292deb 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsgMetadata.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/KafkaTbQueueMsgMetadata.java @@ -1,3 +1,18 @@ +/** + * 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.kafka; import lombok.AllArgsConstructor; diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaAdmin.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaAdmin.java index 9c6d911d33..0bb3a75db6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaAdmin.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaAdmin.java @@ -15,16 +15,17 @@ */ package org.thingsboard.server.kafka; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.JdkFutureAdapters; import com.google.common.util.concurrent.ListenableFuture; -import org.apache.kafka.clients.admin.*; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.admin.AdminClient; +import org.apache.kafka.clients.admin.CreateTopicsResult; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.admin.TopicDescription; import org.apache.kafka.common.KafkaFuture; import org.thingsboard.server.TbQueueAdmin; -import java.time.Duration; import java.util.Collections; -import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -40,6 +41,19 @@ public class TBKafkaAdmin implements TbQueueAdmin { client = AdminClient.create(settings.toProps()); } + @Override + public ListenableFuture createTopicIfNotExists(String topic) { + + KafkaFuture topicDescriptionFuture = client.describeTopics(Collections.singleton(topic)).values().get(topic); + + ListenableFuture topicFuture = JdkFutureAdapters.listenInPoolThread(topicDescriptionFuture); + + return Futures.transformAsync(topicFuture, topicDescription -> { + KafkaFuture resultFuture = createTopic(new NewTopic(topic, 1, (short) 1)).values().get(topic); + return JdkFutureAdapters.listenInPoolThread(resultFuture); + }); + } + public void waitForTopic(String topic, long timeout, TimeUnit timeoutUnit) throws InterruptedException, TimeoutException { synchronized (this) { long timeoutExpiredMs = System.currentTimeMillis() + timeoutUnit.toMillis(timeout); @@ -54,7 +68,7 @@ public class TBKafkaAdmin implements TbQueueAdmin { } } - public CreateTopicsResult createTopic(NewTopic topic){ + public CreateTopicsResult createTopic(NewTopic topic) { return client.createTopics(Collections.singletonList(topic)); } @@ -67,10 +81,4 @@ public class TBKafkaAdmin implements TbQueueAdmin { return false; } } - - @Override - public ListenableFuture createTopicIfNotExists(String topic) { - - return null; - } } diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java index 6f0ce7b55c..558d556d09 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaConsumerTemplate.java @@ -31,7 +31,6 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Properties; -import java.util.UUID; /** * Created by ashvayka on 24.09.18. @@ -42,15 +41,11 @@ public class TBKafkaConsumerTemplate implements TbQueueCon private final KafkaConsumer consumer; private final TbKafkaDecoder decoder; - @Builder.Default - private TbKafkaRequestIdExtractor requestIdExtractor = ((response) -> null); - @Getter private final String topic; @Builder private TBKafkaConsumerTemplate(TbKafkaSettings settings, TbKafkaDecoder decoder, - TbKafkaRequestIdExtractor requestIdExtractor, String clientId, String groupId, String topic, boolean autoCommit, int autoCommitIntervalMs, int maxPollRecords) { @@ -68,7 +63,6 @@ public class TBKafkaConsumerTemplate implements TbQueueCon } this.consumer = new KafkaConsumer<>(props); this.decoder = decoder; - this.requestIdExtractor = requestIdExtractor; this.topic = topic; } @@ -105,22 +99,8 @@ public class TBKafkaConsumerTemplate implements TbQueueCon consumer.unsubscribe(); } -// public void subscribe() { -// consumer.subscribe(Collections.singletonList(topic)); -// } -// - -// -// public ConsumerRecords poll(Duration duration) { -// return consumer.poll(duration); -// } - public T decode(ConsumerRecord record) throws IOException { return decoder.decode(new KafkaTbQueueMsg(record)); } - public UUID extractRequestId(T value) { - return requestIdExtractor.extractRequestId(value); - } - } diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java index 938b278207..d76a2dc625 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TBKafkaProducerTemplate.java @@ -15,9 +15,6 @@ */ package org.thingsboard.server.kafka; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.JdkFutureAdapters; -import com.google.common.util.concurrent.ListenableFuture; import lombok.Builder; import lombok.Getter; import lombok.extern.slf4j.Slf4j; @@ -31,7 +28,6 @@ import org.apache.kafka.common.header.internals.RecordHeader; import org.springframework.util.StringUtils; import org.thingsboard.server.TbQueueCallback; import org.thingsboard.server.TbQueueMsg; -import org.thingsboard.server.TbQueueMsgMetadata; import org.thingsboard.server.TbQueueProducer; import java.util.List; @@ -91,12 +87,12 @@ public class TBKafkaProducerTemplate implements TbQueuePro } @Override - public ListenableFuture send(T msg, TbQueueCallback callback) { - return send(defaultTopic, msg, callback); + public void send(T msg, TbQueueCallback callback) { + send(defaultTopic, msg, callback); } @Override - public ListenableFuture send(String topic, T msg, TbQueueCallback callback) { + public void send(String topic, T msg, TbQueueCallback callback) { String key = msg.getKey().toString(); byte[] data = msg.getData(); ProducerRecord record; @@ -111,42 +107,8 @@ public class TBKafkaProducerTemplate implements TbQueuePro callback.onFailure(exception); } }); - - return Futures.transform(JdkFutureAdapters.listenInPoolThread(result), metadata -> new KafkaTbQueueMsgMetadata(metadata)); } -// public Future send(String key, T value, Callback callback) { -// return send(key, value, null, callback); -// } -// -// public Future send(String key, T value, Iterable
headers, Callback callback) { -// return send(key, value, null, headers, callback); -// } -// -// public Future send(String key, T value, Long timestamp, Iterable
headers, Callback callback) { -// if (!StringUtils.isEmpty(this.defaultTopic)) { -// return send(this.defaultTopic, key, value, timestamp, headers, callback); -// } else { -// throw new RuntimeException("Failed to send message! Default topic is not specified!"); -// } -// } -// -// public Future send(String topic, String key, T value, Iterable
headers, Callback callback) { -// return send(topic, key, value, null, headers, callback); -// } -// -// public Future send(String topic, String key, T value, Callback callback) { -// return send(topic, key, value, null, null, callback); -// } -// -// public Future send(String topic, String key, T value, Long timestamp, Iterable
headers, Callback callback) { -// byte[] data = encoder.encode(value); -// ProducerRecord record; -// Integer partition = getPartition(topic, key, value, data); -// record = new ProducerRecord<>(topic, partition, timestamp, key, data, headers); -// return producer.send(record, callback); -// } - private Integer getPartition(String topic, T value) { if (partitioner == null) { return null; diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java deleted file mode 100644 index 2a7bca119f..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaRequestTemplate.java +++ /dev/null @@ -1,213 +0,0 @@ -/** - * 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.kafka; - -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.SettableFuture; -import lombok.Builder; -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.admin.CreateTopicsResult; -import org.apache.kafka.clients.admin.NewTopic; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.clients.producer.Callback; -import org.apache.kafka.clients.producer.RecordMetadata; -import org.apache.kafka.common.errors.InterruptException; -import org.apache.kafka.common.errors.TopicExistsException; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.internals.RecordHeader; - -import java.io.IOException; -import java.time.Duration; -import java.util.ArrayList; -import java.util.List; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.TimeoutException; - -/** - * Created by ashvayka on 25.09.18. - */ -@Slf4j -public class TbKafkaRequestTemplate extends AbstractTbKafkaTemplate { - - private final TBKafkaProducerTemplate requestTemplate; - private final TBKafkaConsumerTemplate responseTemplate; - private final ConcurrentMap> pendingRequests; - private final boolean internalExecutor; - private final ExecutorService executor; - private final long maxRequestTimeout; - private final long maxPendingRequests; - private final long pollInterval; - private volatile long tickTs = 0L; - private volatile long tickSize = 0L; - private volatile boolean stopped = false; - - @Builder - public TbKafkaRequestTemplate(TBKafkaProducerTemplate requestTemplate, - TBKafkaConsumerTemplate responseTemplate, - long maxRequestTimeout, - long maxPendingRequests, - long pollInterval, - ExecutorService executor) { - this.requestTemplate = requestTemplate; - this.responseTemplate = responseTemplate; - this.pendingRequests = new ConcurrentHashMap<>(); - this.maxRequestTimeout = maxRequestTimeout; - this.maxPendingRequests = maxPendingRequests; - this.pollInterval = pollInterval; - if (executor != null) { - internalExecutor = false; - this.executor = executor; - } else { - internalExecutor = true; - this.executor = Executors.newSingleThreadExecutor(); - } - } - - public void init() { - try { - TBKafkaAdmin admin = new TBKafkaAdmin(this.requestTemplate.getSettings()); - CreateTopicsResult result = admin.createTopic(new NewTopic(responseTemplate.getTopic(), 1, (short) 1)); - result.all().get(); - } catch (Exception e) { - if ((e instanceof TopicExistsException) || (e.getCause() != null && e.getCause() instanceof TopicExistsException)) { - log.trace("[{}] Topic already exists. ", responseTemplate.getTopic()); - } else { - log.info("[{}] Failed to create topic: {}", responseTemplate.getTopic(), e.getMessage(), e); - throw new RuntimeException(e); - } - - } - this.requestTemplate.init(); - tickTs = System.currentTimeMillis(); - responseTemplate.subscribe(); - executor.submit(() -> { - long nextCleanupMs = 0L; - while (!stopped) { - try { - ConsumerRecords responses = responseTemplate.poll(Duration.ofMillis(pollInterval)); - if (responses.count() > 0) { - log.trace("Polling responses completed, consumer records count [{}]", responses.count()); - } - responses.forEach(response -> { - log.trace("Received response to Kafka Template request: {}", response); - Header requestIdHeader = response.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER); - Response decodedResponse = null; - UUID requestId = null; - if (requestIdHeader == null) { - try { - decodedResponse = responseTemplate.decode(response); - requestId = responseTemplate.extractRequestId(decodedResponse); - } catch (IOException e) { - log.error("Failed to decode response", e); - } - } else { - requestId = bytesToUuid(requestIdHeader.value()); - } - if (requestId == null) { - log.error("[{}] Missing requestId in header and body", response); - } else { - log.trace("[{}] Response received", requestId); - ResponseMetaData expectedResponse = pendingRequests.remove(requestId); - if (expectedResponse == null) { - log.trace("[{}] Invalid or stale request", requestId); - } else { - try { - if (decodedResponse == null) { - decodedResponse = responseTemplate.decode(response); - } - expectedResponse.future.set(decodedResponse); - } catch (IOException e) { - expectedResponse.future.setException(e); - } - } - } - }); - tickTs = System.currentTimeMillis(); - tickSize = pendingRequests.size(); - if (nextCleanupMs < tickTs) { - //cleanup; - pendingRequests.forEach((key, value) -> { - if (value.expTime < tickTs) { - ResponseMetaData staleRequest = pendingRequests.remove(key); - if (staleRequest != null) { - log.trace("[{}] Request timeout detected, expTime [{}], tickTs [{}]", key, staleRequest.expTime, tickTs); - staleRequest.future.setException(new TimeoutException()); - } - } - }); - nextCleanupMs = tickTs + maxRequestTimeout; - } - } catch (InterruptException ie) { - if (!stopped) { - log.warn("Fetching data from kafka was interrupted.", ie); - } - } catch (Throwable e) { - log.warn("Failed to obtain responses from queue.", e); - try { - Thread.sleep(pollInterval); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new responses", e2); - } - } - } - }); - } - - public void stop() { - stopped = true; - if (internalExecutor) { - executor.shutdownNow(); - } - } - - public ListenableFuture post(String key, Request request) { - if (tickSize > maxPendingRequests) { - return Futures.immediateFailedFuture(new RuntimeException("Pending request map is full!")); - } - UUID requestId = UUID.randomUUID(); - List
headers = new ArrayList<>(2); - headers.add(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId))); - headers.add(new RecordHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER, stringToBytes(responseTemplate.getTopic()))); - SettableFuture future = SettableFuture.create(); - ResponseMetaData responseMetaData = new ResponseMetaData<>(tickTs + maxRequestTimeout, future); - pendingRequests.putIfAbsent(requestId, responseMetaData); - log.trace("[{}] Sending request, key [{}], expTime [{}]", requestId, key, responseMetaData.expTime); - requestTemplate.send(key, request, headers, (metadata, exception) -> { - if (exception != null) { - log.trace("[{}] Failed to post the request", requestId, exception); - } else { - log.trace("[{}] Posted the request: {}", requestId, metadata); - } - }); - return future; - } - - private static class ResponseMetaData { - private final long expTime; - private final SettableFuture future; - - ResponseMetaData(long ts, SettableFuture future) { - this.expTime = ts; - this.future = future; - } - } - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaResponseTemplate.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaResponseTemplate.java deleted file mode 100644 index 864182a552..0000000000 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaResponseTemplate.java +++ /dev/null @@ -1,161 +0,0 @@ -/** - * 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.kafka; - -import lombok.Builder; -import lombok.extern.slf4j.Slf4j; -import org.apache.kafka.clients.consumer.ConsumerRecords; -import org.apache.kafka.common.errors.InterruptException; -import org.apache.kafka.common.header.Header; -import org.apache.kafka.common.header.internals.RecordHeader; - -import java.time.Duration; -import java.util.Collections; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * Created by ashvayka on 25.09.18. - */ -@Slf4j -public class TbKafkaResponseTemplate extends AbstractTbKafkaTemplate { - - private final TBKafkaConsumerTemplate requestTemplate; - private final TBKafkaProducerTemplate responseTemplate; - private final TbKafkaHandler handler; - private final ConcurrentMap pendingRequests; - private final ExecutorService loopExecutor; - private final ScheduledExecutorService timeoutExecutor; - private final ExecutorService callbackExecutor; - private final int maxPendingRequests; - private final long requestTimeout; - - private final long pollInterval; - private volatile boolean stopped = false; - private final AtomicInteger pendingRequestCount = new AtomicInteger(); - - @Builder - public TbKafkaResponseTemplate(TBKafkaConsumerTemplate requestTemplate, - TBKafkaProducerTemplate responseTemplate, - TbKafkaHandler handler, - long pollInterval, - long requestTimeout, - int maxPendingRequests, - ExecutorService executor) { - this.requestTemplate = requestTemplate; - this.responseTemplate = responseTemplate; - this.handler = handler; - this.pendingRequests = new ConcurrentHashMap<>(); - this.maxPendingRequests = maxPendingRequests; - this.pollInterval = pollInterval; - this.requestTimeout = requestTimeout; - this.callbackExecutor = executor; - this.timeoutExecutor = Executors.newSingleThreadScheduledExecutor(); - this.loopExecutor = Executors.newSingleThreadExecutor(); - } - - public void init() { - this.responseTemplate.init(); - requestTemplate.subscribe(); - loopExecutor.submit(() -> { - while (!stopped) { - try { - while (pendingRequestCount.get() >= maxPendingRequests) { - try { - Thread.sleep(pollInterval); - } catch (InterruptedException e) { - log.trace("Failed to wait until the server has capacity to handle new requests", e); - } - } - ConsumerRecords requests = requestTemplate.poll(Duration.ofMillis(pollInterval)); - requests.forEach(request -> { - Header requestIdHeader = request.headers().lastHeader(TbKafkaSettings.REQUEST_ID_HEADER); - if (requestIdHeader == null) { - log.error("[{}] Missing requestId in header", request); - return; - } - UUID requestId = bytesToUuid(requestIdHeader.value()); - if (requestId == null) { - log.error("[{}] Missing requestId in header and body", request); - return; - } - Header responseTopicHeader = request.headers().lastHeader(TbKafkaSettings.RESPONSE_TOPIC_HEADER); - if (responseTopicHeader == null) { - log.error("[{}] Missing response topic in header", request); - return; - } - String responseTopic = bytesToString(responseTopicHeader.value()); - try { - pendingRequestCount.getAndIncrement(); - Request decodedRequest = requestTemplate.decode(request); - AsyncCallbackTemplate.withCallbackAndTimeout(handler.handle(decodedRequest), - response -> { - pendingRequestCount.decrementAndGet(); - reply(requestId, responseTopic, response); - }, - e -> { - pendingRequestCount.decrementAndGet(); - if (e.getCause() != null && e.getCause() instanceof TimeoutException) { - log.warn("[{}] Timedout to process the request: {}", requestId, request, e); - } else { - log.trace("[{}] Failed to process the request: {}", requestId, request, e); - } - }, - requestTimeout, - timeoutExecutor, - callbackExecutor); - } catch (Throwable e) { - pendingRequestCount.decrementAndGet(); - log.warn("[{}] Failed to process the request: {}", requestId, request, e); - } - }); - } catch (InterruptException ie) { - if (!stopped) { - log.warn("Fetching data from kafka was interrupted.", ie); - } - } catch (Throwable e) { - log.warn("Failed to obtain messages from queue.", e); - try { - Thread.sleep(pollInterval); - } catch (InterruptedException e2) { - log.trace("Failed to wait until the server has capacity to handle new requests", e2); - } - } - } - }); - } - - public void stop() { - stopped = true; - if (timeoutExecutor != null) { - timeoutExecutor.shutdownNow(); - } - if (loopExecutor != null) { - loopExecutor.shutdownNow(); - } - } - - private void reply(UUID requestId, String topic, Response response) { - responseTemplate.send(topic, requestId.toString(), response, Collections.singletonList(new RecordHeader(TbKafkaSettings.REQUEST_ID_HEADER, uuidToBytes(requestId))), null); - } - -} diff --git a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java index 6446643a18..566766fee6 100644 --- a/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java +++ b/common/queue/src/main/java/org/thingsboard/server/kafka/TbKafkaSettings.java @@ -35,22 +35,22 @@ public class TbKafkaSettings { static final String REQUEST_ID_HEADER = "requestId"; static final String RESPONSE_TOPIC_HEADER = "responseTopic"; - @Value("${kafka.bootstrap.servers}") + @Value("${queue.kafka.bootstrap.servers}") private String servers; - @Value("${kafka.acks}") + @Value("${queue.kafka.acks}") private String acks; - @Value("${kafka.retries}") + @Value("${queue.queue.kafka.retries}") private int retries; - @Value("${kafka.batch.size}") + @Value("${queue.kafka.batch.size}") private int batchSize; - @Value("${kafka.linger.ms}") + @Value("${queue.kafka.linger.ms}") private long lingerMs; - @Value("${kafka.buffer.memory}") + @Value("${queue.kafka.buffer.memory}") private long bufferMemory; @Value("${kafka.other:#{null}}") diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java index abf30667b5..ded4cd9810 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryStorage.java @@ -1,3 +1,18 @@ +/** + * 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.memory; import org.thingsboard.server.TbQueueMsg; diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java index a0a35eefb8..b8ddcff89c 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueConsumer.java @@ -1,3 +1,18 @@ +/** + * 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.memory; import org.thingsboard.server.TbQueueConsumer; diff --git a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java index 32b9399164..e71d579c7f 100644 --- a/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java +++ b/common/queue/src/main/java/org/thingsboard/server/memory/InMemoryTbQueueProducer.java @@ -1,11 +1,23 @@ +/** + * 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.memory; -import com.google.common.util.concurrent.Futures; -import com.google.common.util.concurrent.ListenableFuture; import lombok.Data; import org.thingsboard.server.TbQueueCallback; import org.thingsboard.server.TbQueueMsg; -import org.thingsboard.server.TbQueueMsgMetadata; import org.thingsboard.server.TbQueueProducer; @Data @@ -15,6 +27,10 @@ public class InMemoryTbQueueProducer implements TbQueuePro private final String defaultTopic; + public InMemoryTbQueueProducer(String defaultTopic) { + this.defaultTopic = defaultTopic; + } + @Override public void init() { @@ -26,20 +42,18 @@ public class InMemoryTbQueueProducer implements TbQueuePro } @Override - public ListenableFuture send(T msg, TbQueueCallback callback) { - return send(defaultTopic, msg, callback); + public void send(T msg, TbQueueCallback callback) { + send(defaultTopic, msg, callback); } @Override - public ListenableFuture send(String topic, T msg, TbQueueCallback callback) { + public void send(String topic, T msg, TbQueueCallback callback) { boolean result = storage.put(topic, msg); if (result) { callback.onSuccess(null); - return Futures.immediateCheckedFuture(null); } else { Exception e = new RuntimeException("Failure add msg to InMemoryQueue"); callback.onFailure(e); - return Futures.immediateFailedFuture(e); } } } diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTbCoreQueueProvider.java new file mode 100644 index 0000000000..79f5dd7da0 --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTbCoreQueueProvider.java @@ -0,0 +1,79 @@ +/** + * 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.provider; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; +import org.thingsboard.server.TbQueueConsumer; +import org.thingsboard.server.TbQueueCoreSettings; +import org.thingsboard.server.TbQueueProducer; +import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.memory.InMemoryTbQueueConsumer; +import org.thingsboard.server.memory.InMemoryTbQueueProducer; + +@Slf4j +@Component +@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && '${queue.type:null}'=='in-memory'") +public class InMemoryTbCoreQueueProvider implements TbCoreQueueProvider { + + private final TbQueueCoreSettings coreSettings; + + public InMemoryTbCoreQueueProvider(TbQueueCoreSettings coreSettings) { + this.coreSettings = coreSettings; + } + + @Override + public TbQueueProducer> getTransportMsgProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); + return producer; + } + + @Override + public TbQueueProducer> getRuleEngineMsgProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); + return producer; + } + + @Override + public TbQueueProducer> getTbCoreMsgProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); + return producer; + } + + @Override + public TbQueueConsumer> getToCoreMsgConsumer() { + InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); + return consumer; + } + + @Override + public TbQueueConsumer> getTransportApiRequestConsumer() { + InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(coreSettings.getTopic()); + return consumer; + } + + @Override + public TbQueueProducer> getTransportApiResponseProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(coreSettings.getTopic()); + return producer; + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java new file mode 100644 index 0000000000..789ea1cf5a --- /dev/null +++ b/common/queue/src/main/java/org/thingsboard/server/provider/InMemoryTransportQueueProvider.java @@ -0,0 +1,79 @@ +/** + * 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.provider; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.stereotype.Component; +import org.thingsboard.server.TbQueueConsumer; +import org.thingsboard.server.TbQueueProducer; +import org.thingsboard.server.TbQueueRequestTemplate; +import org.thingsboard.server.TbQueueTransportApiSettings; +import org.thingsboard.server.common.DefaultTbQueueRequestTemplate; +import org.thingsboard.server.common.TbProtoQueueMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.memory.InMemoryTbQueueConsumer; +import org.thingsboard.server.memory.InMemoryTbQueueProducer; + +@Component +@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport') && '${queue.type:null}'=='in-memory'") +@Slf4j +public class InMemoryTransportQueueProvider implements TransportQueueProvider { + + private final TbQueueTransportApiSettings transportApiSettings; + + public InMemoryTransportQueueProvider(TbQueueTransportApiSettings transportApiSettings) { + this.transportApiSettings = transportApiSettings; + } + + @Override + public TbQueueRequestTemplate, TbProtoQueueMsg> getTransportApiRequestTemplate() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); + + InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); + + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder + , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); + templateBuilder.requestTemplate(producer); + templateBuilder.responseTemplate(consumer); + templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); + templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout()); + templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval()); + return templateBuilder.build(); + } + + @Override + public TbQueueProducer> getRuleEngineMsgProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); + return producer; + } + + @Override + public TbQueueProducer> getTbCoreMsgProducer() { + InMemoryTbQueueProducer> producer = new InMemoryTbQueueProducer<>(transportApiSettings.getRequestsTopic()); + return producer; + } + + @Override + public TbQueueConsumer> getTransportNotificationsConsumer() { + InMemoryTbQueueConsumer> consumer = new InMemoryTbQueueConsumer<>(transportApiSettings.getResponsesTopic()); + return consumer; + } +} diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java index 2f9a9ad8f8..d830cff8fa 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTbCoreQueueProvider.java @@ -1,42 +1,102 @@ +/** + * 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.provider; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.stereotype.Component; import org.thingsboard.server.TbQueueConsumer; +import org.thingsboard.server.TbQueueCoreSettings; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.common.TbProtoQueueMsg; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; +import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; +import org.thingsboard.server.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.kafka.TbKafkaSettings; +import org.thingsboard.server.kafka.TbNodeIdProvider; @Component -@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && ${queue.type:null}'=='kafka'") -public class KafkaTbCoreQueueProvider implements TbCoreQueueProvider{ +@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core') && '${queue.type:null}'=='kafka'") +public class KafkaTbCoreQueueProvider implements TbCoreQueueProvider { + + private final TbKafkaSettings kafkaSettings; + private final TbNodeIdProvider nodeIdProvider; + private final TbQueueCoreSettings coreSettings; + + public KafkaTbCoreQueueProvider(TbKafkaSettings kafkaSettings, TbNodeIdProvider nodeIdProvider, TbQueueCoreSettings coreSettings) { + this.kafkaSettings = kafkaSettings; + this.nodeIdProvider = nodeIdProvider; + this.coreSettings = coreSettings; + } + @Override - public TbQueueProducer> getTransportMsgProducer() { - return null; + public TbQueueProducer> getTransportMsgProducer() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(coreSettings.getTopic()); + return requestBuilder.build(); } @Override - public TbQueueProducer> getRuleEngineMsgProducer() { - return null; + public TbQueueProducer> getRuleEngineMsgProducer() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(coreSettings.getTopic()); + return requestBuilder.build(); } @Override - public TbQueueProducer> getTbCoreMsgProducer() { - return null; + public TbQueueProducer> getTbCoreMsgProducer() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(coreSettings.getTopic()); + return requestBuilder.build(); } @Override - public TbQueueConsumer> getToCoreMsgConsumer() { - return null; + public TbQueueConsumer> getToCoreMsgConsumer() { + TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + responseBuilder.settings(kafkaSettings); + responseBuilder.topic(coreSettings.getTopic()); + responseBuilder.clientId("consumer-transport-" + nodeIdProvider.getNodeId()); + responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); + responseBuilder.autoCommit(true); + //TODO: 2.5 +// responseBuilder.autoCommitIntervalMs(autoCommitInterval); +// responseBuilder.decoder(new TransportApiResponseDecoder()); + return responseBuilder.build(); } @Override - public TbQueueConsumer> getTransportApiRequestConsumer() { + public TbQueueConsumer> getTransportApiRequestConsumer() { return null; } @Override - public TbQueueProducer> getTransportApiResponseProducer() { - return null; + public TbQueueProducer> getTransportApiResponseProducer() { + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-core-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(coreSettings.getTopic()); + return requestBuilder.build(); } } diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java index 945888a45a..3b671a3adb 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/KafkaTransportQueueProvider.java @@ -1,3 +1,18 @@ +/** + * 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.provider; import lombok.extern.slf4j.Slf4j; @@ -6,30 +21,77 @@ import org.springframework.stereotype.Component; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueProducer; import org.thingsboard.server.TbQueueRequestTemplate; +import org.thingsboard.server.TbQueueTransportApiSettings; +import org.thingsboard.server.common.DefaultTbQueueRequestTemplate; import org.thingsboard.server.common.TbProtoQueueMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportApiResponseMsg; +import org.thingsboard.server.kafka.TBKafkaConsumerTemplate; +import org.thingsboard.server.kafka.TBKafkaProducerTemplate; +import org.thingsboard.server.kafka.TbKafkaSettings; +import org.thingsboard.server.kafka.TbNodeIdProvider; @Component -@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport') && ${queue.type:null}'=='kafka'") +@ConditionalOnExpression("('${service.type:null}'=='monolith' || '${service.type:null}'=='tb-transport') && '${queue.type:null}'=='kafka'") @Slf4j public class KafkaTransportQueueProvider implements TransportQueueProvider { + + private final TbKafkaSettings kafkaSettings; + private final TbNodeIdProvider nodeIdProvider; + private final TbQueueTransportApiSettings transportApiSettings; + + public KafkaTransportQueueProvider(TbKafkaSettings kafkaSettings, TbNodeIdProvider nodeIdProvider, TbQueueTransportApiSettings transportApiSettings) { + this.kafkaSettings = kafkaSettings; + this.nodeIdProvider = nodeIdProvider; + this.transportApiSettings = transportApiSettings; + } + @Override public TbQueueRequestTemplate, TbProtoQueueMsg> getTransportApiRequestTemplate() { - return null; + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-transport-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(transportApiSettings.getRequestsTopic()); + + TBKafkaConsumerTemplate.TBKafkaConsumerTemplateBuilder> responseBuilder = TBKafkaConsumerTemplate.builder(); + responseBuilder.settings(kafkaSettings); + responseBuilder.topic(transportApiSettings.getResponsesTopic()); + responseBuilder.clientId("consumer-transport-" + nodeIdProvider.getNodeId()); + responseBuilder.groupId("rule-engine-node-" + nodeIdProvider.getNodeId()); + responseBuilder.autoCommit(true); + //TODO: 2.5 +// responseBuilder.autoCommitIntervalMs(autoCommitInterval); +// responseBuilder.decoder(new TransportApiResponseDecoder()); + + DefaultTbQueueRequestTemplate.DefaultTbQueueRequestTemplateBuilder + , TbProtoQueueMsg> templateBuilder = DefaultTbQueueRequestTemplate.builder(); + templateBuilder.requestTemplate(requestBuilder.build()); + templateBuilder.responseTemplate(responseBuilder.build()); + templateBuilder.maxPendingRequests(transportApiSettings.getMaxPendingRequests()); + templateBuilder.maxRequestTimeout(transportApiSettings.getMaxRequestsTimeout()); + templateBuilder.pollInterval(transportApiSettings.getResponsePollInterval()); + return templateBuilder.build(); } @Override public TbQueueProducer> getRuleEngineMsgProducer() { - return null; + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-transport-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(transportApiSettings.getRequestsTopic()); + return requestBuilder.build(); } @Override public TbQueueProducer> getTbCoreMsgProducer() { - return null; + TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder> requestBuilder = TBKafkaProducerTemplate.builder(); + requestBuilder.settings(kafkaSettings); + requestBuilder.clientId("producer-transport-" + nodeIdProvider.getNodeId()); + requestBuilder.defaultTopic(transportApiSettings.getRequestsTopic()); + return requestBuilder.build(); } @Override diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java index b7e210f0f6..6123fc0859 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/TbCoreQueueProvider.java @@ -1,9 +1,22 @@ +/** + * 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.provider; import org.thingsboard.server.TbQueueConsumer; import org.thingsboard.server.TbQueueProducer; -import org.thingsboard.server.TbQueueRequestTemplate; -import org.thingsboard.server.TbQueueResponseTemplate; import org.thingsboard.server.common.TbProtoQueueMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; diff --git a/common/queue/src/main/java/org/thingsboard/server/provider/TransportQueueProvider.java b/common/queue/src/main/java/org/thingsboard/server/provider/TransportQueueProvider.java index 14d8efd534..de69e1819a 100644 --- a/common/queue/src/main/java/org/thingsboard/server/provider/TransportQueueProvider.java +++ b/common/queue/src/main/java/org/thingsboard/server/provider/TransportQueueProvider.java @@ -1,3 +1,18 @@ +/** + * 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.provider; import org.thingsboard.server.TbQueueConsumer; diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java index e4dc2ba1f1..cdf6b3f763 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java @@ -5,7 +5,7 @@ * 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 + * 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,