From d380ac710646a2aea8693ea14aba1eaa4b9f4994 Mon Sep 17 00:00:00 2001 From: Andrii Shvaika Date: Mon, 13 Apr 2020 18:33:11 +0300 Subject: [PATCH] Device -> Server RPC implementation and improvements --- .../server/actors/ActorSystemContext.java | 4 - .../server/actors/device/DeviceActor.java | 9 -- .../device/DeviceActorMessageProcessor.java | 108 +++++-------- .../queue/DefaultTbClusterService.java | 6 + .../service/queue/TbClusterService.java | 3 + .../rpc/DefaultTbRuleEngineRpcService.java | 14 +- .../rpc/ToServerRpcResponseActorMsg.java | 47 ------ .../DefaultTbCoreToTransportService.java | 27 ++-- .../transport/TbCoreToTransportService.java | 6 +- .../src/main/resources/thingsboard.yml | 22 +-- .../server/common/msg/MsgType.java | 2 - .../DeviceActorClientSideRpcTimeoutMsg.java | 33 ---- .../ConsistentHashPartitionService.java | 5 - .../queue/discovery/PartitionService.java | 2 - .../queue/kafka/TBKafkaProducerTemplate.java | 2 +- common/queue/src/main/proto/queue.proto | 18 +-- .../transport/coap/CoapTransportResource.java | 2 + .../transport/http/DeviceApiController.java | 2 + .../transport/mqtt/MqttTransportHandler.java | 2 + .../mqtt/session/GatewayDeviceSessionCtx.java | 2 + .../service/DefaultTransportService.java | 144 +++++++++++------- .../service/RpcRequestMetadata.java} | 16 +- .../rule/engine/api/RuleEngineRpcService.java | 4 +- .../rule/engine/rpc/TbSendRPCReplyNode.java | 10 +- .../rpc/TbSendRpcReplyNodeConfiguration.java | 24 ++- .../src/main/resources/tb-coap-transport.yml | 18 +-- .../src/main/resources/tb-http-transport.yml | 18 +-- .../src/main/resources/tb-mqtt-transport.yml | 36 +++-- 28 files changed, 266 insertions(+), 320 deletions(-) delete mode 100644 application/src/main/java/org/thingsboard/server/service/rpc/ToServerRpcResponseActorMsg.java delete mode 100644 common/message/src/main/java/org/thingsboard/server/common/msg/timeout/DeviceActorClientSideRpcTimeoutMsg.java rename common/{message/src/main/java/org/thingsboard/server/common/msg/core/ToServerRpcResponseMsg.java => transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RpcRequestMetadata.java} (76%) diff --git a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java index d31cb62b9b..7822438724 100644 --- a/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java +++ b/application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java @@ -270,10 +270,6 @@ public class ActorSystemContext { @Getter private long queuePersistenceTimeout; - @Value("${actors.client_side_rpc.timeout}") - @Getter - private long clientSideRpcTimeout; - @Value("${actors.rule.chain.error_persist_frequency}") @Getter private long ruleChainErrorPersistFrequency; diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java index 3a15240f33..256ffd0b30 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActor.java @@ -22,11 +22,8 @@ import org.thingsboard.server.actors.service.ContextAwareActor; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.msg.TbActorMsg; -import org.thingsboard.server.common.msg.queue.PartitionChangeMsg; -import org.thingsboard.server.common.msg.timeout.DeviceActorClientSideRpcTimeoutMsg; import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg; import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg; -import org.thingsboard.server.service.rpc.ToServerRpcResponseActorMsg; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; public class DeviceActor extends ContextAwareActor { @@ -72,15 +69,9 @@ public class DeviceActor extends ContextAwareActor { case DEVICE_RPC_REQUEST_TO_DEVICE_ACTOR_MSG: processor.processRpcRequest(context(), (ToDeviceRpcRequestActorMsg) msg); break; - case SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG: - processor.processToServerRPCResponse(context(), (ToServerRpcResponseActorMsg) msg); - break; case DEVICE_ACTOR_SERVER_SIDE_RPC_TIMEOUT_MSG: processor.processServerSideRpcTimeout(context(), (DeviceActorServerSideRpcTimeoutMsg) msg); break; - case DEVICE_ACTOR_CLIENT_SIDE_RPC_TIMEOUT_MSG: - processor.processClientSideRpcTimeout(context(), (DeviceActorClientSideRpcTimeoutMsg) msg); - break; case SESSION_TIMEOUT_MSG: processor.checkSessionsTimeout(); break; diff --git a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java index 83d269ef6d..437ed456e9 100644 --- a/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java @@ -36,12 +36,11 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; import org.thingsboard.server.common.msg.TbMsgMetaData; +import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; -import org.thingsboard.server.common.msg.timeout.DeviceActorClientSideRpcTimeoutMsg; import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg; -import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.AttributeUpdateNotificationMsg; -import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTransportMsg; +import org.thingsboard.server.gen.transport.TransportProtos.DeviceSessionsCacheEntry; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.GetAttributeResponseMsg; import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto; @@ -50,16 +49,19 @@ import org.thingsboard.server.gen.transport.TransportProtos.SessionCloseNotifica import org.thingsboard.server.gen.transport.TransportProtos.SessionEvent; import org.thingsboard.server.gen.transport.TransportProtos.SessionEventMsg; import org.thingsboard.server.gen.transport.TransportProtos.SessionInfoProto; +import org.thingsboard.server.gen.transport.TransportProtos.SessionSubscriptionInfoProto; +import org.thingsboard.server.gen.transport.TransportProtos.SessionType; import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToAttributeUpdatesMsg; import org.thingsboard.server.gen.transport.TransportProtos.SubscribeToRPCMsg; +import org.thingsboard.server.gen.transport.TransportProtos.SubscriptionInfoProto; import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcRequestMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToDeviceRpcResponseMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToServerRpcResponseMsg; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.gen.transport.TransportProtos.TransportToDeviceActorMsg; import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto; -import org.thingsboard.server.common.msg.queue.TbCallback; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg; -import org.thingsboard.server.service.rpc.ToServerRpcResponseActorMsg; import org.thingsboard.server.service.transport.msg.TransportToDeviceActorMsgWrapper; import javax.annotation.Nullable; @@ -91,7 +93,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { private final Map attributeSubscriptions; private final Map rpcSubscriptions; private final Map toDeviceRpcPendingMap; - private final Map toServerRpcPendingMap; private int rpcSeq = 0; private String deviceName; @@ -106,7 +107,6 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { this.attributeSubscriptions = new HashMap<>(); this.rpcSubscriptions = new HashMap<>(); this.toDeviceRpcPendingMap = new HashMap<>(); - this.toServerRpcPendingMap = new HashMap<>(); if (initAttributes()) { restoreSessions(); } @@ -142,7 +142,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { Set syncSessionSet = new HashSet<>(); rpcSubscriptions.forEach((key, value) -> { sendToTransport(rpcRequest, key, value.getNodeId()); - if (TransportProtos.SessionType.SYNC == value.getType()) { + if (SessionType.SYNC == value.getType()) { syncSessionSet.add(key); } }); @@ -177,10 +177,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void sendPendingRequests(ActorContext context, UUID sessionId, SessionInfoProto sessionInfo) { - TransportProtos.SessionType sessionType = getSessionType(sessionId); + SessionType sessionType = getSessionType(sessionId); if (!toDeviceRpcPendingMap.isEmpty()) { log.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId); - if (sessionType == TransportProtos.SessionType.SYNC) { + if (sessionType == SessionType.SYNC) { log.debug("[{}] Cleanup sync rpc session [{}]", deviceId, sessionId); rpcSubscriptions.remove(sessionId); } @@ -188,7 +188,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { log.debug("[{}] No pending RPC messages for new async session [{}]", deviceId, sessionId); } Set sentOneWayIds = new HashSet<>(); - if (sessionType == TransportProtos.SessionType.ASYNC) { + if (sessionType == SessionType.ASYNC) { toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds)); } else { toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, sessionInfo.getNodeId(), sentOneWayIds)); @@ -297,46 +297,8 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { return new HashSet<>(strings); } - private void handleClientSideRPCRequest(ActorContext context, SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg request) { -// UUID sessionId = getSessionId(sessionInfo); -// JsonObject json = new JsonObject(); -// json.addProperty("method", request.getMethodName()); -// json.add("params", JsonUtils.parse(request.getParams())); -// -// TbMsgMetaData requestMetaData = defaultMetaData.copy(); -// requestMetaData.putValue("requestId", Integer.toString(request.getRequestId())); -// TbMsg tbMsg = new TbMsg(UUIDs.timeBased(), SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, requestMetaData, TbMsgDataType.JSON, gson.toJson(json), null, null, null); -// context.parent().tell(new DeviceActorToRuleEngineMsg(context.self(), tbMsg), context.self()); -// -// scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout()); -// toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(sessionId, getSessionType(sessionId), sessionInfo.getNodeId())); - } - - private TransportProtos.SessionType getSessionType(UUID sessionId) { - return sessions.containsKey(sessionId) ? TransportProtos.SessionType.ASYNC : TransportProtos.SessionType.SYNC; - } - - void processClientSideRpcTimeout(ActorContext context, DeviceActorClientSideRpcTimeoutMsg msg) { - ToServerRpcRequestMetadata data = toServerRpcPendingMap.remove(msg.getId()); - if (data != null) { - log.debug("[{}] Client side RPC request [{}] timeout detected!", deviceId, msg.getId()); - sendToTransport(TransportProtos.ToServerRpcResponseMsg.newBuilder() - .setRequestId(msg.getId()).setError("timeout").build() - , data.getSessionId(), data.getNodeId()); - } - } - - void processToServerRPCResponse(ActorContext context, ToServerRpcResponseActorMsg msg) { - int requestId = msg.getMsg().getRequestId(); - ToServerRpcRequestMetadata data = toServerRpcPendingMap.remove(requestId); - if (data != null) { - log.debug("[{}] Pushing reply to [{}][{}]!", deviceId, data.getNodeId(), data.getSessionId()); - sendToTransport(TransportProtos.ToServerRpcResponseMsg.newBuilder() - .setRequestId(requestId).setPayload(msg.getMsg().getData()).build() - , data.getSessionId(), data.getNodeId()); - } else { - log.debug("[{}][{}] Pending RPC request to server not found!", deviceId, requestId); - } + private SessionType getSessionType(UUID sessionId) { + return sessions.containsKey(sessionId) ? SessionType.ASYNC : SessionType.SYNC; } void processAttributesUpdate(ActorContext context, DeviceAttributesEventNotificationMsg msg) { @@ -399,7 +361,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } else { SessionInfoMetaData sessionMD = sessions.get(sessionId); if (sessionMD == null) { - sessionMD = new SessionInfoMetaData(new SessionInfo(TransportProtos.SessionType.SYNC, sessionInfo.getNodeId())); + sessionMD = new SessionInfoMetaData(new SessionInfo(SessionType.SYNC, sessionInfo.getNodeId())); } sessionMD.setSubscribedToAttributes(true); log.debug("[{}] Registering attributes subscription for session [{}]", deviceId, sessionId); @@ -420,7 +382,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } else { SessionInfoMetaData sessionMD = sessions.get(sessionId); if (sessionMD == null) { - sessionMD = new SessionInfoMetaData(new SessionInfo(TransportProtos.SessionType.SYNC, sessionInfo.getNodeId())); + sessionMD = new SessionInfoMetaData(new SessionInfo(SessionType.SYNC, sessionInfo.getNodeId())); } sessionMD.setSubscribedToRPC(true); log.debug("[{}] Registering rpc subscription for session [{}]", deviceId, sessionId); @@ -444,7 +406,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { notifyTransportAboutClosedSession(sessionIdToRemove, sessions.remove(sessionIdToRemove)); } } - sessions.put(sessionId, new SessionInfoMetaData(new SessionInfo(TransportProtos.SessionType.ASYNC, sessionInfo.getNodeId()))); + sessions.put(sessionId, new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfo.getNodeId()))); if (sessions.size() == 1) { reportSessionOpen(); } @@ -462,10 +424,10 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } } - private void handleSessionActivity(ActorContext context, SessionInfoProto sessionInfoProto, TransportProtos.SubscriptionInfoProto subscriptionInfo) { + private void handleSessionActivity(ActorContext context, SessionInfoProto sessionInfoProto, SubscriptionInfoProto subscriptionInfo) { UUID sessionId = getSessionId(sessionInfoProto); SessionInfoMetaData sessionMD = sessions.computeIfAbsent(sessionId, - id -> new SessionInfoMetaData(new SessionInfo(TransportProtos.SessionType.ASYNC, sessionInfoProto.getNodeId()), 0L)); + id -> new SessionInfoMetaData(new SessionInfo(SessionType.ASYNC, sessionInfoProto.getNodeId()), 0L)); sessionMD.setLastActivityTime(subscriptionInfo.getLastActivityTime()); sessionMD.setSubscribedToAttributes(subscriptionInfo.getAttributeSubscription()); @@ -488,7 +450,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void notifyTransportAboutClosedSession(UUID sessionId, SessionInfoMetaData sessionMd) { - DeviceActorToTransportMsg msg = DeviceActorToTransportMsg.newBuilder() + ToTransportMsg msg = ToTransportMsg.newBuilder() .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) .setSessionCloseNotification(SessionCloseNotificationProto.getDefaultInstance()).build(); @@ -504,7 +466,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void sendToTransport(GetAttributeResponseMsg responseMsg, SessionInfoProto sessionInfo) { - DeviceActorToTransportMsg msg = DeviceActorToTransportMsg.newBuilder() + ToTransportMsg msg = ToTransportMsg.newBuilder() .setSessionIdMSB(sessionInfo.getSessionIdMSB()) .setSessionIdLSB(sessionInfo.getSessionIdLSB()) .setGetAttributesResponse(responseMsg).build(); @@ -512,7 +474,7 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void sendToTransport(AttributeUpdateNotificationMsg notificationMsg, UUID sessionId, String nodeId) { - DeviceActorToTransportMsg msg = DeviceActorToTransportMsg.newBuilder() + ToTransportMsg msg = ToTransportMsg.newBuilder() .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) .setAttributeUpdateNotification(notificationMsg).build(); @@ -520,15 +482,15 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { } private void sendToTransport(ToDeviceRpcRequestMsg rpcMsg, UUID sessionId, String nodeId) { - DeviceActorToTransportMsg msg = DeviceActorToTransportMsg.newBuilder() + ToTransportMsg msg = ToTransportMsg.newBuilder() .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) .setToDeviceRequest(rpcMsg).build(); systemContext.getTbCoreToTransportService().process(nodeId, msg); } - private void sendToTransport(TransportProtos.ToServerRpcResponseMsg rpcMsg, UUID sessionId, String nodeId) { - DeviceActorToTransportMsg msg = DeviceActorToTransportMsg.newBuilder() + private void sendToTransport(ToServerRpcResponseMsg rpcMsg, UUID sessionId, String nodeId) { + ToTransportMsg msg = ToTransportMsg.newBuilder() .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) .setToServerResponse(rpcMsg).build(); @@ -584,9 +546,9 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { private void restoreSessions() { log.debug("[{}] Restoring sessions from cache", deviceId); - TransportProtos.DeviceSessionsCacheEntry sessionsDump = null; + DeviceSessionsCacheEntry sessionsDump = null; try { - sessionsDump = TransportProtos.DeviceSessionsCacheEntry.parseFrom(systemContext.getDeviceSessionCacheService().get(deviceId)); + sessionsDump = DeviceSessionsCacheEntry.parseFrom(systemContext.getDeviceSessionCacheService().get(deviceId)); } catch (InvalidProtocolBufferException e) { log.warn("[{}] Failed to decode device sessions from cache", deviceId); return; @@ -595,11 +557,11 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { log.debug("[{}] No session information found", deviceId); return; } - for (TransportProtos.SessionSubscriptionInfoProto sessionSubscriptionInfoProto : sessionsDump.getSessionsList()) { + for (SessionSubscriptionInfoProto sessionSubscriptionInfoProto : sessionsDump.getSessionsList()) { SessionInfoProto sessionInfoProto = sessionSubscriptionInfoProto.getSessionInfo(); UUID sessionId = getSessionId(sessionInfoProto); - SessionInfo sessionInfo = new SessionInfo(TransportProtos.SessionType.ASYNC, sessionInfoProto.getNodeId()); - TransportProtos.SubscriptionInfoProto subInfo = sessionSubscriptionInfoProto.getSubscriptionInfo(); + SessionInfo sessionInfo = new SessionInfo(SessionType.ASYNC, sessionInfoProto.getNodeId()); + SubscriptionInfoProto subInfo = sessionSubscriptionInfoProto.getSubscriptionInfo(); SessionInfoMetaData sessionMD = new SessionInfoMetaData(sessionInfo, subInfo.getLastActivityTime()); sessions.put(sessionId, sessionMD); if (subInfo.getAttributeSubscription()) { @@ -617,27 +579,27 @@ class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcessor { private void dumpSessions() { log.debug("[{}] Dumping sessions: {}, rpc subscriptions: {}, attribute subscriptions: {} to cache", deviceId, sessions.size(), rpcSubscriptions.size(), attributeSubscriptions.size()); - List sessionsList = new ArrayList<>(sessions.size()); + List sessionsList = new ArrayList<>(sessions.size()); sessions.forEach((uuid, sessionMD) -> { - if (sessionMD.getSessionInfo().getType() == TransportProtos.SessionType.SYNC) { + if (sessionMD.getSessionInfo().getType() == SessionType.SYNC) { return; } SessionInfo sessionInfo = sessionMD.getSessionInfo(); - TransportProtos.SubscriptionInfoProto subscriptionInfoProto = TransportProtos.SubscriptionInfoProto.newBuilder() + SubscriptionInfoProto subscriptionInfoProto = SubscriptionInfoProto.newBuilder() .setLastActivityTime(sessionMD.getLastActivityTime()) .setAttributeSubscription(sessionMD.isSubscribedToAttributes()) .setRpcSubscription(sessionMD.isSubscribedToRPC()).build(); - TransportProtos.SessionInfoProto sessionInfoProto = TransportProtos.SessionInfoProto.newBuilder() + SessionInfoProto sessionInfoProto = SessionInfoProto.newBuilder() .setSessionIdMSB(uuid.getMostSignificantBits()) .setSessionIdLSB(uuid.getLeastSignificantBits()) .setNodeId(sessionInfo.getNodeId()).build(); - sessionsList.add(TransportProtos.SessionSubscriptionInfoProto.newBuilder() + sessionsList.add(SessionSubscriptionInfoProto.newBuilder() .setSessionInfo(sessionInfoProto) .setSubscriptionInfo(subscriptionInfoProto).build()); log.debug("[{}] Dumping session: {}", deviceId, sessionMD); }); systemContext.getDeviceSessionCacheService() - .put(deviceId, TransportProtos.DeviceSessionsCacheEntry.newBuilder() + .put(deviceId, DeviceSessionsCacheEntry.newBuilder() .addAllSessions(sessionsList).build().toByteArray()); } diff --git a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java index fa6aa9e886..f93657427d 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/DefaultTbClusterService.java @@ -37,6 +37,7 @@ import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; import java.util.HashSet; import java.util.Set; +import java.util.UUID; @Service @Slf4j @@ -100,7 +101,12 @@ public class DefaultTbClusterService implements TbClusterService { 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 onToTransportMsg(String serviceId, ToTransportMsg response) { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceId); + producerProvider.getTransportNotificationsMsgProducer().send(tpi, new TbProtoQueueMsg<>(UUID.randomUUID(), response), null); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java b/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java index 952b88e4d3..b70a31532e 100644 --- a/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java +++ b/application/src/main/java/org/thingsboard/server/service/queue/TbClusterService.java @@ -20,6 +20,7 @@ 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.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.service.rpc.FromDeviceRpcResponse; public interface TbClusterService { @@ -32,6 +33,8 @@ public interface TbClusterService { void onToRuleEngineMsg(String targetServiceId, FromDeviceRpcResponse response); + void onToTransportMsg(String targetServiceId, ToTransportMsg response); + void onEntityStateChange(TenantId tenantId, EntityId entityId, ComponentLifecycleEvent state); } diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java index c383da3a79..909690ff6c 100644 --- a/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java +++ b/application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java @@ -27,6 +27,7 @@ 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.gen.transport.TransportProtos; import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; import org.thingsboard.server.queue.util.TbRuleEngineComponent; @@ -52,6 +53,7 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi private final TbClusterService clusterService; private final TbServiceInfoProvider serviceInfoProvider; + private final ConcurrentMap> toDeviceRpcRequests = new ConcurrentHashMap<>(); private Optional tbCoreRpcService; @@ -85,8 +87,16 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi } @Override - public void sendRpcReplyToDevice(DeviceId deviceId, int requestId, String body) { -// TODO 2.5 + public void sendRpcReplyToDevice(String serviceId, UUID sessionId, int requestId, String body) { + TransportProtos.ToServerRpcResponseMsg responseMsg = TransportProtos.ToServerRpcResponseMsg.newBuilder() + .setRequestId(requestId) + .setPayload(body).build(); + TransportProtos.ToTransportMsg msg = TransportProtos.ToTransportMsg.newBuilder() + .setSessionIdMSB(sessionId.getMostSignificantBits()) + .setSessionIdLSB(sessionId.getLeastSignificantBits()) + .setToServerResponse(responseMsg) + .build(); + clusterService.onToTransportMsg(serviceId, msg); } @Override diff --git a/application/src/main/java/org/thingsboard/server/service/rpc/ToServerRpcResponseActorMsg.java b/application/src/main/java/org/thingsboard/server/service/rpc/ToServerRpcResponseActorMsg.java deleted file mode 100644 index 570112dab3..0000000000 --- a/application/src/main/java/org/thingsboard/server/service/rpc/ToServerRpcResponseActorMsg.java +++ /dev/null @@ -1,47 +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 lombok.Getter; -import lombok.RequiredArgsConstructor; -import lombok.ToString; -import org.thingsboard.rule.engine.api.msg.ToDeviceActorNotificationMsg; -import org.thingsboard.server.common.data.id.DeviceId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.msg.MsgType; -import org.thingsboard.server.common.msg.core.ToServerRpcResponseMsg; - -/** - * Created by ashvayka on 16.04.18. - */ -@ToString -@RequiredArgsConstructor -public class ToServerRpcResponseActorMsg implements ToDeviceActorNotificationMsg { - - @Getter - private final TenantId tenantId; - - @Getter - private final DeviceId deviceId; - - @Getter - private final ToServerRpcResponseMsg msg; - - @Override - public MsgType getMsgType() { - return MsgType.SERVER_RPC_RESPONSE_TO_DEVICE_ACTOR_MSG; - } -} 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 8f29c46466..c9e37e6791 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 @@ -18,13 +18,14 @@ package org.thingsboard.server.service.transport; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import org.thingsboard.server.common.msg.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import org.thingsboard.server.queue.TbQueueCallback; import org.thingsboard.server.queue.TbQueueMsgMetadata; import org.thingsboard.server.queue.TbQueueProducer; import org.thingsboard.server.queue.common.TbProtoQueueMsg; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.gen.transport.TransportProtos.DeviceActorToTransportMsg; -import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; +import org.thingsboard.server.queue.discovery.PartitionService; import org.thingsboard.server.queue.provider.TbQueueProducerProvider; import org.thingsboard.server.queue.util.TbCoreComponent; @@ -38,28 +39,26 @@ import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID; @TbCoreComponent public class DefaultTbCoreToTransportService implements TbCoreToTransportService { + private final PartitionService partitionService; private final TbQueueProducer> tbTransportProducer; - @Value("${queue.transport.notifications_topic}") - private String notificationsTopic; - - public DefaultTbCoreToTransportService(TbQueueProducerProvider tbQueueProducerProvider) { + public DefaultTbCoreToTransportService(PartitionService partitionService, TbQueueProducerProvider tbQueueProducerProvider) { + this.partitionService = partitionService; this.tbTransportProducer = tbQueueProducerProvider.getTransportNotificationsMsgProducer(); } @Override - public void process(String nodeId, DeviceActorToTransportMsg msg) { + public void process(String nodeId, ToTransportMsg msg) { process(nodeId, msg, null, null); } @Override - public void process(String nodeId, DeviceActorToTransportMsg msg, Runnable onSuccess, Consumer onFailure) { - String topic = notificationsTopic + "." + nodeId; + public void process(String nodeId, ToTransportMsg msg, Runnable onSuccess, Consumer onFailure) { + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, nodeId); 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<>(NULL_UUID, transportMsg); - tbTransportProducer.send(TopicPartitionInfo.builder().topic(topic).build(), queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); + log.trace("[{}][{}] Pushing session data to topic: {}", tpi.getFullTopicName(), sessionId, msg); + TbProtoQueueMsg queueMsg = new TbProtoQueueMsg<>(NULL_UUID, msg); + tbTransportProducer.send(tpi, queueMsg, new QueueCallbackAdaptor(onSuccess, onFailure)); } private static class QueueCallbackAdaptor implements TbQueueCallback { 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 5b11edb5f0..e0d5e2121a 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 @@ -15,14 +15,14 @@ */ package org.thingsboard.server.service.transport; -import org.thingsboard.server.gen.transport.TransportProtos; +import org.thingsboard.server.gen.transport.TransportProtos.ToTransportMsg; import java.util.function.Consumer; public interface TbCoreToTransportService { - void process(String nodeId, TransportProtos.DeviceActorToTransportMsg msg); + void process(String nodeId, ToTransportMsg msg); - void process(String nodeId, TransportProtos.DeviceActorToTransportMsg msg, Runnable onSuccess, Consumer onFailure); + void process(String nodeId, ToTransportMsg msg, Runnable onSuccess, Consumer onFailure); } diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index a2a06b57d0..b5e875e07b 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -251,8 +251,6 @@ actors: enabled: "${ACTORS_QUEUE_ENABLED:true}" # Maximum allowed timeout for persistence into the queue timeout: "${ACTORS_QUEUE_PERSISTENCE_TIMEOUT:30000}" - client_side_rpc: - timeout: "${CLIENT_SIDE_RPC_TIMEOUT:60000}" cache: # caffeine or redis @@ -458,6 +456,8 @@ transport: type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}" # Maximum allowed string value length when processing Telemetry/Attributes JSON (0 value disables string value length check) max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}" + client_side_rpc: + timeout: "${CLIENT_SIDE_RPC_TIMEOUT:60000}" # Local HTTP transport parameters http: enabled: "${HTTP_ENABLED:true}" @@ -552,15 +552,15 @@ queue: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" transport_api: - requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" - responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" + requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" + responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" core: - topic: "${TB_QUEUE_CORE_TOPIC:tb.core}" + topic: "${TB_QUEUE_CORE_TOPIC:tb_core}" poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -569,9 +569,9 @@ queue: print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}" js: # JS Eval request topic - request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js.eval.requests}" + request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" # JS Eval responses topic prefix that is combined with node id - response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js.eval.responses}" + response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js_eval.responses}" # JS Eval max pending requests max_pending_requests: "${REMOTE_JS_MAX_PENDING_REQUESTS:10000}" # JS Eval max request timeout @@ -581,7 +581,7 @@ queue: # JS response auto commit interval response_auto_commit_interval: "${REMOTE_JS_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}" rule-engine: - topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}" poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: @@ -589,7 +589,7 @@ queue: print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" queues: - name: "Main" - topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb.rule-engine.main}" + topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}" poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -604,7 +604,7 @@ queue: failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" - topic: "${TB_QUEUE_RE_HP_TOPIC:tb.rule-engine.hp}" + topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}" pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -620,7 +620,7 @@ queue: pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; transport: # For high priority notifications that require minimum latency and processing time - notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java index 47e09b812f..0345355aa3 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java +++ b/common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java @@ -89,8 +89,6 @@ public enum MsgType { DEVICE_ACTOR_SERVER_SIDE_RPC_TIMEOUT_MSG, - DEVICE_ACTOR_CLIENT_SIDE_RPC_TIMEOUT_MSG, - /** * Message that is sent from the Device Actor to Rule Engine. Requires acknowledgement */ diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/timeout/DeviceActorClientSideRpcTimeoutMsg.java b/common/message/src/main/java/org/thingsboard/server/common/msg/timeout/DeviceActorClientSideRpcTimeoutMsg.java deleted file mode 100644 index 5d47425e43..0000000000 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/timeout/DeviceActorClientSideRpcTimeoutMsg.java +++ /dev/null @@ -1,33 +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.common.msg.timeout; - -import org.thingsboard.server.common.msg.MsgType; - -/** - * @author Andrew Shvayka - */ -public final class DeviceActorClientSideRpcTimeoutMsg extends TimeoutMsg { - - public DeviceActorClientSideRpcTimeoutMsg(Integer id, long timeout) { - super(id, timeout); - } - - @Override - public MsgType getMsgType() { - return MsgType.DEVICE_ACTOR_CLIENT_SIDE_RPC_TIMEOUT_MSG; - } -} diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java index dfd1437b5f..df25ea3ba9 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/ConsistentHashPartitionService.java @@ -196,11 +196,6 @@ public class ConsistentHashPartitionService implements PartitionService { } } - @Override - public Set getIsolatedTenants(ServiceType serviceType) { - throw new RuntimeException("Not Implemented!"); - } - private Map> getServiceKeyListMap(List services) { final Map> currentMap = new HashMap<>(); services.forEach(serviceInfo -> { diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java index 7c6900d8a8..e3b3e76b80 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/discovery/PartitionService.java @@ -47,8 +47,6 @@ public interface PartitionService { */ Set getAllServiceIds(ServiceType serviceType); - Set getIsolatedTenants(ServiceType serviceType); - /** * Each Service should start a consumer for messages that target individual service instance based on serviceId. * This topic is likely to have single partition, and is always assigned to the service. diff --git a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java index 2fbfa7d3df..2d82ed275d 100644 --- a/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java +++ b/common/queue/src/main/java/org/thingsboard/server/queue/kafka/TBKafkaProducerTemplate.java @@ -79,7 +79,7 @@ public class TBKafkaProducerTemplate implements TbQueuePro if (callback != null) { callback.onFailure(exception); } else { - log.warn("Producer template failure", exception); + log.warn("Producer template failure: {}", exception.getMessage(), exception); } } }); diff --git a/common/queue/src/main/proto/queue.proto b/common/queue/src/main/proto/queue.proto index 3db8495d45..dac64a3478 100644 --- a/common/queue/src/main/proto/queue.proto +++ b/common/queue/src/main/proto/queue.proto @@ -245,16 +245,6 @@ message TransportToRuleEngineMsg { ToServerRpcRequestMsg toServerRPCCallRequest = 5; } -message DeviceActorToTransportMsg { - int64 sessionIdMSB = 1; - int64 sessionIdLSB = 2; - SessionCloseNotificationProto sessionCloseNotification = 3; - GetAttributeResponseMsg getAttributesResponse = 4; - AttributeUpdateNotificationMsg attributeUpdateNotification = 5; - ToDeviceRpcRequestMsg toDeviceRequest = 6; - ToServerRpcResponseMsg toServerResponse = 7; -} - /** * TB Core Data Structures */ @@ -410,5 +400,11 @@ message ToRuleEngineNotificationMsg { /* Messages that are handled by ThingsBoard Transport Service */ message ToTransportMsg { - DeviceActorToTransportMsg toDeviceSessionMsg = 1; + int64 sessionIdMSB = 1; + int64 sessionIdLSB = 2; + SessionCloseNotificationProto sessionCloseNotification = 3; + GetAttributeResponseMsg getAttributesResponse = 4; + AttributeUpdateNotificationMsg attributeUpdateNotification = 5; + ToDeviceRpcRequestMsg toDeviceRequest = 6; + ToServerRpcResponseMsg toServerResponse = 7; } diff --git a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java index 9d2cff5c09..82678fde5d 100644 --- a/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java +++ b/common/transport/coap/src/main/java/org/thingsboard/server/transport/coap/CoapTransportResource.java @@ -303,6 +303,8 @@ public class CoapTransportResource extends CoapResource { .setDeviceIdLSB(deviceInfoProto.getDeviceIdLSB()) .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) + .setDeviceName(msg.getDeviceInfo().getDeviceName()) + .setDeviceType(msg.getDeviceInfo().getDeviceType()) .build(); onSuccess.accept(sessionInfo); } else { diff --git a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java index c8e7f76450..c2f171e39b 100644 --- a/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java +++ b/common/transport/http/src/main/java/org/thingsboard/server/transport/http/DeviceApiController.java @@ -221,6 +221,8 @@ public class DeviceApiController { .setDeviceIdLSB(deviceInfoProto.getDeviceIdLSB()) .setSessionIdMSB(sessionId.getMostSignificantBits()) .setSessionIdLSB(sessionId.getLeastSignificantBits()) + .setDeviceName(msg.getDeviceInfo().getDeviceName()) + .setDeviceType(msg.getDeviceInfo().getDeviceType()) .build(); onSuccess.accept(sessionInfo); } else { diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java index fb509e91af..7295cb786b 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java @@ -507,6 +507,8 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement .setDeviceIdLSB(msg.getDeviceInfo().getDeviceIdLSB()) .setTenantIdMSB(msg.getDeviceInfo().getTenantIdMSB()) .setTenantIdLSB(msg.getDeviceInfo().getTenantIdLSB()) + .setDeviceName(msg.getDeviceInfo().getDeviceName()) + .setDeviceType(msg.getDeviceInfo().getDeviceType()) .build(); transportService.process(sessionInfo, DefaultTransportService.getSessionEventMsg(SessionEvent.OPEN), null); transportService.registerAsyncSession(sessionInfo, this); diff --git a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java index 299f5449d9..b4b5f98238 100644 --- a/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java +++ b/common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/session/GatewayDeviceSessionCtx.java @@ -44,6 +44,8 @@ public class GatewayDeviceSessionCtx extends MqttDeviceAwareSessionContext imple .setDeviceIdLSB(deviceInfo.getDeviceIdLSB()) .setTenantIdMSB(deviceInfo.getTenantIdMSB()) .setTenantIdLSB(deviceInfo.getTenantIdLSB()) + .setDeviceName(deviceInfo.getDeviceName()) + .setDeviceType(deviceInfo.getDeviceType()) .build(); setDeviceInfo(deviceInfo); } 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 7b2f157cb4..389f50009f 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 @@ -18,36 +18,25 @@ package org.thingsboard.server.common.transport.service; import com.google.gson.Gson; import com.google.gson.JsonObject; 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.ConditionalOnExpression; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardThreadFactory; -import org.thingsboard.server.common.msg.TbMsg; -import org.thingsboard.server.common.msg.TbMsgMetaData; -import org.thingsboard.server.common.msg.session.SessionMsgType; -import org.thingsboard.server.common.transport.util.JsonUtils; -import org.thingsboard.server.queue.TbQueueCallback; -import org.thingsboard.server.queue.TbQueueConsumer; -import org.thingsboard.server.queue.TbQueueMsgMetadata; -import org.thingsboard.server.queue.TbQueueProducer; -import org.thingsboard.server.queue.TbQueueRequestTemplate; -import org.thingsboard.server.queue.common.TbProtoQueueMsg; import org.thingsboard.server.common.data.EntityType; 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.queue.ServiceType; +import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; +import org.thingsboard.server.common.msg.session.SessionMsgType; import org.thingsboard.server.common.msg.tools.TbRateLimits; import org.thingsboard.server.common.msg.tools.TbRateLimitsException; import org.thingsboard.server.common.transport.SessionMsgListener; import org.thingsboard.server.common.transport.TransportService; import org.thingsboard.server.common.transport.TransportServiceCallback; -import org.thingsboard.server.queue.discovery.PartitionService; -import org.thingsboard.server.common.msg.queue.ServiceType; -import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; -import org.thingsboard.server.queue.discovery.TenantRoutingInfo; -import org.thingsboard.server.queue.discovery.TenantRoutingInfoService; -import org.thingsboard.server.queue.provider.TbQueueProducerProvider; -import org.thingsboard.server.queue.provider.TbTransportQueueFactory; +import org.thingsboard.server.common.transport.util.JsonUtils; import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.gen.transport.TransportProtos.ToCoreMsg; import org.thingsboard.server.gen.transport.TransportProtos.ToRuleEngineMsg; @@ -55,11 +44,23 @@ 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.gen.transport.TransportProtos.TransportToDeviceActorMsg; +import org.thingsboard.server.queue.TbQueueCallback; +import org.thingsboard.server.queue.TbQueueConsumer; +import org.thingsboard.server.queue.TbQueueMsgMetadata; +import org.thingsboard.server.queue.TbQueueProducer; +import org.thingsboard.server.queue.TbQueueRequestTemplate; import org.thingsboard.server.queue.common.AsyncCallbackTemplate; +import org.thingsboard.server.queue.common.TbProtoQueueMsg; +import org.thingsboard.server.queue.discovery.PartitionService; +import org.thingsboard.server.queue.discovery.TbServiceInfoProvider; +import org.thingsboard.server.queue.provider.TbQueueProducerProvider; +import org.thingsboard.server.queue.provider.TbTransportQueueFactory; import javax.annotation.PostConstruct; import javax.annotation.PreDestroy; +import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.Random; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -90,6 +91,8 @@ public class DefaultTransportService implements TransportService { private long sessionInactivityTimeout; @Value("${transport.sessions.report_timeout}") private long sessionReportTimeout; + @Value("${transport.client_side_rpc.timeout:60000}") + private long clientSideRpcTimeout; @Value("${queue.transport.poll_interval}") private int notificationsPollDuration; @@ -97,6 +100,7 @@ public class DefaultTransportService implements TransportService { private final TbTransportQueueFactory queueProvider; private final TbQueueProducerProvider producerProvider; private final PartitionService partitionService; + private final TbServiceInfoProvider serviceInfoProvider; protected TbQueueRequestTemplate, TbProtoQueueMsg> transportApiRequestTemplate; protected TbQueueProducer> ruleEngineMsgProducer; @@ -106,15 +110,17 @@ public class DefaultTransportService implements TransportService { protected ScheduledExecutorService schedulerExecutor; protected ExecutorService transportCallbackExecutor; - private ConcurrentMap sessions = new ConcurrentHashMap<>(); + private final ConcurrentMap sessions = new ConcurrentHashMap<>(); + private final Map toServerRpcPendingMap = new ConcurrentHashMap<>(); //TODO: Implement cleanup of this maps. - private ConcurrentMap perTenantLimits = new ConcurrentHashMap<>(); - private ConcurrentMap perDeviceLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap perTenantLimits = new ConcurrentHashMap<>(); + private final ConcurrentMap perDeviceLimits = new ConcurrentHashMap<>(); private ExecutorService mainConsumerExecutor = Executors.newSingleThreadExecutor(ThingsBoardThreadFactory.forName("transport-consumer")); private volatile boolean stopped = false; - public DefaultTransportService(TbTransportQueueFactory queueProvider, TbQueueProducerProvider producerProvider, PartitionService partitionService) { + public DefaultTransportService(TbServiceInfoProvider serviceInfoProvider, TbTransportQueueFactory queueProvider, TbQueueProducerProvider producerProvider, PartitionService partitionService) { + this.serviceInfoProvider = serviceInfoProvider; this.queueProvider = queueProvider; this.producerProvider = producerProvider; this.partitionService = partitionService; @@ -134,7 +140,8 @@ public class DefaultTransportService implements TransportService { ruleEngineMsgProducer = producerProvider.getRuleEngineMsgProducer(); tbCoreMsgProducer = producerProvider.getTbCoreMsgProducer(); transportNotificationsConsumer = queueProvider.createTransportNotificationsConsumer(); - transportNotificationsConsumer.subscribe(); + TopicPartitionInfo tpi = partitionService.getNotificationsTopic(ServiceType.TB_TRANSPORT, serviceInfoProvider.getServiceId()); + transportNotificationsConsumer.subscribe(Collections.singleton(tpi)); transportApiRequestTemplate.init(); mainConsumerExecutor.execute(() -> { while (!stopped) { @@ -145,10 +152,7 @@ public class DefaultTransportService implements TransportService { } records.forEach(record -> { try { - ToTransportMsg toTransportMsg = record.getValue(); - if (toTransportMsg.hasToDeviceSessionMsg()) { - processToTransportMsg(toTransportMsg.getToDeviceSessionMsg()); - } + processToTransportMsg(record.getValue()); } catch (Throwable e) { log.warn("Failed to process the notification.", e); } @@ -195,7 +199,7 @@ public class DefaultTransportService implements TransportService { @Override public void registerAsyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener) { - sessions.putIfAbsent(toId(sessionInfo), new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listener)); + sessions.putIfAbsent(toSessionId(sessionInfo), new SessionMetaData(sessionInfo, TransportProtos.SessionType.ASYNC, listener)); } @Override @@ -210,22 +214,6 @@ public class DefaultTransportService implements TransportService { } } - public TenantRoutingInfo getRoutingInfo(TenantId tenantId) { - - TransportProtos.GetTenantRoutingInfoRequestMsg msg = TransportProtos.GetTenantRoutingInfoRequestMsg.newBuilder() - .setTenantIdMSB(tenantId.getId().getMostSignificantBits()) - .setTenantIdLSB(tenantId.getId().getLeastSignificantBits()) - .build(); - TbProtoQueueMsg protoMsg = new TbProtoQueueMsg<>(UUID.randomUUID(), TransportApiRequestMsg.newBuilder().setGetTenantRoutingInfoRequestMsg(msg).build()); - try { - TbProtoQueueMsg response = transportApiRequestTemplate.send(protoMsg).get(); - TransportProtos.GetTenantRoutingInfoResponseMsg routingInfo = response.getValue().getGetTenantRoutingInfoResponseMsg(); - return new TenantRoutingInfo(tenantId, routingInfo.getIsolatedTbCore(), routingInfo.getIsolatedTbRuleEngine()); - } catch (InterruptedException | ExecutionException e) { - throw new RuntimeException(e); - } - } - @Override public void process(TransportProtos.ValidateDeviceTokenRequestMsg msg, TransportServiceCallback callback) { log.trace("Processing msg: {}", msg); @@ -253,7 +241,7 @@ public class DefaultTransportService implements TransportService { @Override public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.SubscriptionInfoProto msg, TransportServiceCallback callback) { if (log.isTraceEnabled()) { - log.trace("[{}] Processing msg: {}", toId(sessionInfo), msg); + log.trace("[{}] Processing msg: {}", toSessionId(sessionInfo), msg); } sendToDeviceActor(sessionInfo, TransportToDeviceActorMsg.newBuilder().setSessionInfo(sessionInfo) .setSubscriptionInfo(msg).build(), callback); @@ -340,13 +328,51 @@ public class DefaultTransportService implements TransportService { } } - //TODO 2.5: Need to handle timeouts on the transport level and not on the Device Actor Level. + private void processTimeout(String requestId) { + RpcRequestMetadata data = toServerRpcPendingMap.remove(requestId); + if (data != null) { + SessionMetaData md = sessions.get(data.getSessionId()); + if (md != null) { + SessionMsgListener listener = md.getListener(); + transportCallbackExecutor.submit(() -> { + TransportProtos.ToServerRpcResponseMsg responseMsg = + TransportProtos.ToServerRpcResponseMsg.newBuilder() + .setRequestId(data.getRequestId()) + .setError("timeout").build(); + listener.onToServerRpcResponse(responseMsg); + }); + if (md.getSessionType() == TransportProtos.SessionType.SYNC) { + deregisterSession(md.getSessionInfo()); + } + } else { + log.debug("[{}] Missing session.", data.getSessionId()); + } + } + } + @Override public void process(TransportProtos.SessionInfoProto sessionInfo, TransportProtos.ToServerRpcRequestMsg msg, TransportServiceCallback callback) { if (checkLimits(sessionInfo, msg, callback)) { reportActivityInternal(sessionInfo); -// sendToRuleEngine(sessionInfo, TransportToRuleEngineMsg.newBuilder().setSessionInfo(sessionInfo). -// setToServerRPCCallRequest(msg).build(), new TransportTbQueueCallback(callback)); + UUID sessionId = toSessionId(sessionInfo); + TenantId tenantId = getTenantId(sessionInfo); + DeviceId deviceId = getDeviceId(sessionInfo); + JsonObject json = new JsonObject(); + json.addProperty("method", msg.getMethodName()); + json.add("params", JsonUtils.parse(msg.getParams())); + + TbMsgMetaData metaData = new TbMsgMetaData(); + metaData.putValue("deviceName", sessionInfo.getDeviceName()); + metaData.putValue("deviceType", sessionInfo.getDeviceType()); + metaData.putValue("requestId", Integer.toString(msg.getRequestId())); + metaData.putValue("serviceId", serviceInfoProvider.getServiceId()); + metaData.putValue("sessionId", sessionId.toString()); + TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.TO_SERVER_RPC_REQUEST.name(), deviceId, metaData, TbMsgDataType.JSON, gson.toJson(json)); + sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback)); + + String requestId = sessionId + "-" + msg.getRequestId(); + toServerRpcPendingMap.put(requestId, new RpcRequestMetadata(sessionId, msg.getRequestId())); + schedulerExecutor.schedule(() -> processTimeout(requestId), clientSideRpcTimeout, TimeUnit.MILLISECONDS); } } @@ -364,7 +390,7 @@ public class DefaultTransportService implements TransportService { } private SessionMetaData reportActivityInternal(TransportProtos.SessionInfoProto sessionInfo) { - UUID sessionId = toId(sessionInfo); + UUID sessionId = toSessionId(sessionInfo); SessionMetaData sessionMetaData = sessions.get(sessionId); if (sessionMetaData != null) { sessionMetaData.updateLastActivityTime(); @@ -377,7 +403,7 @@ public class DefaultTransportService implements TransportService { sessions.forEach((uuid, sessionMD) -> { if (sessionMD.getLastActivityTime() < expTime) { if (log.isDebugEnabled()) { - log.debug("[{}] Session has expired due to last activity time: {}", toId(sessionMD.getSessionInfo()), sessionMD.getLastActivityTime()); + log.debug("[{}] Session has expired due to last activity time: {}", toSessionId(sessionMD.getSessionInfo()), sessionMD.getLastActivityTime()); } process(sessionMD.getSessionInfo(), getSessionEventMsg(TransportProtos.SessionEvent.CLOSED), null); sessions.remove(uuid); @@ -407,7 +433,7 @@ public class DefaultTransportService implements TransportService { @Override public void registerSyncSession(TransportProtos.SessionInfoProto sessionInfo, SessionMsgListener listener, long timeout) { SessionMetaData currentSession = new SessionMetaData(sessionInfo, TransportProtos.SessionType.SYNC, listener); - sessions.putIfAbsent(toId(sessionInfo), currentSession); + sessions.putIfAbsent(toSessionId(sessionInfo), currentSession); ScheduledFuture executorFuture = schedulerExecutor.schedule(() -> { listener.onRemoteSessionCloseCommand(TransportProtos.SessionCloseNotificationProto.getDefaultInstance()); @@ -419,18 +445,18 @@ public class DefaultTransportService implements TransportService { @Override public void deregisterSession(TransportProtos.SessionInfoProto sessionInfo) { - SessionMetaData currentSession = sessions.get(toId(sessionInfo)); + SessionMetaData currentSession = sessions.get(toSessionId(sessionInfo)); if (currentSession != null && currentSession.hasScheduledFuture()) { log.debug("Stopping scheduler to avoid resending response if request has been ack."); currentSession.getScheduledFuture().cancel(false); } - sessions.remove(toId(sessionInfo)); + sessions.remove(toSessionId(sessionInfo)); } @Override public boolean checkLimits(TransportProtos.SessionInfoProto sessionInfo, Object msg, TransportServiceCallback callback) { if (log.isTraceEnabled()) { - log.trace("[{}] Processing msg: {}", toId(sessionInfo), msg); + log.trace("[{}] Processing msg: {}", toSessionId(sessionInfo), msg); } if (!rateLimitEnabled) { return true; @@ -442,7 +468,7 @@ public class DefaultTransportService implements TransportService { callback.onError(new TbRateLimitsException(EntityType.TENANT)); } if (log.isTraceEnabled()) { - log.trace("[{}][{}] Tenant level rate limit detected: {}", toId(sessionInfo), tenantId, msg); + log.trace("[{}][{}] Tenant level rate limit detected: {}", toSessionId(sessionInfo), tenantId, msg); } return false; } @@ -453,7 +479,7 @@ public class DefaultTransportService implements TransportService { callback.onError(new TbRateLimitsException(EntityType.DEVICE)); } if (log.isTraceEnabled()) { - log.trace("[{}][{}] Device level rate limit detected: {}", toId(sessionInfo), deviceId, msg); + log.trace("[{}][{}] Device level rate limit detected: {}", toSessionId(sessionInfo), deviceId, msg); } return false; } @@ -461,7 +487,7 @@ public class DefaultTransportService implements TransportService { return true; } - protected void processToTransportMsg(TransportProtos.DeviceActorToTransportMsg toSessionMsg) { + protected void processToTransportMsg(TransportProtos.ToTransportMsg toSessionMsg) { UUID sessionId = new UUID(toSessionMsg.getSessionIdMSB(), toSessionMsg.getSessionIdLSB()); SessionMetaData md = sessions.get(sessionId); if (md != null) { @@ -480,6 +506,8 @@ public class DefaultTransportService implements TransportService { listener.onToDeviceRpcRequest(toSessionMsg.getToDeviceRequest()); } if (toSessionMsg.hasToServerResponse()) { + String requestId = sessionId + "-" + toSessionMsg.getToServerResponse().getRequestId(); + toServerRpcPendingMap.remove(requestId); listener.onToServerRpcResponse(toSessionMsg.getToServerResponse()); } }); @@ -492,7 +520,7 @@ public class DefaultTransportService implements TransportService { } } - protected UUID toId(TransportProtos.SessionInfoProto sessionInfo) { + protected UUID toSessionId(TransportProtos.SessionInfoProto sessionInfo) { return new UUID(sessionInfo.getSessionIdMSB(), sessionInfo.getSessionIdLSB()); } diff --git a/common/message/src/main/java/org/thingsboard/server/common/msg/core/ToServerRpcResponseMsg.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RpcRequestMetadata.java similarity index 76% rename from common/message/src/main/java/org/thingsboard/server/common/msg/core/ToServerRpcResponseMsg.java rename to common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RpcRequestMetadata.java index d06046fd71..2c5da0d3ea 100644 --- a/common/message/src/main/java/org/thingsboard/server/common/msg/core/ToServerRpcResponseMsg.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/RpcRequestMetadata.java @@ -13,20 +13,14 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.thingsboard.server.common.msg.core; +package org.thingsboard.server.common.transport.service; import lombok.Data; -/** - * @author Andrew Shvayka - */ -@Data -public class ToServerRpcResponseMsg { +import java.util.UUID; +@Data +public class RpcRequestMetadata { + private final UUID sessionId; private final int requestId; - private final String data; - - public boolean isSuccess() { - return true; - } } diff --git a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java index 5b76ecf624..baac594a17 100644 --- a/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java +++ b/rule-engine/rule-engine-api/src/main/java/org/thingsboard/rule/engine/api/RuleEngineRpcService.java @@ -16,6 +16,8 @@ package org.thingsboard.rule.engine.api; import org.thingsboard.server.common.data.id.DeviceId; + +import java.util.UUID; import java.util.function.Consumer; /** @@ -23,7 +25,7 @@ import java.util.function.Consumer; */ public interface RuleEngineRpcService { - void sendRpcReplyToDevice(DeviceId deviceId, int requestId, String body); + void sendRpcReplyToDevice(String serviceId, UUID sessionId, int requestId, String body); void sendRpcRequestToDevice(RuleEngineDeviceRpcRequest request, Consumer consumer); diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java index 393ef0bd7c..145e73d450 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRPCReplyNode.java @@ -28,6 +28,8 @@ import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.plugin.ComponentType; import org.thingsboard.server.common.msg.TbMsg; +import java.util.UUID; + @Slf4j @RuleNode( type = ComponentType.ACTION, @@ -50,15 +52,21 @@ public class TbSendRPCReplyNode implements TbNode { @Override public void onMsg(TbContext ctx, TbMsg msg) { + String serviceIdStr = msg.getMetaData().getValue(config.getServiceIdMetaDataAttribute()); + String sessionIdStr = msg.getMetaData().getValue(config.getSessionIdMetaDataAttribute()); String requestIdStr = msg.getMetaData().getValue(config.getRequestIdMetaDataAttribute()); if (msg.getOriginator().getEntityType() != EntityType.DEVICE) { ctx.tellFailure(msg, new RuntimeException("Message originator is not a device entity!")); } else if (StringUtils.isEmpty(requestIdStr)) { ctx.tellFailure(msg, new RuntimeException("Request id is not present in the metadata!")); + } else if (StringUtils.isEmpty(serviceIdStr)) { + ctx.tellFailure(msg, new RuntimeException("Service id is not present in the metadata!")); + } else if (StringUtils.isEmpty(sessionIdStr)) { + ctx.tellFailure(msg, new RuntimeException("Session id is not present in the metadata!")); } else if (StringUtils.isEmpty(msg.getData())) { ctx.tellFailure(msg, new RuntimeException("Request body is empty!")); } else { - ctx.getRpcService().sendRpcReplyToDevice(new DeviceId(msg.getOriginator().getId()), Integer.parseInt(requestIdStr), msg.getData()); + ctx.getRpcService().sendRpcReplyToDevice(serviceIdStr, UUID.fromString(sessionIdStr), Integer.parseInt(requestIdStr), msg.getData()); ctx.tellSuccess(msg); } } diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRpcReplyNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRpcReplyNodeConfiguration.java index d01f28f136..bd3a3cdfb2 100644 --- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRpcReplyNodeConfiguration.java +++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rpc/TbSendRpcReplyNodeConfiguration.java @@ -16,18 +16,40 @@ package org.thingsboard.rule.engine.rpc; import lombok.Data; +import org.springframework.util.StringUtils; import org.thingsboard.rule.engine.api.NodeConfiguration; import org.thingsboard.server.common.data.DataConstants; @Data public class TbSendRpcReplyNodeConfiguration implements NodeConfiguration { + public static final String SERVICE_ID = "serviceId"; + public static final String SESSION_ID = "sessionId"; + public static final String REQUEST_ID = "requestId"; + + private String serviceIdMetaDataAttribute; + private String sessionIdMetaDataAttribute; private String requestIdMetaDataAttribute; @Override public TbSendRpcReplyNodeConfiguration defaultConfiguration() { TbSendRpcReplyNodeConfiguration configuration = new TbSendRpcReplyNodeConfiguration(); - configuration.setRequestIdMetaDataAttribute("requestId"); + configuration.setServiceIdMetaDataAttribute(SERVICE_ID); + configuration.setSessionIdMetaDataAttribute(SESSION_ID); + configuration.setRequestIdMetaDataAttribute(REQUEST_ID); return configuration; } + + public String getServiceIdMetaDataAttribute() { + return !StringUtils.isEmpty(serviceIdMetaDataAttribute) ? serviceIdMetaDataAttribute : SERVICE_ID; + } + + public String getSessionIdMetaDataAttribute() { + return !StringUtils.isEmpty(sessionIdMetaDataAttribute) ? sessionIdMetaDataAttribute : SESSION_ID; + } + + public String getRequestIdMetaDataAttribute() { + return !StringUtils.isEmpty(requestIdMetaDataAttribute) ? requestIdMetaDataAttribute : REQUEST_ID; + } } + diff --git a/transport/coap/src/main/resources/tb-coap-transport.yml b/transport/coap/src/main/resources/tb-coap-transport.yml index 70d1ff8fbe..b94148f608 100644 --- a/transport/coap/src/main/resources/tb-coap-transport.yml +++ b/transport/coap/src/main/resources/tb-coap-transport.yml @@ -81,15 +81,15 @@ queue: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" transport_api: - requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" - responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" + requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" + responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" core: - topic: "${TB_QUEUE_CORE_TOPIC:tb.core}" + topic: "${TB_QUEUE_CORE_TOPIC:tb_core}" poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -98,9 +98,9 @@ queue: print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}" js: # JS Eval request topic - request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js.eval.requests}" + request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" # JS Eval responses topic prefix that is combined with node id - response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js.eval.responses}" + response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js_eval.responses}" # JS Eval max pending requests max_pending_requests: "${REMOTE_JS_MAX_PENDING_REQUESTS:10000}" # JS Eval max request timeout @@ -110,7 +110,7 @@ queue: # JS response auto commit interval response_auto_commit_interval: "${REMOTE_JS_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}" rule-engine: - topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}" poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: @@ -118,7 +118,7 @@ queue: print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" queues: - name: "Main" - topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb.rule-engine.main}" + topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}" poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -133,7 +133,7 @@ queue: failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" - topic: "${TB_QUEUE_RE_HP_TOPIC:tb.rule-engine.hp}" + topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}" pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -149,7 +149,7 @@ queue: pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; transport: # For high priority notifications that require minimum latency and processing time - notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: diff --git a/transport/http/src/main/resources/tb-http-transport.yml b/transport/http/src/main/resources/tb-http-transport.yml index 5ddb7c0553..0fa2788cac 100644 --- a/transport/http/src/main/resources/tb-http-transport.yml +++ b/transport/http/src/main/resources/tb-http-transport.yml @@ -82,15 +82,15 @@ queue: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" transport_api: - requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" - responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" + requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" + responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" core: - topic: "${TB_QUEUE_CORE_TOPIC:tb.core}" + topic: "${TB_QUEUE_CORE_TOPIC:tb_core}" poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -99,9 +99,9 @@ queue: print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}" js: # JS Eval request topic - request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js.eval.requests}" + request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" # JS Eval responses topic prefix that is combined with node id - response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js.eval.responses}" + response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js_eval.responses}" # JS Eval max pending requests max_pending_requests: "${REMOTE_JS_MAX_PENDING_REQUESTS:10000}" # JS Eval max request timeout @@ -111,7 +111,7 @@ queue: # JS response auto commit interval response_auto_commit_interval: "${REMOTE_JS_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}" rule-engine: - topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}" poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: @@ -119,7 +119,7 @@ queue: print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" queues: - name: "Main" - topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb.rule-engine.main}" + topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}" poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -134,7 +134,7 @@ queue: failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" - topic: "${TB_QUEUE_RE_HP_TOPIC:tb.rule-engine.hp}" + topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}" pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -150,7 +150,7 @@ queue: pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; transport: # For high priority notifications that require minimum latency and processing time - notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: diff --git a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml index c421c43e33..747427caf8 100644 --- a/transport/mqtt/src/main/resources/tb-mqtt-transport.yml +++ b/transport/mqtt/src/main/resources/tb-mqtt-transport.yml @@ -17,10 +17,20 @@ spring.main.web-environment: false spring.main.web-application-type: none -# Clustering properties -cluster: - # Unique id for this node (autogenerated if empty) - node_id: "${CLUSTER_NODE_ID:}" +# Zookeeper connection parameters. Used for service discovery. +zk: + # Enable/disable zookeeper discovery service. + enabled: "${ZOOKEEPER_ENABLED:false}" + # Zookeeper connect string + url: "${ZOOKEEPER_URL:localhost:2181}" + # Zookeeper retry interval in milliseconds + retry_interval_ms: "${ZOOKEEPER_RETRY_INTERVAL_MS:3000}" + # Zookeeper connection timeout in milliseconds + connection_timeout_ms: "${ZOOKEEPER_CONNECTION_TIMEOUT_MS:3000}" + # Zookeeper session timeout in milliseconds + session_timeout_ms: "${ZOOKEEPER_SESSION_TIMEOUT_MS:3000}" + # Name of the directory in zookeeper 'filesystem' + zk_dir: "${ZOOKEEPER_NODES_DIR:/thingsboard}" # MQTT server parameters transport: @@ -102,15 +112,15 @@ queue: hash_function_name: "${TB_QUEUE_PARTITIONS_HASH_FUNCTION_NAME:murmur3_128}" virtual_nodes_size: "${TB_QUEUE_PARTITIONS_VIRTUAL_NODES_SIZE:16}" transport_api: - requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb.transport.api.requests}" - responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb.transport.api.responses}" + requests_topic: "${TB_QUEUE_TRANSPORT_API_REQUEST_TOPIC:tb_transport.api.requests}" + responses_topic: "${TB_QUEUE_TRANSPORT_API_RESPONSE_TOPIC:tb_transport.api.responses}" max_pending_requests: "${TB_QUEUE_TRANSPORT_MAX_PENDING_REQUESTS:10000}" max_requests_timeout: "${TB_QUEUE_TRANSPORT_MAX_REQUEST_TIMEOUT:10000}" max_callback_threads: "${TB_QUEUE_TRANSPORT_MAX_CALLBACK_THREADS:100}" request_poll_interval: "${TB_QUEUE_TRANSPORT_REQUEST_POLL_INTERVAL_MS:25}" response_poll_interval: "${TB_QUEUE_TRANSPORT_RESPONSE_POLL_INTERVAL_MS:25}" core: - topic: "${TB_QUEUE_CORE_TOPIC:tb.core}" + topic: "${TB_QUEUE_CORE_TOPIC:tb_core}" poll-interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_CORE_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_CORE_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -119,9 +129,9 @@ queue: print-interval-ms: "${TB_QUEUE_CORE_STATS_PRINT_INTERVAL_MS:10000}" js: # JS Eval request topic - request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js.eval.requests}" + request_topic: "${REMOTE_JS_EVAL_REQUEST_TOPIC:js_eval.requests}" # JS Eval responses topic prefix that is combined with node id - response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js.eval.responses}" + response_topic_prefix: "${REMOTE_JS_EVAL_RESPONSE_TOPIC:js_eval.responses}" # JS Eval max pending requests max_pending_requests: "${REMOTE_JS_MAX_PENDING_REQUESTS:10000}" # JS Eval max request timeout @@ -131,7 +141,7 @@ queue: # JS response auto commit interval response_auto_commit_interval: "${REMOTE_JS_RESPONSE_AUTO_COMMIT_INTERVAL_MS:100}" rule-engine: - topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb.rule-engine}" + topic: "${TB_QUEUE_RULE_ENGINE_TOPIC:tb_rule_engine}" poll-interval: "${TB_QUEUE_RULE_ENGINE_POLL_INTERVAL_MS:25}" pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}" stats: @@ -139,7 +149,7 @@ queue: print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}" queues: - name: "Main" - topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb.rule-engine.main}" + topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}" poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}" pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -154,7 +164,7 @@ queue: failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages; pause-between-retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRY_PAUSE:3}"# Time in seconds to wait in consumer thread before retries; - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}" - topic: "${TB_QUEUE_RE_HP_TOPIC:tb.rule-engine.hp}" + topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}" poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}" partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}" pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}" @@ -170,7 +180,7 @@ queue: pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries; transport: # For high priority notifications that require minimum latency and processing time - notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb.transport.notifications}" + notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}" poll_interval: "${TB_QUEUE_CORE_POLL_INTERVAL_MS:25}" service: