|
|
|
@ -1,12 +1,12 @@ |
|
|
|
/** |
|
|
|
* Copyright © 2016-2018 The Thingsboard Authors |
|
|
|
* |
|
|
|
* <p> |
|
|
|
* 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
|
|
|
|
* |
|
|
|
* <p> |
|
|
|
* http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
* <p> |
|
|
|
* 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. |
|
|
|
@ -25,6 +25,7 @@ import com.google.common.util.concurrent.ListenableFuture; |
|
|
|
import com.google.gson.Gson; |
|
|
|
import com.google.gson.JsonArray; |
|
|
|
import com.google.gson.JsonObject; |
|
|
|
import com.google.gson.JsonParser; |
|
|
|
import org.thingsboard.server.actors.ActorSystemContext; |
|
|
|
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; |
|
|
|
import org.thingsboard.server.common.data.DataConstants; |
|
|
|
@ -53,28 +54,28 @@ import org.thingsboard.server.common.msg.core.RuleEngineErrorMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.SessionCloseMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.SessionCloseNotification; |
|
|
|
import org.thingsboard.server.common.msg.core.SessionOpenMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.StatusCodeResponse; |
|
|
|
import org.thingsboard.server.common.msg.core.TelemetryUploadRequest; |
|
|
|
import org.thingsboard.server.common.msg.core.ToDeviceRpcRequestMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.ToDeviceRpcResponseMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.ToDeviceSessionActorMsg; |
|
|
|
import org.thingsboard.server.common.msg.core.ToServerRpcRequestMsg; |
|
|
|
import org.thingsboard.server.common.msg.device.DeviceToDeviceActorMsg; |
|
|
|
import org.thingsboard.server.common.msg.kv.BasicAttributeKVMsg; |
|
|
|
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|
|
|
import org.thingsboard.server.common.msg.session.FromDeviceMsg; |
|
|
|
import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; |
|
|
|
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|
|
|
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|
|
|
import org.thingsboard.server.common.msg.session.SessionType; |
|
|
|
import org.thingsboard.server.common.msg.session.ToDeviceMsg; |
|
|
|
import org.thingsboard.server.common.msg.timeout.DeviceActorClientSideRpcTimeoutMsg; |
|
|
|
import org.thingsboard.server.common.msg.timeout.DeviceActorQueueTimeoutMsg; |
|
|
|
import org.thingsboard.server.common.msg.timeout.DeviceActorRpcTimeoutMsg; |
|
|
|
import org.thingsboard.server.common.msg.timeout.DeviceActorServerSideRpcTimeoutMsg; |
|
|
|
import org.thingsboard.server.extensions.api.device.DeviceAttributesEventNotificationMsg; |
|
|
|
import org.thingsboard.server.extensions.api.device.DeviceNameOrTypeUpdateMsg; |
|
|
|
import org.thingsboard.server.extensions.api.plugins.PluginCallback; |
|
|
|
import org.thingsboard.server.extensions.api.plugins.PluginContext; |
|
|
|
import org.thingsboard.server.extensions.api.plugins.msg.FromDeviceRpcResponse; |
|
|
|
import org.thingsboard.server.extensions.api.plugins.msg.RpcError; |
|
|
|
import org.thingsboard.server.service.rpc.ToDeviceRpcRequestActorMsg; |
|
|
|
import org.thingsboard.server.service.rpc.ToServerRpcResponseActorMsg; |
|
|
|
|
|
|
|
import javax.annotation.Nullable; |
|
|
|
import java.util.ArrayList; |
|
|
|
@ -87,7 +88,6 @@ import java.util.Map; |
|
|
|
import java.util.Optional; |
|
|
|
import java.util.Set; |
|
|
|
import java.util.UUID; |
|
|
|
import java.util.concurrent.ExecutionException; |
|
|
|
import java.util.concurrent.TimeoutException; |
|
|
|
import java.util.function.Consumer; |
|
|
|
import java.util.function.Predicate; |
|
|
|
@ -103,10 +103,12 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
private final Map<SessionId, SessionInfo> sessions; |
|
|
|
private final Map<SessionId, SessionInfo> attributeSubscriptions; |
|
|
|
private final Map<SessionId, SessionInfo> rpcSubscriptions; |
|
|
|
private final Map<Integer, ToDeviceRpcRequestMetadata> rpcPendingMap; |
|
|
|
private final Map<Integer, ToDeviceRpcRequestMetadata> toDeviceRpcPendingMap; |
|
|
|
private final Map<Integer, ToServerRpcRequestMetadata> toServerRpcPendingMap; |
|
|
|
private final Map<UUID, PendingSessionMsgData> pendingMsgs; |
|
|
|
|
|
|
|
private final Gson gson = new Gson(); |
|
|
|
private final JsonParser jsonParser = new JsonParser(); |
|
|
|
|
|
|
|
private int rpcSeq = 0; |
|
|
|
private String deviceName; |
|
|
|
@ -120,7 +122,8 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
this.sessions = new HashMap<>(); |
|
|
|
this.attributeSubscriptions = new HashMap<>(); |
|
|
|
this.rpcSubscriptions = new HashMap<>(); |
|
|
|
this.rpcPendingMap = new HashMap<>(); |
|
|
|
this.toDeviceRpcPendingMap = new HashMap<>(); |
|
|
|
this.toServerRpcPendingMap = new HashMap<>(); |
|
|
|
this.pendingMsgs = new HashMap<>(); |
|
|
|
initAttributes(); |
|
|
|
} |
|
|
|
@ -134,7 +137,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
this.defaultMetaData.putValue("deviceType", deviceType); |
|
|
|
} |
|
|
|
|
|
|
|
void processRpcRequest(ActorContext context, org.thingsboard.server.service.rpc.ToDeviceRpcRequestMsg msg) { |
|
|
|
void processRpcRequest(ActorContext context, ToDeviceRpcRequestActorMsg msg) { |
|
|
|
ToDeviceRpcRequest request = msg.getMsg(); |
|
|
|
ToDeviceRpcRequestBody body = request.getBody(); |
|
|
|
ToDeviceRpcRequestMsg rpcRequest = new ToDeviceRpcRequestMsg( |
|
|
|
@ -174,14 +177,14 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
|
|
|
|
} |
|
|
|
|
|
|
|
private void registerPendingRpcRequest(ActorContext context, org.thingsboard.server.service.rpc.ToDeviceRpcRequestMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) { |
|
|
|
rpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent)); |
|
|
|
DeviceActorRpcTimeoutMsg timeoutMsg = new DeviceActorRpcTimeoutMsg(rpcRequest.getRequestId(), timeout); |
|
|
|
private void registerPendingRpcRequest(ActorContext context, ToDeviceRpcRequestActorMsg msg, boolean sent, ToDeviceRpcRequestMsg rpcRequest, long timeout) { |
|
|
|
toDeviceRpcPendingMap.put(rpcRequest.getRequestId(), new ToDeviceRpcRequestMetadata(msg, sent)); |
|
|
|
DeviceActorServerSideRpcTimeoutMsg timeoutMsg = new DeviceActorServerSideRpcTimeoutMsg(rpcRequest.getRequestId(), timeout); |
|
|
|
scheduleMsgWithDelay(context, timeoutMsg, timeoutMsg.getTimeout()); |
|
|
|
} |
|
|
|
|
|
|
|
void processRpcTimeout(ActorContext context, DeviceActorRpcTimeoutMsg msg) { |
|
|
|
ToDeviceRpcRequestMetadata requestMd = rpcPendingMap.remove(msg.getId()); |
|
|
|
void processServerSideRpcTimeout(ActorContext context, DeviceActorServerSideRpcTimeoutMsg msg) { |
|
|
|
ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(msg.getId()); |
|
|
|
if (requestMd != null) { |
|
|
|
logger.debug("[{}] RPC request [{}] timeout detected!", deviceId, msg.getId()); |
|
|
|
systemContext.getDeviceRpcService().process(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(), |
|
|
|
@ -200,7 +203,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
|
|
|
|
void processQueueAck(ActorContext context, RuleEngineQueuePutAckMsg msg) { |
|
|
|
PendingSessionMsgData data = pendingMsgs.remove(msg.getId()); |
|
|
|
if (data != null) { |
|
|
|
if (data != null && data.isReplyOnQueueAck()) { |
|
|
|
logger.debug("[{}] Queue put [{}] ack detected!", deviceId, msg.getId()); |
|
|
|
ToDeviceMsg toDeviceMsg = BasicStatusCodeResponse.onSuccess(data.getSessionMsgType(), data.getRequestId()); |
|
|
|
sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(toDeviceMsg, data.getSessionId()), data.getServerAddress()); |
|
|
|
@ -208,8 +211,8 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
} |
|
|
|
|
|
|
|
private void sendPendingRequests(ActorContext context, SessionId sessionId, SessionType type, Optional<ServerAddress> server) { |
|
|
|
if (!rpcPendingMap.isEmpty()) { |
|
|
|
logger.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, rpcPendingMap.size(), sessionId); |
|
|
|
if (!toDeviceRpcPendingMap.isEmpty()) { |
|
|
|
logger.debug("[{}] Pushing {} pending RPC messages to new async session [{}]", deviceId, toDeviceRpcPendingMap.size(), sessionId); |
|
|
|
if (type == SessionType.SYNC) { |
|
|
|
logger.debug("[{}] Cleanup sync rpc session [{}]", deviceId, sessionId); |
|
|
|
rpcSubscriptions.remove(sessionId); |
|
|
|
@ -219,12 +222,12 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
} |
|
|
|
Set<Integer> sentOneWayIds = new HashSet<>(); |
|
|
|
if (type == SessionType.ASYNC) { |
|
|
|
rpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, server, sentOneWayIds)); |
|
|
|
toDeviceRpcPendingMap.entrySet().forEach(processPendingRpc(context, sessionId, server, sentOneWayIds)); |
|
|
|
} else { |
|
|
|
rpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, server, sentOneWayIds)); |
|
|
|
toDeviceRpcPendingMap.entrySet().stream().findFirst().ifPresent(processPendingRpc(context, sessionId, server, sentOneWayIds)); |
|
|
|
} |
|
|
|
|
|
|
|
sentOneWayIds.forEach(rpcPendingMap::remove); |
|
|
|
sentOneWayIds.forEach(toDeviceRpcPendingMap::remove); |
|
|
|
} |
|
|
|
|
|
|
|
private Consumer<Map.Entry<Integer, ToDeviceRpcRequestMetadata>> processPendingRpc(ActorContext context, SessionId sessionId, Optional<ServerAddress> server, Set<Integer> sentOneWayIds) { |
|
|
|
@ -263,8 +266,8 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
handlePostTelemetryRequest(context, msg); |
|
|
|
break; |
|
|
|
case TO_SERVER_RPC_REQUEST: |
|
|
|
handleClientSideRPCRequest(context, msg); |
|
|
|
break; |
|
|
|
//TODO: push to queue and start processing!
|
|
|
|
} |
|
|
|
} |
|
|
|
} |
|
|
|
@ -345,11 +348,47 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
pushToRuleEngineWithTimeout(context, tbMsg, src, request); |
|
|
|
} |
|
|
|
|
|
|
|
private void handleClientSideRPCRequest(ActorContext context, DeviceToDeviceActorMsg src) { |
|
|
|
ToServerRpcRequestMsg request = (ToServerRpcRequestMsg) src.getPayload(); |
|
|
|
|
|
|
|
JsonObject json = new JsonObject(); |
|
|
|
json.addProperty("method", request.getMethod()); |
|
|
|
json.add("params", jsonParser.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)); |
|
|
|
pushToRuleEngineWithTimeout(context, tbMsg, src, request); |
|
|
|
|
|
|
|
scheduleMsgWithDelay(context, new DeviceActorClientSideRpcTimeoutMsg(request.getRequestId(), systemContext.getClientSideRpcTimeout()), systemContext.getClientSideRpcTimeout()); |
|
|
|
toServerRpcPendingMap.put(request.getRequestId(), new ToServerRpcRequestMetadata(src.getSessionId(), src.getSessionType(), src.getServerAddress())); |
|
|
|
} |
|
|
|
|
|
|
|
public void processClientSideRpcTimeout(ActorContext context, DeviceActorClientSideRpcTimeoutMsg msg) { |
|
|
|
ToServerRpcRequestMetadata data = toServerRpcPendingMap.remove(msg.getId()); |
|
|
|
if (data != null) { |
|
|
|
logger.debug("[{}] Client side RPC request [{}] timeout detected!", deviceId, msg.getId()); |
|
|
|
ToDeviceMsg toDeviceMsg = new RuleEngineErrorMsg(SessionMsgType.TO_SERVER_RPC_REQUEST, RuleEngineError.TIMEOUT); |
|
|
|
sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(toDeviceMsg, data.getSessionId()), data.getServer()); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
void processToServerRPCResponse(ActorContext context, ToServerRpcResponseActorMsg msg) { |
|
|
|
ToServerRpcRequestMetadata data = toServerRpcPendingMap.remove(msg.getMsg().getRequestId()); |
|
|
|
if (data != null) { |
|
|
|
sendMsgToSessionActor(new BasicToDeviceSessionActorMsg(msg.getMsg(), data.getSessionId()), data.getServer()); |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, DeviceToDeviceActorMsg src, FromDeviceRequestMsg fromDeviceRequestMsg) { |
|
|
|
pushToRuleEngineWithTimeout(context, tbMsg, src, fromDeviceRequestMsg, true); |
|
|
|
} |
|
|
|
|
|
|
|
private void pushToRuleEngineWithTimeout(ActorContext context, TbMsg tbMsg, DeviceToDeviceActorMsg src, FromDeviceRequestMsg fromDeviceRequestMsg, boolean replyOnAck) { |
|
|
|
SessionMsgType sessionMsgType = fromDeviceRequestMsg.getMsgType(); |
|
|
|
int requestId = fromDeviceRequestMsg.getRequestId(); |
|
|
|
if (systemContext.isQueuePersistenceEnabled()) { |
|
|
|
pendingMsgs.put(tbMsg.getId(), new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), sessionMsgType, requestId)); |
|
|
|
pendingMsgs.put(tbMsg.getId(), new PendingSessionMsgData(src.getSessionId(), src.getServerAddress(), sessionMsgType, requestId, replyOnAck)); |
|
|
|
scheduleMsgWithDelay(context, new DeviceActorQueueTimeoutMsg(tbMsg.getId(), systemContext.getQueuePersistenceTimeout()), systemContext.getQueuePersistenceTimeout()); |
|
|
|
} else { |
|
|
|
ToDeviceSessionActorMsg response = new BasicToDeviceSessionActorMsg(BasicStatusCodeResponse.onSuccess(sessionMsgType, requestId), src.getSessionId()); |
|
|
|
@ -394,7 +433,7 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso |
|
|
|
if (inMsg.getMsgType() == SessionMsgType.TO_DEVICE_RPC_RESPONSE) { |
|
|
|
logger.debug("[{}] Processing rpc command response [{}]", deviceId, sessionId); |
|
|
|
ToDeviceRpcResponseMsg responseMsg = (ToDeviceRpcResponseMsg) inMsg; |
|
|
|
ToDeviceRpcRequestMetadata requestMd = rpcPendingMap.remove(responseMsg.getRequestId()); |
|
|
|
ToDeviceRpcRequestMetadata requestMd = toDeviceRpcPendingMap.remove(responseMsg.getRequestId()); |
|
|
|
boolean success = requestMd != null; |
|
|
|
if (success) { |
|
|
|
systemContext.getDeviceRpcService().process(new FromDeviceRpcResponse(requestMd.getMsg().getMsg().getId(), responseMsg.getData(), null)); |
|
|
|
|
Andrew Shvayka
8 years ago