Andrii Shvaika
6 years ago
93 changed files with 1720 additions and 874 deletions
@ -0,0 +1,125 @@ |
|||
/** |
|||
* 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 com.google.protobuf.ByteString; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
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.TbMsg; |
|||
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.*; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.provider.TbQueueProducerProvider; |
|||
import org.thingsboard.server.service.encoding.DataDecodingEncodingService; |
|||
import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; |
|||
|
|||
import java.util.HashSet; |
|||
import java.util.Set; |
|||
|
|||
@Service |
|||
@Slf4j |
|||
public class DefaultTbClusterService implements TbClusterService { |
|||
|
|||
protected TbQueueProducerProvider producerProvider; |
|||
private final PartitionService partitionService; |
|||
private final DataDecodingEncodingService encodingService; |
|||
|
|||
public DefaultTbClusterService(TbQueueProducerProvider producerProvider, PartitionService partitionService, DataDecodingEncodingService encodingService) { |
|||
this.producerProvider = producerProvider; |
|||
this.partitionService = partitionService; |
|||
this.encodingService = encodingService; |
|||
} |
|||
|
|||
@Override |
|||
public void onToRuleEngineMsg(TenantId tenantId, EntityId entityId, TbMsg tbMsg) { |
|||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_RULE_ENGINE, tenantId, entityId); |
|||
ToRuleEngineMsg msg = ToRuleEngineMsg.newBuilder() |
|||
.setTenantIdMSB(tenantId.getId().getMostSignificantBits()) |
|||
.setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) |
|||
.setTbMsg(TbMsg.toByteString(tbMsg)).build(); |
|||
producerProvider.getRuleEngineMsgProducer().send(tpi, new TbProtoQueueMsg<>(tbMsg.getId(), msg), null); |
|||
} |
|||
|
|||
@Override |
|||
public void onToCoreMsg(ToDeviceActorNotificationMsg msg) { |
|||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, msg.getTenantId(), msg.getDeviceId()); |
|||
byte[] msgBytes = encodingService.encode(msg); |
|||
ToCoreMsg toCoreMsg = ToCoreMsg.newBuilder().setToDeviceActorNotificationMsg(ByteString.copyFrom(msgBytes)).build(); |
|||
producerProvider.getTbCoreMsgProducer().send(tpi, new TbProtoQueueMsg<>(msg.getDeviceId().getId(), toCoreMsg), null); |
|||
} |
|||
|
|||
@Override |
|||
public void onToCoreMsg(String serviceId, FromDeviceRpcResponse response) { |
|||
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|||
FromDeviceRPCResponseProto.Builder builder = FromDeviceRPCResponseProto.newBuilder() |
|||
.setRequestIdMSB(response.getId().getMostSignificantBits()) |
|||
.setRequestIdLSB(response.getId().getLeastSignificantBits()) |
|||
.setError(response.getError().isPresent() ? response.getError().get().ordinal() : -1); |
|||
response.getResponse().ifPresent(builder::setResponse); |
|||
ToCoreNotificationMsg msg = ToCoreNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build(); |
|||
producerProvider.getTbCoreNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(response.getId(), msg), null); |
|||
} |
|||
|
|||
@Override |
|||
public void onToRuleEngineMsg(String serviceId, FromDeviceRpcResponse response) { |
|||
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|||
FromDeviceRPCResponseProto.Builder builder = FromDeviceRPCResponseProto.newBuilder() |
|||
.setRequestIdMSB(response.getId().getMostSignificantBits()) |
|||
.setRequestIdLSB(response.getId().getLeastSignificantBits()) |
|||
.setError(response.getError().isPresent() ? response.getError().get().ordinal() : -1); |
|||
response.getResponse().ifPresent(builder::setResponse); |
|||
ToRuleEngineNotificationMsg msg = ToRuleEngineNotificationMsg.newBuilder().setFromDeviceRpcResponse(builder).build(); |
|||
producerProvider.getRuleEngineNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(response.getId(), msg), null); |
|||
|
|||
} |
|||
|
|||
@Override |
|||
public void onEntityStateChange(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state) { |
|||
log.trace("[{}] Processing {} state change event: {}", tenantId, entityId.getEntityType(), state); |
|||
broadcast(new ComponentLifecycleMsg(tenantId, entityId, state)); |
|||
} |
|||
|
|||
private void broadcast(ComponentLifecycleMsg msg) { |
|||
byte[] msgBytes = encodingService.encode(msg); |
|||
TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> toRuleEngineProducer = producerProvider.getRuleEngineNotificationsMsgProducer(); |
|||
Set<String> tbRuleEngineServices = new HashSet<>(partitionService.getAllServiceIds(ServiceType.TB_RULE_ENGINE)); |
|||
if (msg.getEntityId().getEntityType().equals(EntityType.TENANT)) { |
|||
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> toCoreProducer = producerProvider.getTbCoreNotificationsMsgProducer(); |
|||
Set<String> tbCoreServices = partitionService.getAllServiceIds(ServiceType.TB_CORE); |
|||
for (String serviceId : tbCoreServices) { |
|||
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_CORE, serviceId); |
|||
ToCoreNotificationMsg toCoreMsg = ToCoreNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); |
|||
toCoreProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toCoreMsg), null); |
|||
} |
|||
// No need to push notifications twice
|
|||
tbRuleEngineServices.removeAll(tbCoreServices); |
|||
} |
|||
for (String serviceId : tbRuleEngineServices) { |
|||
TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_RULE_ENGINE, serviceId); |
|||
ToRuleEngineNotificationMsg toRuleEngineMsg = ToRuleEngineNotificationMsg.newBuilder().setComponentLifecycleMsg(ByteString.copyFrom(msgBytes)).build(); |
|||
toRuleEngineProducer.send(tpi, new TbProtoQueueMsg<>(msg.getEntityId().getId(), toRuleEngineMsg), null); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* 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.rule.engine.api.msg.ToDeviceActorNotificationMsg; |
|||
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.TbMsg; |
|||
import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; |
|||
|
|||
public interface TbClusterService { |
|||
|
|||
void onToRuleEngineMsg(TenantId tenantId, EntityId entityId, TbMsg msg); |
|||
|
|||
void onToCoreMsg(ToDeviceActorNotificationMsg msg); |
|||
|
|||
void onToCoreMsg(String targetServiceId, FromDeviceRpcResponse response); |
|||
|
|||
void onToRuleEngineMsg(String targetServiceId, FromDeviceRpcResponse response); |
|||
|
|||
void onEntityStateChange(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state); |
|||
|
|||
} |
|||
@ -1,219 +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.service.rpc; |
|||
|
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.fasterxml.jackson.core.JsonProcessingException; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import com.google.protobuf.InvalidProtocolBufferException; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.rule.engine.api.RpcError; |
|||
import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; |
|||
import org.thingsboard.server.actors.service.ActorService; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgDataType; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.common.msg.cluster.SendToClusterMsg; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.core.ToServerRpcResponseMsg; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.common.msg.queue.QueueToRuleEngineMsg; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.gen.cluster.ClusterAPIProtos; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 27.03.18. |
|||
*/ |
|||
@Service |
|||
@Slf4j |
|||
public class DefaultDeviceRpcService implements DeviceRpcService { |
|||
|
|||
private static final ObjectMapper json = new ObjectMapper(); |
|||
|
|||
@Autowired |
|||
private DeviceService deviceService; |
|||
|
|||
@Autowired |
|||
@Lazy |
|||
private ActorService actorService; |
|||
|
|||
private ScheduledExecutorService rpcCallBackExecutor; |
|||
|
|||
private final ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> localToRuleEngineRpcRequests = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> localToDeviceRpcRequests = new ConcurrentHashMap<>(); |
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
rpcCallBackExecutor = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("rpc-callback")); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (rpcCallBackExecutor != null) { |
|||
rpcCallBackExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processRestAPIRpcRequestToRuleEngine(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer) { |
|||
log.trace("[{}][{}] Processing REST API call to rule engine [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); |
|||
UUID requestId = request.getId(); |
|||
localToRuleEngineRpcRequests.put(requestId, responseConsumer); |
|||
sendRpcRequestToRuleEngine(request); |
|||
scheduleTimeout(request, requestId, localToRuleEngineRpcRequests); |
|||
} |
|||
|
|||
@Override |
|||
public void processResponseToServerSideRPCRequestFromRuleEngine(ServerAddress requestOriginAddress, FromDeviceRpcResponse response) { |
|||
log.trace("[{}] Received response to server-side RPC request from rule engine: [{}]", response.getId(), requestOriginAddress); |
|||
//TODO 2.5
|
|||
if (true) {//routingService.getCurrentServer().equals(requestOriginAddress)
|
|||
UUID requestId = response.getId(); |
|||
Consumer<FromDeviceRpcResponse> consumer = localToRuleEngineRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} else { |
|||
ClusterAPIProtos.FromDeviceRPCResponseProto.Builder builder = ClusterAPIProtos.FromDeviceRPCResponseProto.newBuilder(); |
|||
builder.setRequestIdMSB(response.getId().getMostSignificantBits()); |
|||
builder.setRequestIdLSB(response.getId().getLeastSignificantBits()); |
|||
response.getResponse().ifPresent(builder::setResponse); |
|||
if (response.getError().isPresent()) { |
|||
builder.setError(response.getError().get().ordinal()); |
|||
} else { |
|||
builder.setError(-1); |
|||
} |
|||
//TODO 2.5
|
|||
// rpcService.tell(requestOriginAddress, ClusterAPIProtos.MessageType.CLUSTER_RPC_FROM_DEVICE_RESPONSE_MESSAGE, builder.build().toByteArray());
|
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void forwardServerSideRPCRequestToDeviceActor(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer) { |
|||
log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); |
|||
UUID requestId = request.getId(); |
|||
localToDeviceRpcRequests.put(requestId, responseConsumer); |
|||
sendRpcRequestToDevice(request); |
|||
scheduleTimeout(request, requestId, localToDeviceRpcRequests); |
|||
} |
|||
|
|||
@Override |
|||
public void processResponseToServerSideRPCRequestFromDeviceActor(FromDeviceRpcResponse response) { |
|||
log.trace("[{}] Received response to server-side RPC request from device actor.", response.getId()); |
|||
UUID requestId = response.getId(); |
|||
Consumer<FromDeviceRpcResponse> consumer = localToDeviceRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processResponseToServerSideRPCRequestFromRemoteServer(ServerAddress serverAddress, byte[] data) { |
|||
ClusterAPIProtos.FromDeviceRPCResponseProto proto; |
|||
try { |
|||
proto = ClusterAPIProtos.FromDeviceRPCResponseProto.parseFrom(data); |
|||
} catch (InvalidProtocolBufferException e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
RpcError error = proto.getError() > 0 ? RpcError.values()[proto.getError()] : null; |
|||
FromDeviceRpcResponse response = new FromDeviceRpcResponse(new UUID(proto.getRequestIdMSB(), proto.getRequestIdLSB()), proto.getResponse(), error); |
|||
//TODO 2.5
|
|||
// processResponseToServerSideRPCRequestFromRuleEngine(routingService.getCurrentServer(), response);
|
|||
} |
|||
|
|||
@Override |
|||
public void sendReplyToRpcCallFromDevice(TenantId tenantId, DeviceId deviceId, int requestId, String body) { |
|||
ToServerRpcResponseActorMsg rpcMsg = new ToServerRpcResponseActorMsg(tenantId, deviceId, new ToServerRpcResponseMsg(requestId, body)); |
|||
forward(deviceId, rpcMsg); |
|||
} |
|||
|
|||
private void sendRpcRequestToRuleEngine(ToDeviceRpcRequest msg) { |
|||
ObjectNode entityNode = json.createObjectNode(); |
|||
TbMsgMetaData metaData = new TbMsgMetaData(); |
|||
metaData.putValue("requestUUID", msg.getId().toString()); |
|||
//TODO 2.5
|
|||
// metaData.putValue("originHost", routingService.getCurrentServer().getHost());
|
|||
// metaData.putValue("originPort", Integer.toString(routingService.getCurrentServer().getPort()));
|
|||
metaData.putValue("expirationTime", Long.toString(msg.getExpirationTime())); |
|||
metaData.putValue("oneway", Boolean.toString(msg.isOneway())); |
|||
|
|||
Device device = deviceService.findDeviceById(msg.getTenantId(), msg.getDeviceId()); |
|||
if (device != null) { |
|||
metaData.putValue("deviceName", device.getName()); |
|||
metaData.putValue("deviceType", device.getType()); |
|||
} |
|||
|
|||
entityNode.put("method", msg.getBody().getMethod()); |
|||
entityNode.put("params", msg.getBody().getParams()); |
|||
|
|||
try { |
|||
TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), DataConstants.RPC_CALL_FROM_SERVER_TO_DEVICE, msg.getDeviceId(), metaData, TbMsgDataType.JSON |
|||
, json.writeValueAsString(entityNode) |
|||
, null, null, null); |
|||
actorService.onMsg(new SendToClusterMsg(msg.getDeviceId(), new QueueToRuleEngineMsg(msg.getTenantId(), tbMsg))); |
|||
} catch (JsonProcessingException e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private void sendRpcRequestToDevice(ToDeviceRpcRequest msg) { |
|||
//TODO 2.5
|
|||
// ToDeviceRpcRequestActorMsg rpcMsg = new ToDeviceRpcRequestActorMsg(routingService.getCurrentServer(), msg);
|
|||
// log.trace("[{}] Forwarding msg {} to device actor!", msg.getDeviceId(), msg);
|
|||
// forward(msg.getDeviceId(), rpcMsg);
|
|||
} |
|||
|
|||
private <T extends ToDeviceActorNotificationMsg> void forward(DeviceId deviceId, T msg) { |
|||
actorService.onMsg(new SendToClusterMsg(deviceId, msg)); |
|||
} |
|||
|
|||
private void scheduleTimeout(ToDeviceRpcRequest request, UUID requestId, ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> requestsMap) { |
|||
long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis()); |
|||
log.trace("[{}] processing the request: [{}]", this.hashCode(), requestId); |
|||
rpcCallBackExecutor.schedule(() -> { |
|||
log.trace("[{}] timeout the request: [{}]", this.hashCode(), requestId); |
|||
Consumer<FromDeviceRpcResponse> consumer = requestsMap.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT)); |
|||
} |
|||
}, timeout, TimeUnit.MILLISECONDS); |
|||
} |
|||
|
|||
|
|||
} |
|||
@ -0,0 +1,203 @@ |
|||
/** |
|||
* 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.rpc; |
|||
|
|||
import akka.actor.ActorRef; |
|||
import com.datastax.driver.core.utils.UUIDs; |
|||
import com.fasterxml.jackson.core.JsonProcessingException; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.rule.engine.api.RpcError; |
|||
import org.thingsboard.server.actors.ActorSystemContext; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgDataType; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
import org.thingsboard.server.service.queue.TbClusterService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 27.03.18. |
|||
*/ |
|||
@Service |
|||
@Slf4j |
|||
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core'") |
|||
public class DefaultTbCoreDeviceRpcService implements TbCoreDeviceRpcService { |
|||
|
|||
private static final ObjectMapper json = new ObjectMapper(); |
|||
|
|||
private final DeviceService deviceService; |
|||
private final TbClusterService clusterService; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
private final ActorSystemContext actorContext; |
|||
|
|||
private final ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> localToRuleEngineRpcRequests = new ConcurrentHashMap<>(); |
|||
private final ConcurrentMap<UUID, ToDeviceRpcRequestActorMsg> localToDeviceRpcRequests = new ConcurrentHashMap<>(); |
|||
|
|||
private Optional<TbRuleEngineDeviceRpcService> tbRuleEngineRpcService; |
|||
private ScheduledExecutorService scheduler; |
|||
private String serviceId; |
|||
|
|||
public DefaultTbCoreDeviceRpcService(DeviceService deviceService, TbClusterService clusterService, TbServiceInfoProvider serviceInfoProvider, |
|||
ActorSystemContext actorContext) { |
|||
this.deviceService = deviceService; |
|||
this.clusterService = clusterService; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
this.actorContext = actorContext; |
|||
} |
|||
|
|||
@Autowired |
|||
public void setTbRuleEngineRpcService(Optional<TbRuleEngineDeviceRpcService> tbRuleEngineRpcService) { |
|||
this.tbRuleEngineRpcService = tbRuleEngineRpcService; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("tb-core-rpc-scheduler")); |
|||
serviceId = serviceInfoProvider.getServiceId(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (scheduler != null) { |
|||
scheduler.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void processRestApiRpcRequest(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer) { |
|||
log.trace("[{}][{}] Processing REST API call to rule engine [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); |
|||
UUID requestId = request.getId(); |
|||
localToRuleEngineRpcRequests.put(requestId, responseConsumer); |
|||
sendRpcRequestToRuleEngine(request); |
|||
scheduleToRuleEngineTimeout(request, requestId); |
|||
} |
|||
|
|||
@Override |
|||
public void processRpcResponseFromRuleEngine(FromDeviceRpcResponse response) { |
|||
log.trace("[{}] Received response to server-side RPC request from rule engine: [{}]", response.getId()); |
|||
UUID requestId = response.getId(); |
|||
Consumer<FromDeviceRpcResponse> consumer = localToRuleEngineRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void forwardRpcRequestToDeviceActor(ToDeviceRpcRequestActorMsg rpcMsg) { |
|||
ToDeviceRpcRequest request = rpcMsg.getMsg(); |
|||
log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); |
|||
UUID requestId = request.getId(); |
|||
localToDeviceRpcRequests.put(requestId, rpcMsg); |
|||
actorContext.getAppActor().tell(rpcMsg, ActorRef.noSender()); |
|||
scheduleToDeviceTimeout(request, requestId); |
|||
} |
|||
|
|||
@Override |
|||
public void processRpcResponseFromDeviceActor(FromDeviceRpcResponse response) { |
|||
log.trace("[{}] Received response to server-side RPC request from device actor.", response.getId()); |
|||
UUID requestId = response.getId(); |
|||
ToDeviceRpcRequestActorMsg request = localToDeviceRpcRequests.remove(requestId); |
|||
if (request != null) { |
|||
sendRpcResponseToTbRuleEngine(request.getServiceId(), response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
private void sendRpcResponseToTbRuleEngine(String originServiceId, FromDeviceRpcResponse response) { |
|||
if (serviceId.equals(originServiceId)) { |
|||
if (tbRuleEngineRpcService.isPresent()) { |
|||
tbRuleEngineRpcService.get().processRpcResponseFromDevice(response); |
|||
} else { |
|||
log.warn("Failed to find tbCoreRpcService for local service. Possible duplication of serviceIds."); |
|||
} |
|||
} else { |
|||
clusterService.onToRuleEngineMsg(originServiceId, response); |
|||
} |
|||
} |
|||
|
|||
private void sendRpcRequestToRuleEngine(ToDeviceRpcRequest msg) { |
|||
ObjectNode entityNode = json.createObjectNode(); |
|||
TbMsgMetaData metaData = new TbMsgMetaData(); |
|||
metaData.putValue("requestUUID", msg.getId().toString()); |
|||
metaData.putValue("originServiceId", serviceId); |
|||
metaData.putValue("expirationTime", Long.toString(msg.getExpirationTime())); |
|||
metaData.putValue("oneway", Boolean.toString(msg.isOneway())); |
|||
|
|||
Device device = deviceService.findDeviceById(msg.getTenantId(), msg.getDeviceId()); |
|||
if (device != null) { |
|||
metaData.putValue("deviceName", device.getName()); |
|||
metaData.putValue("deviceType", device.getType()); |
|||
} |
|||
|
|||
entityNode.put("method", msg.getBody().getMethod()); |
|||
entityNode.put("params", msg.getBody().getParams()); |
|||
|
|||
try { |
|||
TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), DataConstants.RPC_CALL_FROM_SERVER_TO_DEVICE, msg.getDeviceId(), metaData, TbMsgDataType.JSON |
|||
, json.writeValueAsString(entityNode) |
|||
, null, null, null); |
|||
clusterService.onToRuleEngineMsg(msg.getTenantId(), msg.getDeviceId(), tbMsg); |
|||
} catch (JsonProcessingException e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
private void scheduleToRuleEngineTimeout(ToDeviceRpcRequest request, UUID requestId) { |
|||
long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis()); |
|||
log.trace("[{}] processing to rule engine request: [{}]", this.hashCode(), requestId); |
|||
scheduler.schedule(() -> { |
|||
log.trace("[{}] timeout for to rule engine request: [{}]", this.hashCode(), requestId); |
|||
Consumer<FromDeviceRpcResponse> consumer = localToRuleEngineRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
consumer.accept(new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT)); |
|||
} |
|||
}, timeout, TimeUnit.MILLISECONDS); |
|||
} |
|||
|
|||
private void scheduleToDeviceTimeout(ToDeviceRpcRequest request, UUID requestId) { |
|||
long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis()); |
|||
log.trace("[{}] processing to device request: [{}]", this.hashCode(), requestId); |
|||
scheduler.schedule(() -> { |
|||
log.trace("[{}] timeout for to device request: [{}]", this.hashCode(), requestId); |
|||
localToDeviceRpcRequests.remove(requestId); |
|||
}, timeout, TimeUnit.MILLISECONDS); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,169 @@ |
|||
/** |
|||
* 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.rpc; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.context.annotation.Lazy; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.ThingsBoardThreadFactory; |
|||
import org.thingsboard.rule.engine.api.RpcError; |
|||
import org.thingsboard.rule.engine.api.RuleEngineDeviceRpcRequest; |
|||
import org.thingsboard.rule.engine.api.RuleEngineDeviceRpcResponse; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.common.msg.queue.ServiceType; |
|||
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.queue.discovery.PartitionService; |
|||
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; |
|||
import org.thingsboard.server.service.queue.TbClusterService; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.Consumer; |
|||
|
|||
@Service |
|||
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-rule-engine'") |
|||
@Slf4j |
|||
public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcService { |
|||
|
|||
private final PartitionService partitionService; |
|||
private final TbClusterService clusterService; |
|||
private final TbServiceInfoProvider serviceInfoProvider; |
|||
|
|||
private final ConcurrentMap<UUID, Consumer<FromDeviceRpcResponse>> toDeviceRpcRequests = new ConcurrentHashMap<>(); |
|||
|
|||
private Optional<TbCoreDeviceRpcService> tbCoreRpcService; |
|||
private ScheduledExecutorService scheduler; |
|||
private String serviceId; |
|||
|
|||
public DefaultTbRuleEngineRpcService(PartitionService partitionService, |
|||
TbClusterService clusterService, |
|||
TbServiceInfoProvider serviceInfoProvider) { |
|||
this.partitionService = partitionService; |
|||
this.clusterService = clusterService; |
|||
this.serviceInfoProvider = serviceInfoProvider; |
|||
} |
|||
|
|||
@Autowired |
|||
public void setTbCoreRpcService(Optional<TbCoreDeviceRpcService> tbCoreRpcService) { |
|||
this.tbCoreRpcService = tbCoreRpcService; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
scheduler = Executors.newSingleThreadScheduledExecutor(ThingsBoardThreadFactory.forName("rule-engine-rpc-scheduler")); |
|||
serviceId = serviceInfoProvider.getServiceId(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (scheduler != null) { |
|||
scheduler.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void sendRpcReplyToDevice(DeviceId deviceId, int requestId, String body) { |
|||
// TODO 2.5
|
|||
} |
|||
|
|||
@Override |
|||
public void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest src, Consumer<RuleEngineDeviceRpcResponse> consumer) { |
|||
ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(), |
|||
src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody())); |
|||
forwardRpcRequestToDeviceActor(request, response -> { |
|||
if (src.isRestApiCall()) { |
|||
sendRpcResponseToTbCore(src.getOriginServiceId(), response); |
|||
} |
|||
consumer.accept(RuleEngineDeviceRpcResponse.builder() |
|||
.deviceId(src.getDeviceId()) |
|||
.requestId(src.getRequestId()) |
|||
.error(response.getError()) |
|||
.response(response.getResponse()) |
|||
.build()); |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public void processRpcResponseFromDevice(FromDeviceRpcResponse response) { |
|||
log.trace("[{}] Received response to server-side RPC request from Core RPC Service", response.getId()); |
|||
UUID requestId = response.getId(); |
|||
Consumer<FromDeviceRpcResponse> consumer = toDeviceRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
scheduler.submit(() -> consumer.accept(response)); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
private void forwardRpcRequestToDeviceActor(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer) { |
|||
log.trace("[{}][{}] Processing local rpc call to device actor [{}]", request.getTenantId(), request.getId(), request.getDeviceId()); |
|||
UUID requestId = request.getId(); |
|||
toDeviceRpcRequests.put(requestId, responseConsumer); |
|||
sendRpcRequestToDevice(request); |
|||
scheduleTimeout(request, requestId); |
|||
} |
|||
|
|||
private void sendRpcRequestToDevice(ToDeviceRpcRequest msg) { |
|||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, msg.getTenantId(), msg.getDeviceId()); |
|||
ToDeviceRpcRequestActorMsg rpcMsg = new ToDeviceRpcRequestActorMsg(serviceId, msg); |
|||
if (tpi.isMyPartition()) { |
|||
log.trace("[{}] Forwarding msg {} to device actor!", msg.getDeviceId(), msg); |
|||
if (tbCoreRpcService.isPresent()) { |
|||
tbCoreRpcService.get().forwardRpcRequestToDeviceActor(rpcMsg); |
|||
} else { |
|||
log.warn("Failed to find tbCoreRpcService for local service. Possible duplication of serviceIds."); |
|||
} |
|||
} else { |
|||
log.trace("[{}] Forwarding msg {} to queue actor!", msg.getDeviceId(), msg); |
|||
clusterService.onToCoreMsg(rpcMsg); |
|||
} |
|||
} |
|||
|
|||
private void sendRpcResponseToTbCore(String originServiceId, FromDeviceRpcResponse response) { |
|||
if (serviceId.equals(originServiceId)) { |
|||
if (tbCoreRpcService.isPresent()) { |
|||
tbCoreRpcService.get().processRpcResponseFromRuleEngine(response); |
|||
} else { |
|||
log.warn("Failed to find tbCoreRpcService for local service. Possible duplication of serviceIds."); |
|||
} |
|||
} else { |
|||
clusterService.onToCoreMsg(originServiceId, response); |
|||
} |
|||
} |
|||
|
|||
private void scheduleTimeout(ToDeviceRpcRequest request, UUID requestId) { |
|||
long timeout = Math.max(0, request.getExpirationTime() - System.currentTimeMillis()); |
|||
log.trace("[{}] processing the request: [{}]", this.hashCode(), requestId); |
|||
scheduler.schedule(() -> { |
|||
log.trace("[{}] timeout the request: [{}]", this.hashCode(), requestId); |
|||
Consumer<FromDeviceRpcResponse> consumer = toDeviceRpcRequests.remove(requestId); |
|||
if (consumer != null) { |
|||
scheduler.submit(() -> consumer.accept(new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT))); |
|||
} |
|||
}, timeout, TimeUnit.MILLISECONDS); |
|||
} |
|||
} |
|||
@ -1,41 +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.service.rpc; |
|||
|
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
|
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 16.04.18. |
|||
*/ |
|||
public interface DeviceRpcService { |
|||
|
|||
void processRestAPIRpcRequestToRuleEngine(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer); |
|||
|
|||
void processResponseToServerSideRPCRequestFromRuleEngine(ServerAddress requestOriginAddress, FromDeviceRpcResponse response); |
|||
|
|||
void forwardServerSideRPCRequestToDeviceActor(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer); |
|||
|
|||
void processResponseToServerSideRPCRequestFromDeviceActor(FromDeviceRpcResponse response); |
|||
|
|||
void processResponseToServerSideRPCRequestFromRemoteServer(ServerAddress serverAddress, byte[] data); |
|||
|
|||
void sendReplyToRpcCallFromDevice(TenantId tenantId, DeviceId deviceId, int requestId, String body); |
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* 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.rpc; |
|||
|
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
|
|||
import java.util.function.Consumer; |
|||
|
|||
/** |
|||
* Handles REST API calls that contain RPC requests to Device. |
|||
*/ |
|||
public interface TbCoreDeviceRpcService { |
|||
|
|||
/** |
|||
* Handles REST API calls that contain RPC requests to Device and pushes them to Rule Engine. |
|||
* Schedules the timeout for the RPC call based on the {@link ToDeviceRpcRequest} |
|||
* |
|||
* @param request the RPC request |
|||
* @param responseConsumer the consumer of the RPC response |
|||
*/ |
|||
void processRestApiRpcRequest(ToDeviceRpcRequest request, Consumer<FromDeviceRpcResponse> responseConsumer); |
|||
|
|||
/** |
|||
* Handles the RPC response from the Rule Engine. |
|||
* |
|||
* @param response the RPC response |
|||
*/ |
|||
void processRpcResponseFromRuleEngine(FromDeviceRpcResponse response); |
|||
|
|||
/** |
|||
* Forwards the RPC request from Rule Engine to Device Actor |
|||
* |
|||
* @param request the RPC request message |
|||
*/ |
|||
void forwardRpcRequestToDeviceActor(ToDeviceRpcRequestActorMsg request); |
|||
|
|||
/** |
|||
* Handles the RPC response from the Device Actor (Transport). |
|||
* |
|||
* @param response the RPC response |
|||
*/ |
|||
void processRpcResponseFromDeviceActor(FromDeviceRpcResponse response); |
|||
|
|||
} |
|||
@ -1,74 +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. |
|||
*/ |
|||
syntax = "proto3"; |
|||
package cluster; |
|||
|
|||
option java_package = "org.thingsboard.server.gen.cluster"; |
|||
option java_outer_classname = "ClusterAPIProtos"; |
|||
|
|||
service ClusterRpcService { |
|||
rpc handleMsgs(stream ClusterMessage) returns (stream ClusterMessage) {} |
|||
} |
|||
|
|||
message ClusterMessage { |
|||
MessageType messageType = 1; |
|||
MessageMataInfo messageMetaInfo = 2; |
|||
ServerAddress serverAddress = 3; |
|||
bytes payload = 4; |
|||
} |
|||
|
|||
message ServerAddress { |
|||
string host = 1; |
|||
int32 port = 2; |
|||
} |
|||
|
|||
message MessageMataInfo { |
|||
string payloadMetaInfo = 1; |
|||
repeated string tags = 2; |
|||
} |
|||
|
|||
enum MessageType { |
|||
|
|||
//Cluster control messages |
|||
RPC_SESSION_CREATE_REQUEST_MSG = 0; |
|||
TO_ALL_NODES_MSG = 1; |
|||
RPC_SESSION_TELL_MSG = 2; |
|||
RPC_BROADCAST_MSG = 3; |
|||
CONNECT_RPC_MESSAGE =4; |
|||
|
|||
CLUSTER_ACTOR_MESSAGE = 5; |
|||
// Messages related to TelemetrySubscriptionService |
|||
CLUSTER_TELEMETRY_SUBSCRIPTION_CREATE_MESSAGE = 6; |
|||
CLUSTER_TELEMETRY_SUBSCRIPTION_UPDATE_MESSAGE = 7; |
|||
CLUSTER_TELEMETRY_SUBSCRIPTION_CLOSE_MESSAGE = 8; |
|||
CLUSTER_TELEMETRY_SESSION_CLOSE_MESSAGE = 9; |
|||
CLUSTER_TELEMETRY_ATTR_UPDATE_MESSAGE = 10; |
|||
CLUSTER_TELEMETRY_TS_UPDATE_MESSAGE = 11; |
|||
CLUSTER_RPC_FROM_DEVICE_RESPONSE_MESSAGE = 12; |
|||
|
|||
CLUSTER_DEVICE_STATE_SERVICE_MESSAGE = 13; |
|||
CLUSTER_TRANSACTION_SERVICE_MESSAGE = 14; |
|||
} |
|||
|
|||
// Messages related to CLUSTER_TELEMETRY_MESSAGE |
|||
|
|||
|
|||
message FromDeviceRPCResponseProto { |
|||
int64 requestIdMSB = 1; |
|||
int64 requestIdLSB = 2; |
|||
string response = 3; |
|||
int32 error = 4; |
|||
} |
|||
@ -0,0 +1,74 @@ |
|||
/** |
|||
* 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.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
@Service |
|||
@ConditionalOnExpression("'${service.type:null}'=='monolith' || '${service.type:null}'=='tb-core'") |
|||
public class TbCoreQueueProducerProvider implements TbQueueProducerProvider { |
|||
|
|||
private final TbCoreQueueProvider tbQueueProvider; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> toTransport; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> toRuleEngine; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> toTbCore; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> toRuleEngineNotifications; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> toTbCoreNotifications; |
|||
|
|||
public TbCoreQueueProducerProvider(TbCoreQueueProvider tbQueueProvider) { |
|||
this.tbQueueProvider = tbQueueProvider; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.toTbCore = tbQueueProvider.getTbCoreMsgProducer(); |
|||
this.toTransport = tbQueueProvider.getTransportNotificationsMsgProducer(); |
|||
this.toRuleEngine = tbQueueProvider.getRuleEngineMsgProducer(); |
|||
this.toRuleEngineNotifications = tbQueueProvider.getRuleEngineNotificationsMsgProducer(); |
|||
this.toTbCoreNotifications = tbQueueProvider.getTbCoreNotificationsMsgProducer(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
return toTransport; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return toRuleEngine; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
return toRuleEngineNotifications; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return toTbCore; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
return toTbCoreNotifications; |
|||
} |
|||
} |
|||
@ -0,0 +1,66 @@ |
|||
/** |
|||
* 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.queue.provider; |
|||
|
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToCoreNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineNotificationMsg; |
|||
import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
|
|||
/** |
|||
* Responsible for providing various Producers to other services. |
|||
*/ |
|||
public interface TbQueueProducerProvider { |
|||
|
|||
/** |
|||
* Used to push messages to instances of TB Transport Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToTransportMsg>> getTransportNotificationsMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push messages to instances of TB RuleEngine Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToRuleEngineMsg>> getRuleEngineMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push notifications to instances of TB RuleEngine Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push messages to other instances of TB Core Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToCoreMsg>> getTbCoreMsgProducer(); |
|||
|
|||
/** |
|||
* Used to push messages to other instances of TB Core Service |
|||
* |
|||
* @return |
|||
*/ |
|||
TbQueueProducer<TbProtoQueueMsg<ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer(); |
|||
|
|||
} |
|||
@ -0,0 +1,75 @@ |
|||
/** |
|||
* 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.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
@Service |
|||
@ConditionalOnExpression("'${service.type:null}'=='tb-rule-engine'") |
|||
public class TbRuleEngineProducerProvider implements TbQueueProducerProvider { |
|||
|
|||
private final TbRuleEngineQueueProvider tbQueueProvider; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> toTransport; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> toRuleEngine; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> toTbCore; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> toRuleEngineNotifications; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> toTbCoreNotifications; |
|||
|
|||
|
|||
public TbRuleEngineProducerProvider(TbRuleEngineQueueProvider tbQueueProvider) { |
|||
this.tbQueueProvider = tbQueueProvider; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.toTbCore = tbQueueProvider.getTbCoreMsgProducer(); |
|||
this.toTransport = tbQueueProvider.getTransportNotificationsMsgProducer(); |
|||
this.toRuleEngine = tbQueueProvider.getRuleEngineMsgProducer(); |
|||
this.toRuleEngineNotifications = tbQueueProvider.getRuleEngineNotificationsMsgProducer(); |
|||
this.toTbCoreNotifications = tbQueueProvider.getTbCoreNotificationsMsgProducer(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
return toTransport; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return toRuleEngine; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
return toRuleEngineNotifications; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return toTbCore; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
return toTbCoreNotifications; |
|||
} |
|||
} |
|||
@ -0,0 +1,70 @@ |
|||
/** |
|||
* 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.queue.provider; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.gen.transport.TransportProtos; |
|||
import org.thingsboard.server.queue.TbQueueProducer; |
|||
import org.thingsboard.server.queue.common.TbProtoQueueMsg; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
|
|||
//TODO 2.5 Maybe remove this service if it is not used.
|
|||
@Service |
|||
@ConditionalOnExpression("'${service.type:null}'=='tb-transport'") |
|||
public class TbTransportQueueProducerProvider implements TbQueueProducerProvider { |
|||
|
|||
private final TbTransportQueueProvider tbQueueProvider; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> toTransport; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> toRuleEngine; |
|||
private TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> toTbCore; |
|||
|
|||
public TbTransportQueueProducerProvider(TbTransportQueueProvider tbQueueProvider) { |
|||
this.tbQueueProvider = tbQueueProvider; |
|||
} |
|||
|
|||
@PostConstruct |
|||
public void init() { |
|||
this.toTbCore = tbQueueProvider.getTbCoreMsgProducer(); |
|||
this.toRuleEngine = tbQueueProvider.getRuleEngineMsgProducer(); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToTransportMsg>> getTransportNotificationsMsgProducer() { |
|||
throw new RuntimeException("Not Implemented! Should not be used by Transport!"); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineMsg>> getRuleEngineMsgProducer() { |
|||
return toRuleEngine; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreMsg>> getTbCoreMsgProducer() { |
|||
return toTbCore; |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToRuleEngineNotificationMsg>> getRuleEngineNotificationsMsgProducer() { |
|||
throw new RuntimeException("Not Implemented! Should not be used by Transport!"); |
|||
} |
|||
|
|||
@Override |
|||
public TbQueueProducer<TbProtoQueueMsg<TransportProtos.ToCoreNotificationMsg>> getTbCoreNotificationsMsgProducer() { |
|||
throw new RuntimeException("Not Implemented! Should not be used by Transport!"); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue