Andrew Shvayka
8 years ago
34 changed files with 595 additions and 531 deletions
@ -0,0 +1,148 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.controller; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.http.HttpStatus; |
|||
import org.springframework.http.ResponseEntity; |
|||
import org.springframework.security.access.prepost.PreAuthorize; |
|||
import org.springframework.web.bind.annotation.PathVariable; |
|||
import org.springframework.web.bind.annotation.RequestBody; |
|||
import org.springframework.web.bind.annotation.RequestMapping; |
|||
import org.springframework.web.bind.annotation.RequestMethod; |
|||
import org.springframework.web.bind.annotation.ResponseBody; |
|||
import org.springframework.web.bind.annotation.RestController; |
|||
import org.springframework.web.context.request.async.DeferredResult; |
|||
import org.thingsboard.server.actors.plugin.ValidationResult; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; |
|||
import org.thingsboard.server.common.data.exception.ThingsboardException; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.extensions.api.exception.ToErrorResponseEntity; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginConstants; |
|||
import org.thingsboard.server.common.data.rpc.RpcRequest; |
|||
import org.thingsboard.server.service.rpc.LocalRequestMetaData; |
|||
import org.thingsboard.server.service.rpc.RpcService; |
|||
import org.thingsboard.server.service.security.AccessValidator; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import javax.annotation.Nullable; |
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.io.IOException; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutorService; |
|||
import java.util.concurrent.Executors; |
|||
|
|||
/** |
|||
* Created by ashvayka on 22.03.18. |
|||
*/ |
|||
@RestController |
|||
@RequestMapping(PluginConstants.RPC_URL_PREFIX) |
|||
@Slf4j |
|||
public class RpcController extends BaseController { |
|||
|
|||
public static final int DEFAULT_TIMEOUT = 10000; |
|||
protected final ObjectMapper jsonMapper = new ObjectMapper(); |
|||
|
|||
@Autowired |
|||
private RpcService rpcService; |
|||
|
|||
@Autowired |
|||
private AccessValidator accessValidator; |
|||
|
|||
private ExecutorService executor; |
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
executor = Executors.newSingleThreadExecutor(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (executor != null) { |
|||
executor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/oneway/{deviceId}", method = RequestMethod.POST) |
|||
@ResponseBody |
|||
public DeferredResult<ResponseEntity> handleOneWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException { |
|||
return handleDeviceRPCRequest(true, new DeviceId(UUID.fromString(deviceIdStr)), requestBody); |
|||
} |
|||
|
|||
@PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") |
|||
@RequestMapping(value = "/twoway/{deviceId}", method = RequestMethod.POST) |
|||
@ResponseBody |
|||
public DeferredResult<ResponseEntity> handleTwoWayDeviceRPCRequest(@PathVariable("deviceId") String deviceIdStr, @RequestBody String requestBody) throws ThingsboardException { |
|||
return handleDeviceRPCRequest(false, new DeviceId(UUID.fromString(deviceIdStr)), requestBody); |
|||
} |
|||
|
|||
|
|||
private DeferredResult<ResponseEntity> handleDeviceRPCRequest(boolean oneWay, DeviceId deviceId, String requestBody) throws ThingsboardException { |
|||
try { |
|||
JsonNode rpcRequestBody = jsonMapper.readTree(requestBody); |
|||
RpcRequest cmd = new RpcRequest(rpcRequestBody.get("method").asText(), |
|||
jsonMapper.writeValueAsString(rpcRequestBody.get("params"))); |
|||
|
|||
if (rpcRequestBody.has("timeout")) { |
|||
cmd.setTimeout(rpcRequestBody.get("timeout").asLong()); |
|||
} |
|||
SecurityUser currentUser = getCurrentUser(); |
|||
TenantId tenantId = currentUser.getTenantId(); |
|||
final DeferredResult<ResponseEntity> response = new DeferredResult<>(); |
|||
long timeout = System.currentTimeMillis() + (cmd.getTimeout() != null ? cmd.getTimeout() : DEFAULT_TIMEOUT); |
|||
ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(cmd.getMethodName(), cmd.getRequestData()); |
|||
accessValidator.validate(currentUser, deviceId, new FutureCallback<ValidationResult>() { |
|||
@Override |
|||
public void onSuccess(@Nullable ValidationResult result) { |
|||
ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(UUID.randomUUID(), |
|||
tenantId, |
|||
deviceId, |
|||
oneWay, |
|||
timeout, |
|||
body |
|||
); |
|||
rpcService.process(rpcRequest, new LocalRequestMetaData(rpcRequest, currentUser, response)); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable e) { |
|||
ResponseEntity entity; |
|||
if (e instanceof ToErrorResponseEntity) { |
|||
entity = ((ToErrorResponseEntity) e).toErrorResponseEntity(); |
|||
} else { |
|||
entity = new ResponseEntity(HttpStatus.UNAUTHORIZED); |
|||
} |
|||
rpcService.logRpcCall(currentUser, deviceId, body, oneWay, Optional.empty(), e); |
|||
response.setResult(entity); |
|||
} |
|||
}); |
|||
return response; |
|||
} catch (IOException ioe) { |
|||
throw new ThingsboardException("Invalid request body", ioe, ThingsboardErrorCode.BAD_REQUEST_PARAMS); |
|||
} |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,201 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.rpc; |
|||
|
|||
import com.fasterxml.jackson.databind.ObjectMapper; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.http.HttpEntity; |
|||
import org.springframework.http.HttpStatus; |
|||
import org.springframework.http.ResponseEntity; |
|||
import org.springframework.stereotype.Service; |
|||
import org.springframework.util.StringUtils; |
|||
import org.springframework.web.context.request.async.DeferredResult; |
|||
import org.thingsboard.server.actors.service.ActorService; |
|||
import org.thingsboard.server.common.data.audit.ActionType; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.UUIDBased; |
|||
import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.controller.BaseController; |
|||
import org.thingsboard.server.dao.audit.AuditLogService; |
|||
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.cluster.routing.ClusterRoutingService; |
|||
import org.thingsboard.server.service.cluster.rpc.ClusterRpcService; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import javax.annotation.PostConstruct; |
|||
import javax.annotation.PreDestroy; |
|||
import java.io.IOException; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ConcurrentHashMap; |
|||
import java.util.concurrent.ConcurrentMap; |
|||
import java.util.concurrent.Executors; |
|||
import java.util.concurrent.ScheduledExecutorService; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.function.BiConsumer; |
|||
|
|||
/** |
|||
* Created by ashvayka on 27.03.18. |
|||
*/ |
|||
@Service |
|||
@Slf4j |
|||
public class DefaultRpcService implements RpcService { |
|||
|
|||
private static final ObjectMapper jsonMapper = new ObjectMapper(); |
|||
|
|||
@Autowired |
|||
private ClusterRoutingService routingService; |
|||
|
|||
@Autowired |
|||
private ClusterRpcService rpcService; |
|||
|
|||
@Autowired |
|||
private ActorService actorService; |
|||
|
|||
@Autowired |
|||
private AuditLogService auditLogService; |
|||
|
|||
private ScheduledExecutorService rpcCallBackExecutor; |
|||
|
|||
private final ConcurrentMap<UUID, LocalRequestMetaData> localRpcRequests = new ConcurrentHashMap<>(); |
|||
|
|||
|
|||
@PostConstruct |
|||
public void initExecutor() { |
|||
rpcCallBackExecutor = Executors.newSingleThreadScheduledExecutor(); |
|||
} |
|||
|
|||
@PreDestroy |
|||
public void shutdownExecutor() { |
|||
if (rpcCallBackExecutor != null) { |
|||
rpcCallBackExecutor.shutdownNow(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void process(ToDeviceRpcRequest request, LocalRequestMetaData metaData) { |
|||
log.trace("[{}] Processing local rpc call for device [{}]", request.getTenantId(), request.getDeviceId()); |
|||
sendRpcRequest(request); |
|||
UUID requestId = request.getId(); |
|||
localRpcRequests.put(requestId, metaData); |
|||
long timeout = Math.max(0, System.currentTimeMillis() - request.getExpirationTime()); |
|||
rpcCallBackExecutor.schedule(() -> { |
|||
LocalRequestMetaData localMetaData = localRpcRequests.remove(requestId); |
|||
if (localMetaData != null) { |
|||
reply(localMetaData, new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT)); |
|||
} |
|||
}, timeout, TimeUnit.MILLISECONDS); |
|||
} |
|||
|
|||
@Override |
|||
public void process(FromDeviceRpcResponse response) { |
|||
UUID requestId = response.getId(); |
|||
LocalRequestMetaData md = localRpcRequests.remove(requestId); |
|||
if (md != null) { |
|||
log.trace("[{}] Processing local rpc response from device [{}]", requestId, md.getRequest().getDeviceId()); |
|||
reply(md, response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
public void reply(LocalRequestMetaData rpcRequest, FromDeviceRpcResponse response) { |
|||
Optional<RpcError> rpcError = response.getError(); |
|||
DeferredResult<ResponseEntity> responseWriter = rpcRequest.getResponseWriter(); |
|||
if (rpcError.isPresent()) { |
|||
logRpcCall(rpcRequest, rpcError, null); |
|||
RpcError error = rpcError.get(); |
|||
switch (error) { |
|||
case TIMEOUT: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT)); |
|||
break; |
|||
case NO_ACTIVE_CONNECTION: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.CONFLICT)); |
|||
break; |
|||
default: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT)); |
|||
break; |
|||
} |
|||
} else { |
|||
Optional<String> responseData = response.getResponse(); |
|||
if (responseData.isPresent() && !StringUtils.isEmpty(responseData.get())) { |
|||
String data = responseData.get(); |
|||
try { |
|||
logRpcCall(rpcRequest, rpcError, null); |
|||
responseWriter.setResult(new ResponseEntity<>(jsonMapper.readTree(data), HttpStatus.OK)); |
|||
} catch (IOException e) { |
|||
log.debug("Failed to decode device response: {}", data, e); |
|||
logRpcCall(rpcRequest, rpcError, e); |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.NOT_ACCEPTABLE)); |
|||
} |
|||
} else { |
|||
logRpcCall(rpcRequest, rpcError, null); |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.OK)); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private void sendRpcRequest(ToDeviceRpcRequest msg) { |
|||
log.trace("[{}] Forwarding msg {} to device actor!", msg.getDeviceId(), msg); |
|||
ToDeviceRpcRequestMsg rpcMsg = new ToDeviceRpcRequestMsg(msg); |
|||
forward(msg.getDeviceId(), rpcMsg, rpcService::tell); |
|||
} |
|||
|
|||
private void forward(DeviceId deviceId, ToDeviceRpcRequestMsg msg, BiConsumer<ServerAddress, ToDeviceRpcRequestMsg> rpcFunction) { |
|||
Optional<ServerAddress> instance = routingService.resolveById(deviceId); |
|||
if (instance.isPresent()) { |
|||
log.trace("[{}] Forwarding msg {} to remote device actor!", msg.getTenantId(), msg); |
|||
rpcFunction.accept(instance.get(), msg); |
|||
} else { |
|||
log.trace("[{}] Forwarding msg {} to local device actor!", msg.getTenantId(), msg); |
|||
actorService.onMsg(msg); |
|||
} |
|||
} |
|||
|
|||
private void logRpcCall(LocalRequestMetaData rpcRequest, Optional<RpcError> rpcError, Throwable e) { |
|||
logRpcCall(rpcRequest.getUser(), rpcRequest.getRequest().getDeviceId(), rpcRequest.getRequest().getBody(), rpcRequest.getRequest().isOneway(), rpcError, null); |
|||
} |
|||
|
|||
@Override |
|||
public void logRpcCall(SecurityUser user, EntityId entityId, ToDeviceRpcRequestBody body, boolean oneWay, Optional<RpcError> rpcError, Throwable e) { |
|||
String rpcErrorStr = ""; |
|||
if (rpcError.isPresent()) { |
|||
rpcErrorStr = "RPC Error: " + rpcError.get().name(); |
|||
} |
|||
String method = body.getMethod(); |
|||
String params = body.getParams(); |
|||
|
|||
auditLogService.logEntityAction( |
|||
user.getTenantId(), |
|||
user.getCustomerId(), |
|||
user.getId(), |
|||
user.getName(), |
|||
(UUIDBased & EntityId) entityId, |
|||
null, |
|||
ActionType.RPC_CALL, |
|||
BaseController.toException(e), |
|||
rpcErrorStr, |
|||
oneWay, |
|||
method, |
|||
params); |
|||
} |
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* Copyright © 2016-2018 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.service.rpc; |
|||
|
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.rpc.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.FromDeviceRpcResponse; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.RpcError; |
|||
import org.thingsboard.server.service.security.model.SecurityUser; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 16.04.18. |
|||
*/ |
|||
public interface RpcService { |
|||
|
|||
void process(ToDeviceRpcRequest request, LocalRequestMetaData metaData); |
|||
|
|||
void process(FromDeviceRpcResponse response); |
|||
|
|||
void logRpcCall(SecurityUser user, EntityId entityId, ToDeviceRpcRequestBody body, boolean oneWay, Optional<RpcError> rpcError, Throwable e); |
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.msg.cluster.ServerAddress; |
|||
import org.thingsboard.server.common.msg.rpc.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.extensions.api.device.ToDeviceActorNotificationMsg; |
|||
|
|||
import java.util.Optional; |
|||
|
|||
/** |
|||
* Created by ashvayka on 16.04.18. |
|||
*/ |
|||
@ToString |
|||
@RequiredArgsConstructor |
|||
public class ToDeviceRpcRequestMsg implements ToDeviceActorNotificationMsg { |
|||
|
|||
private final ServerAddress serverAddress; |
|||
@Getter |
|||
private final ToDeviceRpcRequest msg; |
|||
|
|||
public ToDeviceRpcRequestMsg(ToDeviceRpcRequest msg) { |
|||
this(null, msg); |
|||
} |
|||
|
|||
public Optional<ServerAddress> getServerAddress() { |
|||
return Optional.ofNullable(serverAddress); |
|||
} |
|||
|
|||
@Override |
|||
public DeviceId getDeviceId() { |
|||
return msg.getDeviceId(); |
|||
} |
|||
|
|||
@Override |
|||
public TenantId getTenantId() { |
|||
return msg.getTenantId(); |
|||
} |
|||
} |
|||
@ -1,69 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.extensions.core.plugin.rpc; |
|||
|
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginContext; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.*; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.handlers.RpcRestMsgHandler; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
public class RpcManager { |
|||
|
|||
@Setter |
|||
private RpcRestMsgHandler restHandler; |
|||
|
|||
private Map<UUID, LocalRequestMetaData> localRpcRequests = new HashMap<>(); |
|||
|
|||
public void process(PluginContext ctx, LocalRequestMetaData requestMd) { |
|||
ToDeviceRpcRequest request = requestMd.getRequest(); |
|||
log.trace("[{}] Processing local rpc call for device [{}]", request.getId(), request.getDeviceId()); |
|||
ctx.sendRpcRequest(request); |
|||
localRpcRequests.put(request.getId(), requestMd); |
|||
ctx.scheduleTimeoutMsg(new TimeoutUUIDMsg(request.getId(), request.getExpirationTime() - System.currentTimeMillis())); |
|||
} |
|||
|
|||
public void process(PluginContext ctx, FromDeviceRpcResponse response) { |
|||
UUID requestId = response.getId(); |
|||
LocalRequestMetaData md = localRpcRequests.remove(requestId); |
|||
if (md != null) { |
|||
log.trace("[{}] Processing local rpc response from device [{}]", requestId, md.getRequest().getDeviceId()); |
|||
restHandler.reply(ctx, md.getRequest(), md.getResponseWriter(), response); |
|||
} else { |
|||
log.trace("[{}] Unknown or stale rpc response received [{}]", requestId, response); |
|||
} |
|||
} |
|||
|
|||
public void process(PluginContext ctx, TimeoutMsg msg) { |
|||
if (msg instanceof TimeoutUUIDMsg) { |
|||
UUID requestId = ((TimeoutUUIDMsg) msg).getId(); |
|||
FromDeviceRpcResponse timeoutReponse = new FromDeviceRpcResponse(requestId, null, RpcError.TIMEOUT); |
|||
LocalRequestMetaData md = localRpcRequests.remove(requestId); |
|||
if (md != null) { |
|||
log.trace("[{}] Processing rpc timeout for local device [{}]", requestId, md.getRequest().getDeviceId()); |
|||
restHandler.reply(ctx, md.getRequest(), md.getResponseWriter(), timeoutReponse); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,86 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.extensions.core.plugin.rpc; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.server.extensions.api.component.Plugin; |
|||
import org.thingsboard.server.extensions.api.plugins.AbstractPlugin; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginContext; |
|||
import org.thingsboard.server.extensions.api.plugins.handlers.DefaultRuleMsgHandler; |
|||
import org.thingsboard.server.extensions.api.plugins.handlers.RestMsgHandler; |
|||
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.FromDeviceRpcResponse; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.TimeoutMsg; |
|||
import org.thingsboard.server.extensions.core.action.rpc.ServerSideRpcCallAction; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.handlers.RpcRestMsgHandler; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.handlers.RpcRuleMsgHandler; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Plugin(name = "RPC Plugin", actions = {ServerSideRpcCallAction.class}, descriptor = "RpcPluginDescriptor.json", configuration = RpcPluginConfiguration.class) |
|||
@Slf4j |
|||
public class RpcPlugin extends AbstractPlugin<RpcPluginConfiguration> { |
|||
|
|||
private final RpcManager rpcManager; |
|||
private final RpcRestMsgHandler restMsgHandler; |
|||
|
|||
public RpcPlugin() { |
|||
this.rpcManager = new RpcManager(); |
|||
this.restMsgHandler = new RpcRestMsgHandler(rpcManager); |
|||
this.rpcManager.setRestHandler(restMsgHandler); |
|||
} |
|||
|
|||
@Override |
|||
public void process(PluginContext ctx, FromDeviceRpcResponse msg) { |
|||
rpcManager.process(ctx, msg); |
|||
} |
|||
|
|||
@Override |
|||
public void process(PluginContext ctx, TimeoutMsg<?> msg) { |
|||
rpcManager.process(ctx, msg); |
|||
} |
|||
|
|||
@Override |
|||
protected RestMsgHandler getRestMsgHandler() { |
|||
return restMsgHandler; |
|||
} |
|||
|
|||
@Override |
|||
public void init(RpcPluginConfiguration configuration) { |
|||
restMsgHandler.setDefaultTimeout(configuration.getDefaultTimeout()); |
|||
} |
|||
|
|||
@Override |
|||
protected RuleMsgHandler getRuleMsgHandler() { |
|||
return new RpcRuleMsgHandler(); |
|||
} |
|||
|
|||
@Override |
|||
public void resume(PluginContext ctx) { |
|||
//Do nothing
|
|||
} |
|||
|
|||
@Override |
|||
public void suspend(PluginContext ctx) { |
|||
//Do nothing
|
|||
} |
|||
|
|||
@Override |
|||
public void stop(PluginContext ctx) { |
|||
//Do nothing
|
|||
} |
|||
} |
|||
@ -1,26 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.extensions.core.plugin.rpc; |
|||
|
|||
import lombok.Data; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Data |
|||
public class RpcPluginConfiguration { |
|||
private long defaultTimeout; |
|||
} |
|||
@ -1,161 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.extensions.core.plugin.rpc.handlers; |
|||
|
|||
import com.fasterxml.jackson.core.JsonProcessingException; |
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.Setter; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.http.HttpStatus; |
|||
import org.springframework.http.ResponseEntity; |
|||
import org.springframework.util.StringUtils; |
|||
import org.springframework.web.context.request.async.DeferredResult; |
|||
import org.thingsboard.server.common.data.DataConstants; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.extensions.api.exception.ToErrorResponseEntity; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginApiCallSecurityContext; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginCallback; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginContext; |
|||
import org.thingsboard.server.extensions.api.plugins.handlers.DefaultRestMsgHandler; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.FromDeviceRpcResponse; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.RpcError; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.extensions.api.plugins.rest.PluginRestMsg; |
|||
import org.thingsboard.server.extensions.api.plugins.rest.RestRequest; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.LocalRequestMetaData; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.RpcManager; |
|||
import org.thingsboard.server.extensions.core.plugin.rpc.cmd.RpcRequest; |
|||
|
|||
import javax.servlet.ServletException; |
|||
import java.io.IOException; |
|||
import java.util.Optional; |
|||
import java.util.UUID; |
|||
|
|||
/** |
|||
* @author Andrew Shvayka |
|||
*/ |
|||
@Slf4j |
|||
@RequiredArgsConstructor |
|||
public class RpcRestMsgHandler extends DefaultRestMsgHandler { |
|||
|
|||
private final RpcManager rpcManager; |
|||
@Setter |
|||
private long defaultTimeout; |
|||
|
|||
@Override |
|||
public void handleHttpPostRequest(PluginContext ctx, PluginRestMsg msg) throws ServletException { |
|||
boolean valid = false; |
|||
RestRequest request = msg.getRequest(); |
|||
try { |
|||
String[] pathParams = request.getPathParams(); |
|||
if (pathParams.length == 2) { |
|||
String method = pathParams[0].toUpperCase(); |
|||
if (DataConstants.ONEWAY.equals(method) || DataConstants.TWOWAY.equals(method)) { |
|||
final TenantId tenantId = ctx.getSecurityCtx().orElseThrow(() -> new IllegalStateException("Security context is empty!")).getTenantId(); |
|||
JsonNode rpcRequestBody = jsonMapper.readTree(request.getRequestBody()); |
|||
|
|||
RpcRequest cmd = new RpcRequest(rpcRequestBody.get("method").asText(), |
|||
jsonMapper.writeValueAsString(rpcRequestBody.get("params"))); |
|||
if (rpcRequestBody.has("timeout")) { |
|||
cmd.setTimeout(rpcRequestBody.get("timeout").asLong()); |
|||
} |
|||
|
|||
boolean oneWay = DataConstants.ONEWAY.equals(method); |
|||
|
|||
DeviceId deviceId = DeviceId.fromString(pathParams[1]); |
|||
valid = handleDeviceRPCRequest(ctx, msg, tenantId, deviceId, cmd, oneWay); |
|||
} |
|||
} |
|||
} catch (IOException e) { |
|||
log.debug("Failed to process POST request due to IO exception", e); |
|||
} catch (RuntimeException e) { |
|||
log.debug("Failed to process POST request due to Runtime exception", e); |
|||
} |
|||
if (!valid) { |
|||
msg.getResponseHolder().setResult(new ResponseEntity<>(HttpStatus.BAD_REQUEST)); |
|||
} |
|||
} |
|||
|
|||
private boolean handleDeviceRPCRequest(PluginContext ctx, final PluginRestMsg msg, TenantId tenantId, DeviceId deviceId, RpcRequest cmd, boolean oneWay) throws JsonProcessingException { |
|||
long timeout = System.currentTimeMillis() + (cmd.getTimeout() != null ? cmd.getTimeout() : defaultTimeout); |
|||
ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(cmd.getMethodName(), cmd.getRequestData()); |
|||
ctx.checkAccess(deviceId, new PluginCallback<Void>() { |
|||
@Override |
|||
public void onSuccess(PluginContext ctx, Void value) { |
|||
ToDeviceRpcRequest rpcRequest = new ToDeviceRpcRequest(UUID.randomUUID(), |
|||
msg.getSecurityCtx(), |
|||
tenantId, |
|||
deviceId, |
|||
oneWay, |
|||
timeout, |
|||
body |
|||
); |
|||
rpcManager.process(ctx, new LocalRequestMetaData(rpcRequest, msg.getResponseHolder())); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(PluginContext ctx, Exception e) { |
|||
ResponseEntity response; |
|||
if (e instanceof ToErrorResponseEntity) { |
|||
response = ((ToErrorResponseEntity)e).toErrorResponseEntity(); |
|||
} else { |
|||
response = new ResponseEntity(HttpStatus.UNAUTHORIZED); |
|||
} |
|||
ctx.logRpcRequest(msg.getSecurityCtx(), deviceId, body, oneWay, Optional.empty(), e); |
|||
msg.getResponseHolder().setResult(response); |
|||
} |
|||
}); |
|||
return true; |
|||
} |
|||
|
|||
public void reply(PluginContext ctx, ToDeviceRpcRequest rpcRequest, DeferredResult<ResponseEntity> responseWriter, FromDeviceRpcResponse response) { |
|||
Optional<RpcError> rpcError = response.getError(); |
|||
if (rpcError.isPresent()) { |
|||
ctx.logRpcRequest(rpcRequest.getSecurityCtx(), rpcRequest.getDeviceId(), rpcRequest.getBody(), rpcRequest.isOneway(), rpcError, null); |
|||
RpcError error = rpcError.get(); |
|||
switch (error) { |
|||
case TIMEOUT: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT)); |
|||
break; |
|||
case NO_ACTIVE_CONNECTION: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.CONFLICT)); |
|||
break; |
|||
default: |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.REQUEST_TIMEOUT)); |
|||
break; |
|||
} |
|||
} else { |
|||
Optional<String> responseData = response.getResponse(); |
|||
if (responseData.isPresent() && !StringUtils.isEmpty(responseData.get())) { |
|||
String data = responseData.get(); |
|||
try { |
|||
ctx.logRpcRequest(rpcRequest.getSecurityCtx(), rpcRequest.getDeviceId(), rpcRequest.getBody(), rpcRequest.isOneway(), rpcError, null); |
|||
responseWriter.setResult(new ResponseEntity<>(jsonMapper.readTree(data), HttpStatus.OK)); |
|||
} catch (IOException e) { |
|||
log.debug("Failed to decode device response: {}", data, e); |
|||
ctx.logRpcRequest(rpcRequest.getSecurityCtx(), rpcRequest.getDeviceId(), rpcRequest.getBody(), rpcRequest.isOneway(), rpcError, e); |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.NOT_ACCEPTABLE)); |
|||
} |
|||
} else { |
|||
ctx.logRpcRequest(rpcRequest.getSecurityCtx(), rpcRequest.getDeviceId(), rpcRequest.getBody(), rpcRequest.isOneway(), rpcError, null); |
|||
responseWriter.setResult(new ResponseEntity<>(HttpStatus.OK)); |
|||
} |
|||
} |
|||
} |
|||
} |
|||
@ -1,102 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.extensions.core.plugin.rpc.handlers; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.util.StringUtils; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginCallback; |
|||
import org.thingsboard.server.extensions.api.plugins.PluginContext; |
|||
import org.thingsboard.server.extensions.api.plugins.handlers.RuleMsgHandler; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.RuleToPluginMsg; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.ToDeviceRpcRequest; |
|||
import org.thingsboard.server.extensions.api.plugins.msg.ToDeviceRpcRequestBody; |
|||
import org.thingsboard.server.extensions.api.rules.RuleException; |
|||
import org.thingsboard.server.extensions.core.action.rpc.ServerSideRpcCallActionMsg; |
|||
import org.thingsboard.server.extensions.core.action.rpc.ServerSideRpcCallRuleToPluginActionMsg; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.TimeUnit; |
|||
import java.util.stream.Collectors; |
|||
|
|||
/** |
|||
* Created by ashvayka on 14.09.17. |
|||
*/ |
|||
@Slf4j |
|||
public class RpcRuleMsgHandler implements RuleMsgHandler { |
|||
|
|||
@Override |
|||
public void process(PluginContext ctx, TenantId tenantId, RuleId ruleId, RuleToPluginMsg<?> msg) throws RuleException { |
|||
if (msg instanceof ServerSideRpcCallRuleToPluginActionMsg) { |
|||
handle(ctx, tenantId, ruleId, ((ServerSideRpcCallRuleToPluginActionMsg) msg).getPayload()); |
|||
} else { |
|||
throw new RuntimeException("Not supported msg: " + msg + "!"); |
|||
} |
|||
} |
|||
|
|||
private void handle(final PluginContext ctx, TenantId tenantId, RuleId ruleId, ServerSideRpcCallActionMsg msg) { |
|||
DeviceId deviceId = new DeviceId(UUID.fromString(msg.getDeviceId())); |
|||
ctx.checkAccess(deviceId, new PluginCallback<Void>() { |
|||
@Override |
|||
public void onSuccess(PluginContext dummy, Void value) { |
|||
try { |
|||
List<EntityId> deviceIds; |
|||
if (StringUtils.isEmpty(msg.getFromDeviceRelation()) && StringUtils.isEmpty(msg.getToDeviceRelation())) { |
|||
deviceIds = Collections.singletonList(deviceId); |
|||
} else if (!StringUtils.isEmpty(msg.getFromDeviceRelation())) { |
|||
List<EntityRelation> relations = ctx.findByFromAndType(deviceId, msg.getFromDeviceRelation()).get(); |
|||
deviceIds = relations.stream().map(EntityRelation::getTo).collect(Collectors.toList()); |
|||
} else { |
|||
List<EntityRelation> relations = ctx.findByToAndType(deviceId, msg.getFromDeviceRelation()).get(); |
|||
deviceIds = relations.stream().map(EntityRelation::getFrom).collect(Collectors.toList()); |
|||
} |
|||
ToDeviceRpcRequestBody body = new ToDeviceRpcRequestBody(msg.getRpcCallMethod(), msg.getRpcCallBody()); |
|||
long expirationTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(msg.getRpcCallTimeoutInSec()); |
|||
for (EntityId address : deviceIds) { |
|||
DeviceId tmpId = new DeviceId(address.getId()); |
|||
ctx.checkAccess(tmpId, new PluginCallback<Void>() { |
|||
@Override |
|||
public void onSuccess(PluginContext ctx, Void value) { |
|||
ctx.sendRpcRequest(new ToDeviceRpcRequest(UUID.randomUUID(), |
|||
null, tenantId, tmpId, true, expirationTime, body) |
|||
); |
|||
log.trace("[{}] Sent RPC Call Action msg", tmpId); |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(PluginContext ctx, Exception e) { |
|||
log.info("[{}] Failed to process RPC Call Action msg", tmpId, e); |
|||
} |
|||
}); |
|||
} |
|||
} catch (Exception e) { |
|||
log.info("Failed to process RPC Call Action msg", e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(PluginContext dummy, Exception e) { |
|||
log.info("[{}] Failed to process RPC Call Action msg", deviceId, e); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
Loading…
Reference in new issue