From 6823b8969f478ef6628ef7841add7b57aee147a4 Mon Sep 17 00:00:00 2001 From: Andrew Shvayka Date: Fri, 30 Mar 2018 15:44:54 +0300 Subject: [PATCH] Refactoring of Telemetry Websockets --- .../server/config/WebSocketConfiguration.java | 8 +- .../controller/TelemetryController.java | 556 +++++++++++------- .../server/controller/ValidationCallback.java | 8 +- .../plugin/PluginApiController.java | 110 ++-- ...etHandler.java => TbWebSocketHandler.java} | 139 ++--- .../service/security/AccessValidator.java | 262 +++++++++ .../DefaultTelemetrySubscriptionService.java | 42 ++ .../DefaultTelemetryWebSocketService.java | 261 ++++++++ .../TelemetrySubscriptionService.java | 24 + .../TelemetryWebSocketMsgEndpoint.java | 14 + .../telemetry/TelemetryWebSocketService.java | 13 + .../TelemetryWebSocketSessionRef.java | 53 ++ .../telemetry/TelemetryWebSocketTextMsg.java | 19 + .../service/telemetry/WsSessionMetaData.java | 38 ++ .../common/data/id/EntityIdFactory.java | 4 + .../plugin/telemetry/SubscriptionManager.java | 4 +- .../handlers/TelemetryRpcMsgHandler.java | 16 +- 17 files changed, 1209 insertions(+), 362 deletions(-) rename application/src/main/java/org/thingsboard/server/controller/plugin/{PluginWebSocketHandler.java => TbWebSocketHandler.java} (69%) create mode 100644 application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketSessionRef.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketTextMsg.java create mode 100644 application/src/main/java/org/thingsboard/server/service/telemetry/WsSessionMetaData.java diff --git a/application/src/main/java/org/thingsboard/server/config/WebSocketConfiguration.java b/application/src/main/java/org/thingsboard/server/config/WebSocketConfiguration.java index a75ecb1353..d44c50e9cd 100644 --- a/application/src/main/java/org/thingsboard/server/config/WebSocketConfiguration.java +++ b/application/src/main/java/org/thingsboard/server/config/WebSocketConfiguration.java @@ -19,7 +19,7 @@ import java.util.Map; import org.thingsboard.server.exception.ThingsboardErrorCode; import org.thingsboard.server.exception.ThingsboardException; -import org.thingsboard.server.controller.plugin.PluginWebSocketHandler; +import org.thingsboard.server.controller.plugin.TbWebSocketHandler; import org.thingsboard.server.service.security.model.SecurityUser; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -54,7 +54,7 @@ public class WebSocketConfiguration implements WebSocketConfigurer { @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { - registry.addHandler(pluginWsHandler(), WS_PLUGIN_MAPPING).setAllowedOrigins("*") + registry.addHandler(wsHandler(), WS_PLUGIN_MAPPING).setAllowedOrigins("*") .addInterceptors(new HttpSessionHandshakeInterceptor(), new HandshakeInterceptor() { @Override @@ -82,8 +82,8 @@ public class WebSocketConfiguration implements WebSocketConfigurer { } @Bean - public WebSocketHandler pluginWsHandler() { - return new PluginWebSocketHandler(); + public WebSocketHandler wsHandler() { + return new TbWebSocketHandler(); } protected SecurityUser getCurrentUser() throws ThingsboardException { diff --git a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java index 1944fa2e68..a2867cecdc 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java @@ -1,9 +1,27 @@ +/** + * 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.google.common.base.Function; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; +import com.google.gson.JsonElement; +import com.google.gson.JsonParser; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; @@ -11,47 +29,54 @@ import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.util.StringUtils; 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.RequestParam; -import org.springframework.web.bind.annotation.ResponseStatus; +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.Customer; import org.thingsboard.server.common.data.DataConstants; -import org.thingsboard.server.common.data.Device; -import org.thingsboard.server.common.data.Tenant; -import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.audit.ActionType; -import org.thingsboard.server.common.data.id.AssetId; -import org.thingsboard.server.common.data.id.CustomerId; -import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.EntityIdFactory; -import org.thingsboard.server.common.data.id.RuleChainId; -import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UUIDBased; +import org.thingsboard.server.common.data.kv.Aggregation; import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseTsKvQuery; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.BooleanDataEntry; +import org.thingsboard.server.common.data.kv.DoubleDataEntry; import org.thingsboard.server.common.data.kv.KvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; -import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.common.data.kv.TsKvQuery; +import org.thingsboard.server.common.msg.core.TelemetryUploadRequest; +import org.thingsboard.server.common.transport.adaptor.JsonConverter; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.timeseries.TimeseriesService; import org.thingsboard.server.exception.ThingsboardException; -import org.thingsboard.server.extensions.api.exception.ToErrorResponseEntity; +import org.thingsboard.server.extensions.api.exception.InvalidParametersException; +import org.thingsboard.server.extensions.api.exception.UncheckedApiException; import org.thingsboard.server.extensions.api.plugins.PluginConstants; import org.thingsboard.server.extensions.core.plugin.telemetry.AttributeData; +import org.thingsboard.server.extensions.core.plugin.telemetry.TsData; +import org.thingsboard.server.service.security.AccessValidator; import org.thingsboard.server.service.security.model.SecurityUser; +import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; import javax.annotation.Nullable; import javax.annotation.PreDestroy; import java.util.ArrayList; import java.util.Arrays; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; -import java.util.function.BiConsumer; import java.util.stream.Collectors; /** @@ -62,9 +87,8 @@ import java.util.stream.Collectors; @Slf4j public class TelemetryController extends BaseController { - public static final String CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION = "Customer user is not allowed to perform this operation!"; - public static final String SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION = "System administrator is not allowed to perform this operation!"; - public static final String DEVICE_WITH_REQUESTED_ID_NOT_FOUND = "Device with requested id wasn't found!"; + @Autowired + private TelemetrySubscriptionService subscriptionService; @Autowired private AttributesService attributesService; @@ -72,6 +96,9 @@ public class TelemetryController extends BaseController { @Autowired private TimeseriesService tsService; + @Autowired + private AccessValidator accessValidator; + private ExecutorService executor; public void initExecutor() { @@ -87,117 +114,277 @@ public class TelemetryController extends BaseController { @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/keys/ATTRIBUTES", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getAttributeKeys( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr) throws ThingsboardException { - return validateEntityAndCallback(entityType, entityIdStr, - this::getAttributeKeysCallback, - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, this::getAttributeKeysCallback); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/keys/ATTRIBUTES/{scope}", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getAttributeKeysByScope( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr , @PathVariable("scope") String scope) throws ThingsboardException { - return validateEntityAndCallback(entityType, entityIdStr, - (result, entityId) -> getAttributeKeysCallback(result, entityId, scope), - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, + (result, entityId) -> getAttributeKeysCallback(result, entityId, scope)); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/values/ATTRIBUTES", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getAttributes( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, @RequestParam(name = "keys", required = false) String keysStr) throws ThingsboardException { SecurityUser user = getCurrentUser(); - return validateEntityAndCallback(entityType, entityIdStr, - (result, entityId) -> getAttributeValuesCallback(result, user, entityId, null, keysStr), - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, + (result, entityId) -> getAttributeValuesCallback(result, user, entityId, null, keysStr)); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/values/ATTRIBUTES/{scope}", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getAttributesByScope( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, @PathVariable("scope") String scope, @RequestParam(name = "keys", required = false) String keysStr) throws ThingsboardException { SecurityUser user = getCurrentUser(); - return validateEntityAndCallback(entityType, entityIdStr, - (result, entityId) -> getAttributeValuesCallback(result, user, entityId, scope, keysStr), - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, + (result, entityId) -> getAttributeValuesCallback(result, user, entityId, scope, keysStr)); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/keys/TIMESERIES", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getTimeseriesKeys( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr) throws ThingsboardException { - return validateEntityAndCallback(entityType, entityIdStr, + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, (result, entityId) -> { Futures.addCallback(tsService.findAllLatest(entityId), getTsKeysToResponseCallback(result)); - }, - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + }); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/values/TIMESERIES", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) + @ResponseBody public DeferredResult getLatestTimeseries( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, - @PathVariable("scope") String scope, @RequestParam(name = "keys", required = false) String keysStr) throws ThingsboardException { SecurityUser user = getCurrentUser(); - return validateEntityAndCallback(entityType, entityIdStr, - (result, entityId) -> getAttributeValuesCallback(result, user, entityId, scope, keysStr), - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, + (result, entityId) -> getLatestTimeseriesValuesCallback(result, user, entityId, keysStr)); } @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") @RequestMapping(value = "/{entityType}/{entityId}/values/TIMESERIES", method = RequestMethod.GET) - @ResponseStatus(value = HttpStatus.OK) - public DeferredResult getLatestTimeseries( + @ResponseBody + public DeferredResult getTimeseries( @PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, - @PathVariable("scope") String scope, - @RequestParam(name = "keys", required = false) String keysStr) throws ThingsboardException { - SecurityUser user = getCurrentUser(); + @RequestParam(name = "keys") String keys, + @RequestParam(name = "startTs") Long startTs, + @RequestParam(name = "endTs") Long endTs, + @RequestParam(name = "interval", defaultValue = "0") Long interval, + @RequestParam(name = "limit", defaultValue = "100") Integer limit, + @RequestParam(name = "agg", defaultValue = "NONE") String aggStr + ) throws ThingsboardException { + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityType, entityIdStr, + (result, entityId) -> { + // If interval is 0, convert this to a NONE aggregation, which is probably what the user really wanted + Aggregation agg = interval == 0L ? Aggregation.valueOf(Aggregation.NONE.name()) : Aggregation.valueOf(aggStr); + List queries = toKeysList(keys).stream().map(key -> new BaseTsKvQuery(key, startTs, endTs, interval, limit, agg)) + .collect(Collectors.toList()); + + Futures.addCallback(tsService.findAll(entityId, queries), getTsKvListCallback(result)); + }); + } + + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{deviceId}/{scope}", method = RequestMethod.POST) + @ResponseBody + public DeferredResult saveDeviceAttributes(@PathVariable("deviceId") String deviceIdStr, @PathVariable("scope") String scope, + @RequestBody JsonNode request) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE, deviceIdStr); + return saveAttributes(entityId, scope, request); + } + + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/{scope}", method = RequestMethod.POST) + @ResponseBody + public DeferredResult saveEntityAttributesV1(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @PathVariable("scope") String scope, + @RequestBody JsonNode request) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return saveAttributes(entityId, scope, request); + } - return validateEntityAndCallback(entityType, entityIdStr, - (result, entityId) -> getAttributeValuesCallback(result, user, entityId, scope, keysStr), - (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/ATTRIBUTES/{scope}", method = RequestMethod.POST) + @ResponseBody + public DeferredResult saveEntityAttributesV2(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @PathVariable("scope") String scope, + @RequestBody JsonNode request) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return saveAttributes(entityId, scope, request); + } + + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/TIMESERIES/{scope}", method = RequestMethod.POST) + @ResponseBody + public DeferredResult saveEntityTelemetry(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @PathVariable("scope") String scope, + @RequestBody String requestBody) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return saveTelemetry(entityId, requestBody, 0L); + } + + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/TIMESERIES/{scope}/{ttl}", method = RequestMethod.POST) + @ResponseBody + public DeferredResult saveEntityTelemetryWithTTL(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @PathVariable("scope") String scope, @PathVariable("ttl") Long ttl, + @RequestBody String requestBody) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return saveTelemetry(entityId, requestBody, ttl); } - private DeferredResult validateEntityAndCallback(String entityType, String entityIdStr, - BiConsumer, EntityId> onSuccess, BiConsumer, Throwable> onFailure) throws ThingsboardException { - final DeferredResult response = new DeferredResult<>(); + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{deviceId}/{scope}", method = RequestMethod.DELETE) + @ResponseBody + public DeferredResult deleteEntityAttributes(@PathVariable("deviceId") String deviceIdStr, + @PathVariable("scope") String scope, + @RequestParam(name = "keys") String keysStr) throws ThingsboardException { + EntityId entityId = EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE, deviceIdStr); + return deleteAttributes(entityId, scope, keysStr); + } + + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @RequestMapping(value = "/{entityType}/{entityId}/{scope}", method = RequestMethod.DELETE) + @ResponseBody + public DeferredResult deleteEntityAttributes(@PathVariable("entityType") String entityType, @PathVariable("entityId") String entityIdStr, + @PathVariable("scope") String scope, + @RequestParam(name = "keys") String keysStr) throws ThingsboardException { EntityId entityId = EntityIdFactory.getByTypeAndId(entityType, entityIdStr); + return deleteAttributes(entityId, scope, keysStr); + } + + private DeferredResult deleteAttributes(EntityId entityIdStr, String scope, String keysStr) throws ThingsboardException { + List keys = toKeysList(keysStr); + if (keys.isEmpty()) { + return getImmediateDeferredResult("Empty keys: " + keysStr, HttpStatus.BAD_REQUEST); + } + SecurityUser user = getCurrentUser(); + if (DataConstants.SERVER_SCOPE.equals(scope) || + DataConstants.SHARED_SCOPE.equals(scope) || + DataConstants.CLIENT_SCOPE.equals(scope)) { + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityIdStr, (result, entityId) -> { + ListenableFuture> future = attributesService.removeAll(entityId, scope, keys); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List tmp) { + logAttributesDeleted(user, entityId, scope, keys, null); + result.setResult(new ResponseEntity<>(HttpStatus.OK)); + } - validate(getCurrentUser(), entityId, new ValidationCallback(response, - new FutureCallback>() { @Override - public void onSuccess(@Nullable DeferredResult result) { - onSuccess.accept(response, entityId); + public void onFailure(Throwable t) { + logAttributesDeleted(user, entityId, scope, keys, t); + result.setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR)); + } + }, executor); + }); + } else { + return getImmediateDeferredResult("Invalid attribute scope: " + scope, HttpStatus.BAD_REQUEST); + } + } + + private DeferredResult saveAttributes(EntityId entityIdSrc, String scope, JsonNode json) throws ThingsboardException { + if (!DataConstants.SERVER_SCOPE.equals(scope) && !DataConstants.SHARED_SCOPE.equals(scope)) { + return getImmediateDeferredResult("Invalid scope: " + scope, HttpStatus.BAD_REQUEST); + } + if (json.isObject()) { + List attributes = extractRequestAttributes(json); + if (attributes.isEmpty()) { + return getImmediateDeferredResult("No attributes data found in request body!", HttpStatus.BAD_REQUEST); + } + SecurityUser user = getCurrentUser(); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityIdSrc, (result, entityId) -> { + ListenableFuture> future = attributesService.save(entityId, scope, attributes); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List tmp) { + logAttributesUpdated(user, entityId, scope, attributes, null); + result.setResult(new ResponseEntity(HttpStatus.OK)); + subscriptionService.onAttributesUpdateFromServer(entityId, scope, attributes); } @Override public void onFailure(Throwable t) { - onFailure.accept(response, t); + logAttributesUpdated(user, entityId, scope, attributes, t); + AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); } - })); + }); + result.setResult(new ResponseEntity(HttpStatus.OK)); + }); + } else { + return getImmediateDeferredResult("Request is not a JSON object", HttpStatus.BAD_REQUEST); + } + } + + private DeferredResult saveTelemetry(EntityId entityIdSrc, String requestBody, long ttl) throws ThingsboardException { + TelemetryUploadRequest telemetryRequest; + JsonElement telemetryJson; + try { + telemetryJson = new JsonParser().parse(requestBody); + } catch (Exception e) { + return getImmediateDeferredResult("Unable to parse timeseries payload: Invalid JSON body!", HttpStatus.BAD_REQUEST); + } + try { + telemetryRequest = JsonConverter.convertToTelemetry(telemetryJson); + } catch (Exception e) { + return getImmediateDeferredResult("Unable to parse timeseries payload. Invalid JSON body: " + e.getMessage(), HttpStatus.BAD_REQUEST); + } + List entries = new ArrayList<>(); + for (Map.Entry> entry : telemetryRequest.getData().entrySet()) { + for (KvEntry kv : entry.getValue()) { + entries.add(new BasicTsKvEntry(entry.getKey(), kv)); + } + } + if (entries.isEmpty()) { + return getImmediateDeferredResult("No timeseries data found in request body!", HttpStatus.BAD_REQUEST); + } + SecurityUser user = getCurrentUser(); + return accessValidator.validateEntityAndCallback(getCurrentUser(), entityIdSrc, (result, entityId) -> { + ListenableFuture> future = tsService.save(entityId, entries, ttl); + Futures.addCallback(future, new FutureCallback>() { + @Override + public void onSuccess(@Nullable List tmp) { + result.setResult(new ResponseEntity(HttpStatus.OK)); + subscriptionService.onTimeseriesUpdateFromServer(entityId, entries); + } - return response; + @Override + public void onFailure(Throwable t) { + AccessValidator.handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR); + } + }); + result.setResult(new ResponseEntity(HttpStatus.OK)); + }); } - private void getAttributeValuesCallback(@Nullable DeferredResult result, SecurityUser user, EntityId entityId, String scope, String keys) { - List keyList = null; - if (!StringUtils.isEmpty(keys)) { - keyList = Arrays.asList(keys.split(",")); + private void getLatestTimeseriesValuesCallback(@Nullable DeferredResult result, SecurityUser user, EntityId entityId, String keys) { + ListenableFuture> future; + if (StringUtils.isEmpty(keys)) { + future = tsService.findAllLatest(entityId); + } else { + future = tsService.findLatest(entityId, toKeysList(keys)); } + Futures.addCallback(future, getTsKvListCallback(result)); + } + + private void getAttributeValuesCallback(@Nullable DeferredResult result, SecurityUser user, EntityId entityId, String scope, String keys) { + List keyList = toKeysList(keys); FutureCallback> callback = getAttributeValuesToResponseCallback(result, user, scope, entityId, keyList); if (!StringUtils.isEmpty(scope)) { if (keyList != null && !keyList.isEmpty()) { @@ -247,7 +434,7 @@ public class TelemetryController extends BaseController { @Override public void onFailure(Throwable e) { log.error("Failed to fetch attributes", e); - handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); + AccessValidator.handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); } }; } @@ -264,12 +451,13 @@ public class TelemetryController extends BaseController { @Override public void onFailure(Throwable e) { log.error("Failed to fetch attributes", e); - handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); + AccessValidator.handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); } }; } - private FutureCallback> getAttributeValuesToResponseCallback(final DeferredResult response, final SecurityUser user, final String scope, + private FutureCallback> getAttributeValuesToResponseCallback(final DeferredResult response, + final SecurityUser user, final String scope, final EntityId entityId, final List keyList) { return new FutureCallback>() { @Override @@ -284,12 +472,32 @@ public class TelemetryController extends BaseController { public void onFailure(Throwable e) { log.error("Failed to fetch attributes", e); logAttributesRead(user, entityId, scope, keyList, e); - handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); + AccessValidator.handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); } }; } - private void logAttributesRead(SecurityUser user, EntityId entityId, String scope, List keys, Throwable e) { + private FutureCallback> getTsKvListCallback(final DeferredResult response) { + return new FutureCallback>() { + @Override + public void onSuccess(List data) { + Map> result = new LinkedHashMap<>(); + for (TsKvEntry entry : data) { + result.computeIfAbsent(entry.getKey(), k -> new ArrayList<>()) + .add(new TsData(entry.getTs(), entry.getValueAsString())); + } + response.setResult(new ResponseEntity<>(result, HttpStatus.OK)); + } + + @Override + public void onFailure(Throwable e) { + log.error("Failed to fetch historical data", e); + AccessValidator.handleError(e, response, HttpStatus.INTERNAL_SERVER_ERROR); + } + }; + } + + private void logAttributesDeleted(SecurityUser user, EntityId entityId, String scope, List keys, Throwable e) { auditLogService.logEntityAction( user.getTenantId(), user.getCustomerId(), @@ -297,163 +505,39 @@ public class TelemetryController extends BaseController { user.getName(), (UUIDBased & EntityId) entityId, null, - ActionType.ATTRIBUTES_READ, + ActionType.ATTRIBUTES_DELETED, toException(e), scope, keys); } - private void handleError(Throwable e, final DeferredResult response, HttpStatus defaultErrorStatus) { - ResponseEntity responseEntity; - if (e != null && e instanceof ToErrorResponseEntity) { - responseEntity = ((ToErrorResponseEntity) e).toErrorResponseEntity(); - } else if (e != null && e instanceof IllegalArgumentException) { - responseEntity = new ResponseEntity<>(e.getMessage(), HttpStatus.BAD_REQUEST); - } else { - responseEntity = new ResponseEntity<>(defaultErrorStatus); - } - response.setResult(responseEntity); - } - - private void validate(SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - switch (entityId.getEntityType()) { - case DEVICE: - validateDevice(currentUser, entityId, callback); - return; - case ASSET: - validateAsset(currentUser, entityId, callback); - return; - case RULE_CHAIN: - validateRuleChain(currentUser, entityId, callback); - return; - case CUSTOMER: - validateCustomer(currentUser, entityId, callback); - return; - case TENANT: - validateTenant(currentUser, entityId, callback); - return; - default: - //TODO: add support of other entities - throw new IllegalStateException("Not Implemented!"); - } - } - - private void validateDevice(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - if (currentUser.isSystemAdmin()) { - callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); - } else { - ListenableFuture deviceFuture = deviceService.findDeviceByIdAsync(new DeviceId(entityId.getId())); - Futures.addCallback(deviceFuture, getCallback(callback, device -> { - if (device == null) { - return ValidationResult.entityNotFound(DEVICE_WITH_REQUESTED_ID_NOT_FOUND); - } else { - if (!device.getTenantId().equals(currentUser.getTenantId())) { - return ValidationResult.accessDenied("Device doesn't belong to the current Tenant!"); - } else if (currentUser.isCustomerUser() && !device.getCustomerId().equals(currentUser.getCustomerId())) { - return ValidationResult.accessDenied("Device doesn't belong to the current Customer!"); - } else { - return ValidationResult.ok(); - } - } - })); - } - } - - private void validateAsset(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - if (currentUser.isSystemAdmin()) { - callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); - } else { - ListenableFuture assetFuture = assetService.findAssetByIdAsync(new AssetId(entityId.getId())); - Futures.addCallback(assetFuture, getCallback(callback, asset -> { - if (asset == null) { - return ValidationResult.entityNotFound("Asset with requested id wasn't found!"); - } else { - if (!asset.getTenantId().equals(currentUser.getTenantId())) { - return ValidationResult.accessDenied("Asset doesn't belong to the current Tenant!"); - } else if (currentUser.isCustomerUser() && !asset.getCustomerId().equals(currentUser.getCustomerId())) { - return ValidationResult.accessDenied("Asset doesn't belong to the current Customer!"); - } else { - return ValidationResult.ok(); - } - } - })); - } - } - - - private void validateRuleChain(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - if (currentUser.isCustomerUser()) { - callback.onSuccess(ValidationResult.accessDenied(CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); - } else { - ListenableFuture ruleChainFuture = ruleChainService.findRuleChainByIdAsync(new RuleChainId(entityId.getId())); - Futures.addCallback(ruleChainFuture, getCallback(callback, ruleChain -> { - if (ruleChain == null) { - return ValidationResult.entityNotFound("Rule chain with requested id wasn't found!"); - } else { - if (currentUser.isTenantAdmin() && !ruleChain.getTenantId().equals(currentUser.getTenantId())) { - return ValidationResult.accessDenied("Rule chain doesn't belong to the current Tenant!"); - } else if (currentUser.isSystemAdmin() && !ruleChain.getTenantId().isNullUid()) { - return ValidationResult.accessDenied("Rule chain is not in system scope!"); - } else { - return ValidationResult.ok(); - } - } - })); - } - } - - private void validateCustomer(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - if (currentUser.isSystemAdmin()) { - callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); - } else { - ListenableFuture customerFuture = customerService.findCustomerByIdAsync(new CustomerId(entityId.getId())); - Futures.addCallback(customerFuture, getCallback(callback, customer -> { - if (customer == null) { - return ValidationResult.entityNotFound("Customer with requested id wasn't found!"); - } else { - if (!customer.getTenantId().equals(currentUser.getTenantId())) { - return ValidationResult.accessDenied("Customer doesn't belong to the current Tenant!"); - } else if (currentUser.isCustomerUser() && !customer.getId().equals(currentUser.getCustomerId())) { - return ValidationResult.accessDenied("Customer doesn't relate to the currently authorized customer user!"); - } else { - return ValidationResult.ok(); - } - } - })); - } - } - - private void validateTenant(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { - if (currentUser.isCustomerUser()) { - callback.onSuccess(ValidationResult.accessDenied(CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); - } else if (currentUser.isSystemAdmin()) { - callback.onSuccess(ValidationResult.ok()); - } else { - ListenableFuture tenantFuture = tenantService.findTenantByIdAsync(new TenantId(entityId.getId())); - Futures.addCallback(tenantFuture, getCallback(callback, tenant -> { - if (tenant == null) { - return ValidationResult.entityNotFound("Tenant with requested id wasn't found!"); - } else if (!tenant.getId().equals(currentUser.getTenantId())) { - return ValidationResult.accessDenied("Tenant doesn't relate to the currently authorized user!"); - } else { - return ValidationResult.ok(); - } - })); - } + private void logAttributesUpdated(SecurityUser user, EntityId entityId, String scope, List attributes, Throwable e) { + auditLogService.logEntityAction( + user.getTenantId(), + user.getCustomerId(), + user.getId(), + user.getName(), + (UUIDBased & EntityId) entityId, + null, + ActionType.ATTRIBUTES_UPDATED, + toException(e), + scope, + attributes); } - private FutureCallback getCallback(ValidationCallback callback, Function transformer) { - return new FutureCallback() { - @Override - public void onSuccess(@Nullable T result) { - callback.onSuccess(transformer.apply(result)); - } - @Override - public void onFailure(Throwable t) { - callback.onFailure(t); - } - }; + private void logAttributesRead(SecurityUser user, EntityId entityId, String scope, List keys, Throwable e) { + auditLogService.logEntityAction( + user.getTenantId(), + user.getCustomerId(), + user.getId(), + user.getName(), + (UUIDBased & EntityId) entityId, + null, + ActionType.ATTRIBUTES_READ, + toException(e), + scope, + keys); } private ListenableFuture> mergeAllAttributesFutures(List>> futures) { @@ -467,4 +551,40 @@ public class TelemetryController extends BaseController { }, executor); } + private List toKeysList(String keys) { + List keyList = null; + if (!StringUtils.isEmpty(keys)) { + keyList = Arrays.asList(keys.split(",")); + } + return keyList; + } + + private DeferredResult getImmediateDeferredResult(String message, HttpStatus status) { + DeferredResult result = new DeferredResult<>(); + result.setResult(new ResponseEntity<>(message, status)); + return result; + } + + private List extractRequestAttributes(JsonNode jsonNode) { + long ts = System.currentTimeMillis(); + List attributes = new ArrayList<>(); + jsonNode.fields().forEachRemaining(entry -> { + String key = entry.getKey(); + JsonNode value = entry.getValue(); + if (entry.getValue().isTextual()) { + attributes.add(new BaseAttributeKvEntry(new StringDataEntry(key, value.textValue()), ts)); + } else if (entry.getValue().isBoolean()) { + attributes.add(new BaseAttributeKvEntry(new BooleanDataEntry(key, value.booleanValue()), ts)); + } else if (entry.getValue().isDouble()) { + attributes.add(new BaseAttributeKvEntry(new DoubleDataEntry(key, value.doubleValue()), ts)); + } else if (entry.getValue().isNumber()) { + if (entry.getValue().isBigInteger()) { + throw new UncheckedApiException(new InvalidParametersException("Big integer values are not supported!")); + } else { + attributes.add(new BaseAttributeKvEntry(new LongDataEntry(key, value.longValue()), ts)); + } + } + }); + return attributes; + } } diff --git a/application/src/main/java/org/thingsboard/server/controller/ValidationCallback.java b/application/src/main/java/org/thingsboard/server/controller/ValidationCallback.java index ead90ea2f2..6b2718f568 100644 --- a/application/src/main/java/org/thingsboard/server/controller/ValidationCallback.java +++ b/application/src/main/java/org/thingsboard/server/controller/ValidationCallback.java @@ -1,12 +1,12 @@ /** * 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 - *

+ * + * 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. diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/PluginApiController.java b/application/src/main/java/org/thingsboard/server/controller/plugin/PluginApiController.java index 8e3cee4d5d..045835e1b6 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/PluginApiController.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/PluginApiController.java @@ -48,59 +48,59 @@ import javax.servlet.http.HttpServletRequest; @Slf4j public class PluginApiController extends BaseController { - @SuppressWarnings("rawtypes") - @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") - @RequestMapping(value = "/{pluginToken}/**") - @ResponseStatus(value = HttpStatus.OK) - public DeferredResult processRequest( - @PathVariable("pluginToken") String pluginToken, - RequestEntity requestEntity, - HttpServletRequest request) - throws ThingsboardException { - log.debug("[{}] Going to process requst uri: {}", pluginToken, requestEntity.getUrl()); - DeferredResult result = new DeferredResult(); - PluginMetaData pluginMd = pluginService.findPluginByApiToken(pluginToken); - if (pluginMd == null) { - result.setErrorResult(new PluginNotFoundException("Plugin with token: " + pluginToken + " not found!")); - } else { - TenantId tenantId = getCurrentUser().getTenantId(); - CustomerId customerId = getCurrentUser().getCustomerId(); - if (validatePluginAccess(pluginMd, tenantId, customerId)) { - if(tenantId != null && ModelConstants.NULL_UUID.equals(tenantId.getId())){ - tenantId = null; - } - UserId userId = getCurrentUser().getId(); - String userName = getCurrentUser().getName(); - PluginApiCallSecurityContext securityCtx = new PluginApiCallSecurityContext(pluginMd.getTenantId(), pluginMd.getId(), - tenantId, customerId, userId, userName); - actorService.process(new BasicPluginRestMsg(securityCtx, new RestRequest(requestEntity, request), result)); - } else { - result.setResult(new ResponseEntity<>(HttpStatus.FORBIDDEN)); - } - - } - return result; - } - - public static boolean validatePluginAccess(PluginMetaData pluginMd, TenantId tenantId, CustomerId customerId) { - boolean systemAdministrator = tenantId == null || ModelConstants.NULL_UUID.equals(tenantId.getId()); - boolean tenantAdministrator = !systemAdministrator && (customerId == null || ModelConstants.NULL_UUID.equals(customerId.getId())); - boolean systemPlugin = ModelConstants.NULL_UUID.equals(pluginMd.getTenantId().getId()); - - boolean validUser = false; - if (systemPlugin) { - if (pluginMd.isPublicAccess() || systemAdministrator) { - // All users can access public system plugins. Only system - // users can access private system plugins - validUser = true; - } - } else { - if ((pluginMd.isPublicAccess() || tenantAdministrator) && tenantId != null && tenantId.equals(pluginMd.getTenantId())) { - // All tenant users can access public tenant plugins. Only tenant - // administrator can access private tenant plugins - validUser = true; - } - } - return validUser; - } +// @SuppressWarnings("rawtypes") +// @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") +// @RequestMapping(value = "/{pluginToken}/**") +// @ResponseStatus(value = HttpStatus.OK) +// public DeferredResult processRequest( +// @PathVariable("pluginToken") String pluginToken, +// RequestEntity requestEntity, +// HttpServletRequest request) +// throws ThingsboardException { +// log.debug("[{}] Going to process requst uri: {}", pluginToken, requestEntity.getUrl()); +// DeferredResult result = new DeferredResult(); +// PluginMetaData pluginMd = pluginService.findPluginByApiToken(pluginToken); +// if (pluginMd == null) { +// result.setErrorResult(new PluginNotFoundException("Plugin with token: " + pluginToken + " not found!")); +// } else { +// TenantId tenantId = getCurrentUser().getTenantId(); +// CustomerId customerId = getCurrentUser().getCustomerId(); +// if (validatePluginAccess(pluginMd, tenantId, customerId)) { +// if(tenantId != null && ModelConstants.NULL_UUID.equals(tenantId.getId())){ +// tenantId = null; +// } +// UserId userId = getCurrentUser().getId(); +// String userName = getCurrentUser().getName(); +// PluginApiCallSecurityContext securityCtx = new PluginApiCallSecurityContext(pluginMd.getTenantId(), pluginMd.getId(), +// tenantId, customerId, userId, userName); +// actorService.process(new BasicPluginRestMsg(securityCtx, new RestRequest(requestEntity, request), result)); +// } else { +// result.setResult(new ResponseEntity<>(HttpStatus.FORBIDDEN)); +// } +// +// } +// return result; +// } +// +// public static boolean validatePluginAccess(PluginMetaData pluginMd, TenantId tenantId, CustomerId customerId) { +// boolean systemAdministrator = tenantId == null || ModelConstants.NULL_UUID.equals(tenantId.getId()); +// boolean tenantAdministrator = !systemAdministrator && (customerId == null || ModelConstants.NULL_UUID.equals(customerId.getId())); +// boolean systemPlugin = ModelConstants.NULL_UUID.equals(pluginMd.getTenantId().getId()); +// +// boolean validUser = false; +// if (systemPlugin) { +// if (pluginMd.isPublicAccess() || systemAdministrator) { +// // All users can access public system plugins. Only system +// // users can access private system plugins +// validUser = true; +// } +// } else { +// if ((pluginMd.isPublicAccess() || tenantAdministrator) && tenantId != null && tenantId.equals(pluginMd.getTenantId())) { +// // All tenant users can access public tenant plugins. Only tenant +// // administrator can access private tenant plugins +// validUser = true; +// } +// } +// return validUser; +// } } diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/PluginWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java similarity index 69% rename from application/src/main/java/org/thingsboard/server/controller/plugin/PluginWebSocketHandler.java rename to application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index 7f835ed5bf..800bc3bbf9 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/PluginWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -1,12 +1,12 @@ /** * 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 - * + *

+ * 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. @@ -15,57 +15,42 @@ */ package org.thingsboard.server.controller.plugin; -import java.io.IOException; -import java.net.URI; -import java.security.InvalidParameterException; -import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.BeanCreationNotAllowedException; -import org.springframework.context.annotation.Lazy; -import org.springframework.web.bind.annotation.RequestMapping; -import org.springframework.web.bind.annotation.RestController; -import org.thingsboard.server.actors.service.ActorService; -import org.thingsboard.server.common.data.id.UserId; -import org.thingsboard.server.config.WebSocketConfiguration; -import org.thingsboard.server.extensions.api.plugins.PluginConstants; -import org.thingsboard.server.service.security.model.SecurityUser; -import org.thingsboard.server.common.data.id.CustomerId; -import org.thingsboard.server.common.data.id.TenantId; -import org.thingsboard.server.common.data.plugin.PluginMetaData; -import org.thingsboard.server.dao.plugin.PluginService; -import org.thingsboard.server.extensions.api.plugins.PluginApiCallSecurityContext; -import org.thingsboard.server.extensions.api.plugins.ws.BasicPluginWebsocketSessionRef; -import org.thingsboard.server.extensions.api.plugins.ws.PluginWebsocketSessionRef; -import org.thingsboard.server.extensions.api.plugins.ws.SessionEvent; -import org.thingsboard.server.extensions.api.plugins.ws.msg.PluginWebsocketMsg; -import org.thingsboard.server.extensions.api.plugins.ws.msg.SessionEventPluginWebSocketMsg; -import org.thingsboard.server.extensions.api.plugins.ws.msg.TextPluginWebSocketMsg; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.web.socket.CloseStatus; import org.springframework.web.socket.TextMessage; import org.springframework.web.socket.WebSocketSession; import org.springframework.web.socket.handler.TextWebSocketHandler; +import org.thingsboard.server.config.WebSocketConfiguration; +import org.thingsboard.server.extensions.api.plugins.ws.PluginWebsocketSessionRef; +import org.thingsboard.server.extensions.api.plugins.ws.SessionEvent; +import org.thingsboard.server.extensions.api.plugins.ws.msg.PluginWebsocketMsg; +import org.thingsboard.server.extensions.api.plugins.ws.msg.TextPluginWebSocketMsg; +import org.thingsboard.server.service.security.model.SecurityUser; +import org.thingsboard.server.service.telemetry.TelemetryWebSocketMsgEndpoint; +import org.thingsboard.server.service.telemetry.TelemetryWebSocketService; +import org.thingsboard.server.service.telemetry.TelemetryWebSocketSessionRef; + +import java.io.IOException; +import java.net.URI; +import java.security.InvalidParameterException; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; @Service @Slf4j -public class PluginWebSocketHandler extends TextWebSocketHandler implements PluginWebSocketMsgEndpoint { +public class TbWebSocketHandler extends TextWebSocketHandler implements PluginWebSocketMsgEndpoint, TelemetryWebSocketMsgEndpoint { private static final ConcurrentMap internalSessionMap = new ConcurrentHashMap<>(); private static final ConcurrentMap externalSessionMap = new ConcurrentHashMap<>(); @Autowired @Lazy - private ActorService actorService; - - @Autowired - @Lazy - private PluginService pluginService; + private TelemetryWebSocketService webSocketService; @Override public void handleTextMessage(WebSocketSession session, TextMessage message) { @@ -73,7 +58,7 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug log.info("[{}] Processing {}", session.getId(), message); SessionMetaData sessionMd = internalSessionMap.get(session.getId()); if (sessionMd != null) { - actorService.process(new TextPluginWebSocketMsg(sessionMd.sessionRef, message.getPayload())); + webSocketService.handleWebSocketMsg(sessionMd.sessionRef, message.getPayload()); } else { log.warn("[{}] Failed to find session", session.getId()); session.close(CloseStatus.SERVER_ERROR.withReason("Session not found!")); @@ -88,11 +73,11 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug super.afterConnectionEstablished(session); try { String internalSessionId = session.getId(); - PluginWebsocketSessionRef sessionRef = toRef(session); + TelemetryWebSocketSessionRef sessionRef = toRef(session); String externalSessionId = sessionRef.getSessionId(); internalSessionMap.put(internalSessionId, new SessionMetaData(session, sessionRef)); externalSessionMap.put(externalSessionId, internalSessionId); - actorService.process(new SessionEventPluginWebSocketMsg(sessionRef, SessionEvent.onEstablished())); + processInWebSocketService(sessionRef, SessionEvent.onEstablished()); log.info("[{}][{}] Session is started", externalSessionId, session.getId()); } catch (InvalidParameterException e) { log.warn("[[{}] Failed to start session", session.getId(), e); @@ -108,7 +93,7 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug super.handleTransportError(session, tError); SessionMetaData sessionMd = internalSessionMap.get(session.getId()); if (sessionMd != null) { - processInActorService(new SessionEventPluginWebSocketMsg(sessionMd.sessionRef, SessionEvent.onError(tError))); + processInWebSocketService(sessionMd.sessionRef, SessionEvent.onError(tError)); } else { log.warn("[{}] Failed to find session", session.getId()); } @@ -121,20 +106,20 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug SessionMetaData sessionMd = internalSessionMap.remove(session.getId()); if (sessionMd != null) { externalSessionMap.remove(sessionMd.sessionRef.getSessionId()); - processInActorService(new SessionEventPluginWebSocketMsg(sessionMd.sessionRef, SessionEvent.onClosed())); + processInWebSocketService(sessionMd.sessionRef, SessionEvent.onClosed()); } log.info("[{}] Session is closed", session.getId()); } - private void processInActorService(SessionEventPluginWebSocketMsg msg) { + private void processInWebSocketService(TelemetryWebSocketSessionRef sessionRef, SessionEvent event) { try { - actorService.process(msg); + webSocketService.handleWebSocketSessionEvent(sessionRef, event); } catch (BeanCreationNotAllowedException e) { - log.warn("[{}] Failed to close session due to possible shutdown state", msg.getSessionRef().getSessionId()); + log.warn("[{}] Failed to close session due to possible shutdown state", sessionRef.getSessionId()); } } - private PluginWebsocketSessionRef toRef(WebSocketSession session) throws IOException { + private TelemetryWebSocketSessionRef toRef(WebSocketSession session) throws IOException { URI sessionUri = session.getUri(); String path = sessionUri.getPath(); path = path.substring(WebSocketConfiguration.WS_PLUGIN_PREFIX.length()); @@ -142,39 +127,61 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug throw new IllegalArgumentException("URL should contain plugin token!"); } String[] pathElements = path.split("/"); - String pluginToken = pathElements[0]; - // TODO: cache - PluginMetaData pluginMd = pluginService.findPluginByApiToken(pluginToken); - if (pluginMd == null) { + String serviceToken = pathElements[0]; + if (!"telemetry".equalsIgnoreCase(serviceToken)) { throw new InvalidParameterException("Can't find plugin with specified token!"); } else { SecurityUser currentUser = (SecurityUser) session.getAttributes().get(WebSocketConfiguration.WS_SECURITY_USER_ATTRIBUTE); - TenantId tenantId = currentUser.getTenantId(); - CustomerId customerId = currentUser.getCustomerId(); - if (PluginApiController.validatePluginAccess(pluginMd, tenantId, customerId)) { - UserId userId = currentUser.getId(); - String userName = currentUser.getName(); - PluginApiCallSecurityContext securityCtx = new PluginApiCallSecurityContext(pluginMd.getTenantId(), pluginMd.getId(), tenantId, - currentUser.getCustomerId(), userId, userName); - return new BasicPluginWebsocketSessionRef(UUID.randomUUID().toString(), securityCtx, session.getUri(), session.getAttributes(), - session.getLocalAddress(), session.getRemoteAddress()); - } else { - throw new SecurityException("Current user is not allowed to use this plugin!"); - } + return new TelemetryWebSocketSessionRef(UUID.randomUUID().toString(), currentUser, session.getLocalAddress(), session.getRemoteAddress()); } } private static class SessionMetaData { private final WebSocketSession session; - private final PluginWebsocketSessionRef sessionRef; + private final TelemetryWebSocketSessionRef sessionRef; - public SessionMetaData(WebSocketSession session, PluginWebsocketSessionRef sessionRef) { + public SessionMetaData(WebSocketSession session, TelemetryWebSocketSessionRef sessionRef) { super(); this.session = session; this.sessionRef = sessionRef; } } + @Override + public void send(TelemetryWebSocketSessionRef sessionRef, String msg) throws IOException { + String externalId = sessionRef.getSessionId(); + log.debug("[{}] Processing {}", externalId, msg); + String internalId = externalSessionMap.get(externalId); + if (internalId != null) { + SessionMetaData sessionMd = internalSessionMap.get(internalId); + if (sessionMd != null) { + sessionMd.session.sendMessage(new TextMessage(msg)); + } else { + log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId); + } + } else { + log.warn("[{}] Failed to find session by external id", externalId); + } + } + + @Override + public void close(TelemetryWebSocketSessionRef sessionRef) throws IOException { + String externalId = sessionRef.getSessionId(); + log.debug("[{}] Processing close request", externalId); + String internalId = externalSessionMap.get(externalId); + if (internalId != null) { + SessionMetaData sessionMd = internalSessionMap.get(internalId); + if (sessionMd != null) { + sessionMd.session.close(CloseStatus.NORMAL); + } else { + log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId); + } + } else { + log.warn("[{}] Failed to find session by external id", externalId); + } + } + + //TODO: remove @Override public void send(PluginWebsocketMsg wsMsg) throws IOException { PluginWebsocketSessionRef sessionRef = wsMsg.getSessionRef(); @@ -196,6 +203,7 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug } } + //TODO: remove @Override public void close(PluginWebsocketSessionRef sessionRef) throws IOException { String externalId = sessionRef.getSessionId(); @@ -212,5 +220,4 @@ public class PluginWebSocketHandler extends TextWebSocketHandler implements Plug log.warn("[{}] Failed to find session by external id", externalId); } } - } diff --git a/application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java b/application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java new file mode 100644 index 0000000000..01bd23869d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/security/AccessValidator.java @@ -0,0 +1,262 @@ +package org.thingsboard.server.service.security; + +import com.google.common.base.Function; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Component; +import org.springframework.web.context.request.async.DeferredResult; +import org.thingsboard.server.actors.plugin.ValidationResult; +import org.thingsboard.server.common.data.Customer; +import org.thingsboard.server.common.data.Device; +import org.thingsboard.server.common.data.Tenant; +import org.thingsboard.server.common.data.asset.Asset; +import org.thingsboard.server.common.data.id.AssetId; +import org.thingsboard.server.common.data.id.CustomerId; +import org.thingsboard.server.common.data.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.id.RuleChainId; +import org.thingsboard.server.common.data.id.TenantId; +import org.thingsboard.server.common.data.rule.RuleChain; +import org.thingsboard.server.controller.ValidationCallback; +import org.thingsboard.server.dao.alarm.AlarmService; +import org.thingsboard.server.dao.asset.AssetService; +import org.thingsboard.server.dao.customer.CustomerService; +import org.thingsboard.server.dao.device.DeviceService; +import org.thingsboard.server.dao.rule.RuleChainService; +import org.thingsboard.server.dao.tenant.TenantService; +import org.thingsboard.server.dao.user.UserService; +import org.thingsboard.server.exception.ThingsboardException; +import org.thingsboard.server.extensions.api.exception.ToErrorResponseEntity; +import org.thingsboard.server.service.security.model.SecurityUser; + +import javax.annotation.Nullable; +import java.util.function.BiConsumer; + +/** + * Created by ashvayka on 27.03.18. + */ +@Component +public class AccessValidator { + + public static final String CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION = "Customer user is not allowed to perform this operation!"; + public static final String SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION = "System administrator is not allowed to perform this operation!"; + public static final String DEVICE_WITH_REQUESTED_ID_NOT_FOUND = "Device with requested id wasn't found!"; + + @Autowired + protected TenantService tenantService; + + @Autowired + protected CustomerService customerService; + + @Autowired + protected UserService userService; + + @Autowired + protected DeviceService deviceService; + + @Autowired + protected AssetService assetService; + + @Autowired + protected AlarmService alarmService; + + @Autowired + protected RuleChainService ruleChainService; + + public DeferredResult validateEntityAndCallback(SecurityUser currentUser, String entityType, String entityIdStr, + BiConsumer, EntityId> onSuccess) throws ThingsboardException { + return validateEntityAndCallback(currentUser, entityType, entityIdStr, onSuccess, (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + } + + public DeferredResult validateEntityAndCallback(SecurityUser currentUser, String entityType, String entityIdStr, + BiConsumer, EntityId> onSuccess, + BiConsumer, Throwable> onFailure) throws ThingsboardException { + return validateEntityAndCallback(currentUser, EntityIdFactory.getByTypeAndId(entityType, entityIdStr), + onSuccess, onFailure); + } + + public DeferredResult validateEntityAndCallback(SecurityUser currentUser, EntityId entityId, + BiConsumer, EntityId> onSuccess) throws ThingsboardException { + return validateEntityAndCallback(currentUser, entityId, onSuccess, (result, t) -> handleError(t, result, HttpStatus.INTERNAL_SERVER_ERROR)); + } + + public DeferredResult validateEntityAndCallback(SecurityUser currentUser, EntityId entityId, + BiConsumer, EntityId> onSuccess, + BiConsumer, Throwable> onFailure) throws ThingsboardException { + + final DeferredResult response = new DeferredResult<>(); + + validate(currentUser, entityId, new ValidationCallback(response, + new FutureCallback>() { + @Override + public void onSuccess(@Nullable DeferredResult result) { + onSuccess.accept(response, entityId); + } + + @Override + public void onFailure(Throwable t) { + onFailure.accept(response, t); + } + })); + + return response; + } + + public void validate(SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + switch (entityId.getEntityType()) { + case DEVICE: + validateDevice(currentUser, entityId, callback); + return; + case ASSET: + validateAsset(currentUser, entityId, callback); + return; + case RULE_CHAIN: + validateRuleChain(currentUser, entityId, callback); + return; + case CUSTOMER: + validateCustomer(currentUser, entityId, callback); + return; + case TENANT: + validateTenant(currentUser, entityId, callback); + return; + default: + //TODO: add support of other entities + throw new IllegalStateException("Not Implemented!"); + } + } + + private void validateDevice(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + if (currentUser.isSystemAdmin()) { + callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); + } else { + ListenableFuture deviceFuture = deviceService.findDeviceByIdAsync(new DeviceId(entityId.getId())); + Futures.addCallback(deviceFuture, getCallback(callback, device -> { + if (device == null) { + return ValidationResult.entityNotFound(DEVICE_WITH_REQUESTED_ID_NOT_FOUND); + } else { + if (!device.getTenantId().equals(currentUser.getTenantId())) { + return ValidationResult.accessDenied("Device doesn't belong to the current Tenant!"); + } else if (currentUser.isCustomerUser() && !device.getCustomerId().equals(currentUser.getCustomerId())) { + return ValidationResult.accessDenied("Device doesn't belong to the current Customer!"); + } else { + return ValidationResult.ok(); + } + } + })); + } + } + + private void validateAsset(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + if (currentUser.isSystemAdmin()) { + callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); + } else { + ListenableFuture assetFuture = assetService.findAssetByIdAsync(new AssetId(entityId.getId())); + Futures.addCallback(assetFuture, getCallback(callback, asset -> { + if (asset == null) { + return ValidationResult.entityNotFound("Asset with requested id wasn't found!"); + } else { + if (!asset.getTenantId().equals(currentUser.getTenantId())) { + return ValidationResult.accessDenied("Asset doesn't belong to the current Tenant!"); + } else if (currentUser.isCustomerUser() && !asset.getCustomerId().equals(currentUser.getCustomerId())) { + return ValidationResult.accessDenied("Asset doesn't belong to the current Customer!"); + } else { + return ValidationResult.ok(); + } + } + })); + } + } + + + private void validateRuleChain(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + if (currentUser.isCustomerUser()) { + callback.onSuccess(ValidationResult.accessDenied(CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); + } else { + ListenableFuture ruleChainFuture = ruleChainService.findRuleChainByIdAsync(new RuleChainId(entityId.getId())); + Futures.addCallback(ruleChainFuture, getCallback(callback, ruleChain -> { + if (ruleChain == null) { + return ValidationResult.entityNotFound("Rule chain with requested id wasn't found!"); + } else { + if (currentUser.isTenantAdmin() && !ruleChain.getTenantId().equals(currentUser.getTenantId())) { + return ValidationResult.accessDenied("Rule chain doesn't belong to the current Tenant!"); + } else if (currentUser.isSystemAdmin() && !ruleChain.getTenantId().isNullUid()) { + return ValidationResult.accessDenied("Rule chain is not in system scope!"); + } else { + return ValidationResult.ok(); + } + } + })); + } + } + + private void validateCustomer(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + if (currentUser.isSystemAdmin()) { + callback.onSuccess(ValidationResult.accessDenied(SYSTEM_ADMINISTRATOR_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); + } else { + ListenableFuture customerFuture = customerService.findCustomerByIdAsync(new CustomerId(entityId.getId())); + Futures.addCallback(customerFuture, getCallback(callback, customer -> { + if (customer == null) { + return ValidationResult.entityNotFound("Customer with requested id wasn't found!"); + } else { + if (!customer.getTenantId().equals(currentUser.getTenantId())) { + return ValidationResult.accessDenied("Customer doesn't belong to the current Tenant!"); + } else if (currentUser.isCustomerUser() && !customer.getId().equals(currentUser.getCustomerId())) { + return ValidationResult.accessDenied("Customer doesn't relate to the currently authorized customer user!"); + } else { + return ValidationResult.ok(); + } + } + })); + } + } + + private void validateTenant(final SecurityUser currentUser, EntityId entityId, ValidationCallback callback) { + if (currentUser.isCustomerUser()) { + callback.onSuccess(ValidationResult.accessDenied(CUSTOMER_USER_IS_NOT_ALLOWED_TO_PERFORM_THIS_OPERATION)); + } else if (currentUser.isSystemAdmin()) { + callback.onSuccess(ValidationResult.ok()); + } else { + ListenableFuture tenantFuture = tenantService.findTenantByIdAsync(new TenantId(entityId.getId())); + Futures.addCallback(tenantFuture, getCallback(callback, tenant -> { + if (tenant == null) { + return ValidationResult.entityNotFound("Tenant with requested id wasn't found!"); + } else if (!tenant.getId().equals(currentUser.getTenantId())) { + return ValidationResult.accessDenied("Tenant doesn't relate to the currently authorized user!"); + } else { + return ValidationResult.ok(); + } + })); + } + } + + private FutureCallback getCallback(ValidationCallback callback, Function transformer) { + return new FutureCallback() { + @Override + public void onSuccess(@Nullable T result) { + callback.onSuccess(transformer.apply(result)); + } + + @Override + public void onFailure(Throwable t) { + callback.onFailure(t); + } + }; + } + + public static void handleError(Throwable e, final DeferredResult response, HttpStatus defaultErrorStatus) { + ResponseEntity responseEntity; + if (e != null && e instanceof ToErrorResponseEntity) { + responseEntity = ((ToErrorResponseEntity) e).toErrorResponseEntity(); + } else if (e != null && e instanceof IllegalArgumentException) { + responseEntity = new ResponseEntity<>(e.getMessage(), HttpStatus.BAD_REQUEST); + } else { + responseEntity = new ResponseEntity<>(defaultErrorStatus); + } + response.setResult(responseEntity); + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java new file mode 100644 index 0000000000..359949eef9 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetrySubscriptionService.java @@ -0,0 +1,42 @@ +package org.thingsboard.server.service.telemetry; + +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.Subscription; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Created by ashvayka on 27.03.18. + */ +@Service +@Slf4j +public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptionService { + + @Autowired + private TelemetryWebSocketService wsService; + + + private final Map> subscriptionsByEntityId = new HashMap<>(); + + private final Map> subscriptionsByWsSessionId = new HashMap<>(); + + + + @Override + public void onAttributesUpdateFromServer(EntityId entityId, String scope, List attributes) { + + } + + @Override + public void onTimeseriesUpdateFromServer(EntityId entityId, List entries) { + + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java new file mode 100644 index 0000000000..6d6c33e65b --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -0,0 +1,261 @@ +package org.thingsboard.server.service.telemetry; + +import com.fasterxml.jackson.core.JsonProcessingException; +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.stereotype.Service; +import org.springframework.util.StringUtils; +import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.id.EntityIdFactory; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.attributes.AttributesService; +import org.thingsboard.server.dao.timeseries.TimeseriesService; +import org.thingsboard.server.extensions.api.exception.UnauthorizedException; +import org.thingsboard.server.extensions.api.plugins.PluginCallback; +import org.thingsboard.server.extensions.api.plugins.PluginContext; +import org.thingsboard.server.extensions.api.plugins.ws.PluginWebsocketSessionRef; +import org.thingsboard.server.extensions.api.plugins.ws.SessionEvent; +import org.thingsboard.server.extensions.core.plugin.telemetry.cmd.AttributesSubscriptionCmd; +import org.thingsboard.server.extensions.core.plugin.telemetry.cmd.SubscriptionCmd; +import org.thingsboard.server.extensions.core.plugin.telemetry.cmd.TelemetryPluginCmd; +import org.thingsboard.server.extensions.core.plugin.telemetry.cmd.TelemetryPluginCmdsWrapper; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionErrorCode; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionState; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionType; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionUpdate; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.stream.Collectors; + +/** + * Created by ashvayka on 27.03.18. + */ +@Service +@Slf4j +public class DefaultTelemetryWebSocketService implements TelemetryWebSocketService { + + private static final int UNKNOWN_SUBSCRIPTION_ID = 0; + private static final String PROCESSING_MSG = "[{}] Processing: {}"; + private static final ObjectMapper jsonMapper = new ObjectMapper(); + private static final String FAILED_TO_FETCH_DATA = "Failed to fetch data!"; + private static final String FAILED_TO_FETCH_ATTRIBUTES = "Failed to fetch attributes!"; + private static final String SESSION_META_DATA_NOT_FOUND = "Session meta-data not found!"; + + private final ConcurrentMap wsSessionsMap = new ConcurrentHashMap<>(); + + @Autowired + private TelemetrySubscriptionService subscriptionManager; + + @Autowired + private TelemetryWebSocketMsgEndpoint msgEndpoint; + + @Autowired + private AttributesService attributesService; + + @Autowired + private TimeseriesService tsService; + + @Override + public void handleWebSocketSessionEvent(TelemetryWebSocketSessionRef sessionRef, SessionEvent event) { + String sessionId = sessionRef.getSessionId(); + log.debug(PROCESSING_MSG, sessionId, event); + switch (event.getEventType()) { + case ESTABLISHED: + wsSessionsMap.put(sessionId, new WsSessionMetaData(sessionRef)); + break; + case ERROR: + log.debug("[{}] Unknown websocket session error: {}. ", sessionId, event.getError().orElse(null)); + break; + case CLOSED: + wsSessionsMap.remove(sessionId); + subscriptionManager.cleanupLocalWsSessionSubscriptions(sessionRef, sessionId); + break; + } + } + + @Override + public void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg) { + if (log.isTraceEnabled()) { + log.trace("[{}] Processing: {}", sessionRef.getSessionId(), msg); + } + + try { + TelemetryPluginCmdsWrapper cmdsWrapper = jsonMapper.readValue(msg, TelemetryPluginCmdsWrapper.class); + if (cmdsWrapper != null) { + if (cmdsWrapper.getAttrSubCmds() != null) { + cmdsWrapper.getAttrSubCmds().forEach(cmd -> handleWsAttributesSubscriptionCmd(sessionRef, cmd)); + } + if (cmdsWrapper.getTsSubCmds() != null) { + cmdsWrapper.getTsSubCmds().forEach(cmd -> handleWsTimeseriesSubscriptionCmd(sessionRef, cmd)); + } + if (cmdsWrapper.getHistoryCmds() != null) { + cmdsWrapper.getHistoryCmds().forEach(cmd -> handleWsHistoryCmd(sessionRef, cmd)); + } + } + } catch (IOException e) { + log.warn("Failed to decode subscription cmd: {}", e.getMessage(), e); + SubscriptionUpdate update = new SubscriptionUpdate(UNKNOWN_SUBSCRIPTION_ID, SubscriptionErrorCode.INTERNAL_ERROR, SESSION_META_DATA_NOT_FOUND); + sendWsMsg(sessionRef, update); + } + } + + private void handleWsAttributesSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, AttributesSubscriptionCmd cmd) { + String sessionId = sessionRef.getSessionId(); + log.debug("[{}] Processing: {}", sessionId, cmd); + + if (validateSessionMetadata(sessionRef, cmd, sessionId)) { + if (cmd.isUnsubscribe()) { + unsubscribe(sessionRef, cmd, sessionId); + } else if (validateSubscriptionCmd(sessionRef, cmd)) { + EntityId entityId = EntityIdFactory.getByTypeAndId(cmd.getEntityType(), cmd.getEntityId()); + log.debug("[{}] fetching latest attributes ({}) values for device: {}", sessionId, cmd.getKeys(), entityId); + Optional> keysOptional = getKeys(cmd); + if (keysOptional.isPresent()) { + List keys = new ArrayList<>(keysOptional.get()); + handleWsAttributesSubscriptionByKeys(sessionRef, cmd, sessionId, entityId, keys); + } else { + handleWsAttributesSubscription(sessionRef, cmd, sessionId, entityId); + } + } + } + } + + private void handleWsAttributesSubscriptionByKeys(TelemetryWebSocketSessionRef sessionRef, + AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId, + List keys) { + FutureCallback> callback = new FutureCallback>() { + @Override + public void onSuccess(List data) { + List attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); + sendWsMsg(sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData)); + + Map subState = new HashMap<>(keys.size()); + keys.forEach(key -> subState.put(key, 0L)); + attributesData.forEach(v -> subState.put(v.getKey(), v.getTs())); + + SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), entityId, SubscriptionType.ATTRIBUTES, false, subState, cmd.getScope()); + subscriptionManager.addLocalWsSubscription(sessionId, entityId, sub); + } + + @Override + public void onFailure(Throwable e) { + log.error(FAILED_TO_FETCH_ATTRIBUTES, e); + SubscriptionUpdate update; + if (UnauthorizedException.class.isInstance(e)) { + update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.UNAUTHORIZED, + SubscriptionErrorCode.UNAUTHORIZED.getDefaultMsg()); + } else { + update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + FAILED_TO_FETCH_ATTRIBUTES); + } + sendWsMsg(sessionRef, update); + } + }; + + if (StringUtils.isEmpty(cmd.getScope())) { + //ValidationCallback? + ctx.loadAttributes(entityId, Arrays.asList(DataConstants.allScopes()), keys, callback); + } else { + ctx.loadAttributes(entityId, cmd.getScope(), keys, callback); + } + } + + private void handleWsAttributesSubscription(PluginContext ctx, PluginWebsocketSessionRef sessionRef, + AttributesSubscriptionCmd cmd, String sessionId, EntityId entityId) { + PluginCallback> callback = new PluginCallback>() { + @Override + public void onSuccess(PluginContext ctx, List data) { + List attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); + sendWsMsg(ctx, sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData)); + + Map subState = new HashMap<>(attributesData.size()); + attributesData.forEach(v -> subState.put(v.getKey(), v.getTs())); + + SubscriptionState sub = new SubscriptionState(sessionId, cmd.getCmdId(), entityId, SubscriptionType.ATTRIBUTES, true, subState, cmd.getScope()); + subscriptionManager.addLocalWsSubscription(ctx, sessionId, entityId, sub); + } + + @Override + public void onFailure(PluginContext ctx, Exception e) { + log.error(FAILED_TO_FETCH_ATTRIBUTES, e); + SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + FAILED_TO_FETCH_ATTRIBUTES); + sendWsMsg(ctx, sessionRef, update); + } + }; + + if (StringUtils.isEmpty(cmd.getScope())) { + ctx.loadAttributes(entityId, Arrays.asList(DataConstants.allScopes()), callback); + } else { + ctx.loadAttributes(entityId, cmd.getScope(), callback); + } + } + + private void unsubscribe(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { + if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { + subscriptionManager.cleanupLocalWsSessionSubscriptions(sessionRef, sessionId); + } else { + subscriptionManager.removeSubscription(sessionId, cmd.getCmdId()); + } + } + + private boolean validateSubscriptionCmd(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd) { + if (cmd.getEntityId() == null || cmd.getEntityId().isEmpty()) { + SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.BAD_REQUEST, + "Device id is empty!"); + sendWsMsg(sessionRef, update); + return false; + } + return true; + } + + private boolean validateSessionMetadata(TelemetryWebSocketSessionRef sessionRef, SubscriptionCmd cmd, String sessionId) { + WsSessionMetaData sessionMD = wsSessionsMap.get(sessionId); + if (sessionMD == null) { + log.warn("[{}] Session meta data not found. ", sessionId); + SubscriptionUpdate update = new SubscriptionUpdate(cmd.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR, + SESSION_META_DATA_NOT_FOUND); + sendWsMsg(sessionRef, update); + return false; + } else { + return true; + } + } + + private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, SubscriptionUpdate update) { + try { + msgEndpoint.send(sessionRef, jsonMapper.writeValueAsString(update)); + } catch (JsonProcessingException e) { + log.warn("[{}] Failed to encode reply: {}", sessionRef.getSessionId(), update, e); + } catch (IOException e) { + log.warn("[{}] Failed to send reply: {}", sessionRef.getSessionId(), update, e); + } + } + + private static Optional> getKeys(TelemetryPluginCmd cmd) { + if (!StringUtils.isEmpty(cmd.getKeys())) { + Set keys = new HashSet<>(); + Collections.addAll(keys, cmd.getKeys().split(",")); + return Optional.of(keys); + } else { + return Optional.empty(); + } + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java new file mode 100644 index 0000000000..9673629e9d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetrySubscriptionService.java @@ -0,0 +1,24 @@ +package org.thingsboard.server.service.telemetry; + +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionState; + +import java.util.List; + +/** + * Created by ashvayka on 27.03.18. + */ +public interface TelemetrySubscriptionService { + + void onAttributesUpdateFromServer(EntityId entityId, String scope, List attributes); + + void onTimeseriesUpdateFromServer(EntityId entityId, List entries); + + void cleanupLocalWsSessionSubscriptions(TelemetryWebSocketSessionRef sessionRef, String sessionId); + + void removeSubscription(String sessionId, int cmdId); + + void addLocalWsSubscription(String sessionId, EntityId entityId, SubscriptionState sub); +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java new file mode 100644 index 0000000000..a7e7cad1c3 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketMsgEndpoint.java @@ -0,0 +1,14 @@ +package org.thingsboard.server.service.telemetry; + +import java.io.IOException; + +/** + * Created by ashvayka on 27.03.18. + */ +public interface TelemetryWebSocketMsgEndpoint { + + void send(TelemetryWebSocketSessionRef sessionRef, String msg) throws IOException; + + void close(TelemetryWebSocketSessionRef sessionRef) throws IOException; + +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java new file mode 100644 index 0000000000..883a174eed --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketService.java @@ -0,0 +1,13 @@ +package org.thingsboard.server.service.telemetry; + +import org.thingsboard.server.extensions.api.plugins.ws.SessionEvent; + +/** + * Created by ashvayka on 27.03.18. + */ +public interface TelemetryWebSocketService { + + void handleWebSocketSessionEvent(TelemetryWebSocketSessionRef sessionRef, SessionEvent sessionEvent); + + void handleWebSocketMsg(TelemetryWebSocketSessionRef sessionRef, String msg); +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketSessionRef.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketSessionRef.java new file mode 100644 index 0000000000..3fd4b19fb3 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketSessionRef.java @@ -0,0 +1,53 @@ +package org.thingsboard.server.service.telemetry; + +import lombok.Getter; +import org.thingsboard.server.service.security.model.SecurityUser; + +import java.net.InetSocketAddress; +import java.util.Objects; + +/** + * Created by ashvayka on 27.03.18. + */ +public class TelemetryWebSocketSessionRef { + + private static final long serialVersionUID = 1L; + + @Getter + private final String sessionId; + @Getter + private final SecurityUser securityCtx; + @Getter + private final InetSocketAddress localAddress; + @Getter + private final InetSocketAddress remoteAddress; + + public TelemetryWebSocketSessionRef(String sessionId, SecurityUser securityCtx, InetSocketAddress localAddress, InetSocketAddress remoteAddress) { + this.sessionId = sessionId; + this.securityCtx = securityCtx; + this.localAddress = localAddress; + this.remoteAddress = remoteAddress; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + TelemetryWebSocketSessionRef that = (TelemetryWebSocketSessionRef) o; + return Objects.equals(sessionId, that.sessionId); + } + + @Override + public int hashCode() { + return Objects.hash(sessionId); + } + + @Override + public String toString() { + return "TelemetryWebSocketSessionRef{" + + "sessionId='" + sessionId + '\'' + + ", localAddress=" + localAddress + + ", remoteAddress=" + remoteAddress + + '}'; + } +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketTextMsg.java b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketTextMsg.java new file mode 100644 index 0000000000..6d57122a49 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/TelemetryWebSocketTextMsg.java @@ -0,0 +1,19 @@ +package org.thingsboard.server.service.telemetry; + +import lombok.Data; +import lombok.Getter; +import org.thingsboard.server.service.security.model.SecurityUser; + +import java.net.InetSocketAddress; +import java.util.Objects; + +/** + * Created by ashvayka on 27.03.18. + */ +@Data +public class TelemetryWebSocketTextMsg { + + private final TelemetryWebSocketSessionRef sessionRef; + private final String payload; + +} diff --git a/application/src/main/java/org/thingsboard/server/service/telemetry/WsSessionMetaData.java b/application/src/main/java/org/thingsboard/server/service/telemetry/WsSessionMetaData.java new file mode 100644 index 0000000000..c0b162b93d --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/WsSessionMetaData.java @@ -0,0 +1,38 @@ +package org.thingsboard.server.service.telemetry; + +import org.thingsboard.server.extensions.api.plugins.ws.PluginWebsocketSessionRef; + +/** + * Created by ashvayka on 27.03.18. + */ +public class WsSessionMetaData { + private TelemetryWebSocketSessionRef sessionRef; + private long lastActivityTime; + + public WsSessionMetaData(TelemetryWebSocketSessionRef sessionRef) { + super(); + this.sessionRef = sessionRef; + this.lastActivityTime = System.currentTimeMillis(); + } + + public TelemetryWebSocketSessionRef getSessionRef() { + return sessionRef; + } + + public void setSessionRef(TelemetryWebSocketSessionRef sessionRef) { + this.sessionRef = sessionRef; + } + + public long getLastActivityTime() { + return lastActivityTime; + } + + public void setLastActivityTime(long lastActivityTime) { + this.lastActivityTime = lastActivityTime; + } + + @Override + public String toString() { + return "WsSessionMetaData [sessionRef=" + sessionRef + ", lastActivityTime=" + lastActivityTime + "]"; + } +} diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java index 76b3e336d8..31c1cdae90 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/id/EntityIdFactory.java @@ -33,6 +33,10 @@ public class EntityIdFactory { return getByTypeAndUuid(EntityType.valueOf(type), uuid); } + public static EntityId getByTypeAndUuid(EntityType type, String uuid) { + return getByTypeAndUuid(type, UUID.fromString(uuid)); + } + public static EntityId getByTypeAndUuid(EntityType type, UUID uuid) { switch (type) { case TENANT: diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/SubscriptionManager.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/SubscriptionManager.java index ec00677af5..9d9c2f322a 100644 --- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/SubscriptionManager.java +++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/SubscriptionManager.java @@ -344,9 +344,7 @@ public class SubscriptionManager { } private void checkSubsciptionsPrevAddress(Set subscriptions) { - Iterator subscriptionIterator = subscriptions.iterator(); - while (subscriptionIterator.hasNext()) { - Subscription s = subscriptionIterator.next(); + for (Subscription s : subscriptions) { if (s.isLocal()) { if (s.getServer() != null) { log.trace("[{}] Local subscription is no longer handled on remote server address [{}]", s.getWsSessionId(), s.getServer()); diff --git a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRpcMsgHandler.java b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRpcMsgHandler.java index c6e7a54da0..158945c1f8 100644 --- a/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRpcMsgHandler.java +++ b/extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRpcMsgHandler.java @@ -243,27 +243,19 @@ public class TelemetryRpcMsgHandler implements RpcMsgHandler { switch (attr.getDataType()) { case BOOLEAN: Optional booleanValue = attr.getBooleanValue(); - if (booleanValue.isPresent()) { - dataBuilder.setBoolValue(booleanValue.get()); - } + booleanValue.ifPresent(dataBuilder::setBoolValue); break; case LONG: Optional longValue = attr.getLongValue(); - if (longValue.isPresent()) { - dataBuilder.setLongValue(longValue.get()); - } + longValue.ifPresent(dataBuilder::setLongValue); break; case DOUBLE: Optional doubleValue = attr.getDoubleValue(); - if (doubleValue.isPresent()) { - dataBuilder.setDoubleValue(doubleValue.get()); - } + doubleValue.ifPresent(dataBuilder::setDoubleValue); break; case STRING: Optional stringValue = attr.getStrValue(); - if (stringValue.isPresent()) { - dataBuilder.setStrValue(stringValue.get()); - } + stringValue.ifPresent(dataBuilder::setStrValue); break; } return dataBuilder;