From 16686afeae1865317de82d2b9862c61cf337addb Mon Sep 17 00:00:00 2001 From: Igor Kulikov Date: Fri, 16 Nov 2018 19:07:39 +0200 Subject: [PATCH] Add max string value lenght parameter for attributes/timeseries. Improve WS updates handling. --- .../server/controller/TelemetryController.java | 8 ++++++++ .../controller/plugin/TbWebSocketHandler.java | 5 +++++ .../DefaultTelemetryWebSocketService.java | 16 +++++++++------- application/src/main/resources/thingsboard.yml | 2 ++ .../common/transport/adaptor/JsonConverter.java | 15 +++++++++++++++ .../transport/adaptor/JsonConverterConfig.java | 6 ++++++ 6 files changed, 45 insertions(+), 7 deletions(-) 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 630352495e..bebd2cc4d3 100644 --- a/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/TelemetryController.java @@ -24,6 +24,7 @@ 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.beans.factory.annotation.Value; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; @@ -99,6 +100,9 @@ public class TelemetryController extends BaseController { @Autowired private AccessValidator accessValidator; + @Value("${transport.json.max_string_value_length:0}") + private int maxStringValueLength; + private ExecutorService executor; @PostConstruct @@ -628,6 +632,10 @@ public class TelemetryController extends BaseController { String key = entry.getKey(); JsonNode value = entry.getValue(); if (entry.getValue().isTextual()) { + if (maxStringValueLength > 0 && entry.getValue().textValue().length() > maxStringValueLength) { + String message = String.format("String value length [%d] for key [%s] is greater than maximum allowed [%d]", entry.getValue().textValue().length(), key, maxStringValueLength); + throw new UncheckedApiException(new InvalidParametersException(message)); + } 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)); diff --git a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java index ebd068dae5..f579db8588 100644 --- a/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java +++ b/application/src/main/java/org/thingsboard/server/controller/plugin/TbWebSocketHandler.java @@ -200,7 +200,12 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr } } synchronized (sessionMd) { + long start = System.currentTimeMillis(); sessionMd.session.sendMessage(new TextMessage(msg)); + long took = System.currentTimeMillis() - start; + if (took >= 1000) { + log.info("[{}][{}] Sending message took more than 1 second [{}ms] {}", sessionRef.getSecurityCtx().getTenantId(), externalId, took, msg); + } } } else { log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId); 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 index 683a8c504f..6b87e033fc 100644 --- a/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java +++ b/application/src/main/java/org/thingsboard/server/service/telemetry/DefaultTelemetryWebSocketService.java @@ -581,13 +581,15 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi } private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, SubscriptionUpdate update) { - try { - msgEndpoint.send(sessionRef, update.getSubscriptionId(), 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); - } + executor.submit(() -> { + try { + msgEndpoint.send(sessionRef, update.getSubscriptionId(), 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) { diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml index c18c2f164f..9f60fb6a90 100644 --- a/application/src/main/resources/thingsboard.yml +++ b/application/src/main/resources/thingsboard.yml @@ -439,6 +439,8 @@ transport: json: # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}" + # Maximum allowed string value length when processing Telemetry/Attributes JSON (0 value disables string value length check) + max_string_value_length: "${JSON_MAX_STRING_VALUE_LENGTH:0}" # Local HTTP transport parameters http: enabled: "${HTTP_ENABLED:true}" diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java index d52a896f10..59d719392a 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverter.java @@ -61,6 +61,8 @@ public class JsonConverter { private static boolean isTypeCastEnabled = true; + private static int maxStringValueLength = 0; + public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonObject) throws JsonSyntaxException { long systemTs = System.currentTimeMillis(); PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder(); @@ -131,6 +133,10 @@ public class JsonConverter { if (element.isJsonPrimitive()) { JsonPrimitive value = element.getAsJsonPrimitive(); if (value.isString()) { + if (maxStringValueLength > 0 && value.getAsString().length() > maxStringValueLength) { + String message = String.format("String value length [%d] for key [%s] is greater than maximum allowed [%d]", value.getAsString().length(), valueEntry.getKey(), maxStringValueLength); + throw new JsonSyntaxException(message); + } if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) { try { result.add(buildNumericKeyValueProto(value, valueEntry.getKey())); @@ -389,6 +395,10 @@ public class JsonConverter { if (element.isJsonPrimitive()) { JsonPrimitive value = element.getAsJsonPrimitive(); if (value.isString()) { + if (maxStringValueLength > 0 && value.getAsString().length() > maxStringValueLength) { + String message = String.format("String value length [%d] for key [%s] is greater than maximum allowed [%d]", value.getAsString().length(), valueEntry.getKey(), maxStringValueLength); + throw new JsonSyntaxException(message); + } if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) { try { parseNumericValue(result, valueEntry, value); @@ -456,4 +466,9 @@ public class JsonConverter { public static void setTypeCastEnabled(boolean enabled) { isTypeCastEnabled = enabled; } + + public static void setMaxStringValueLength(int length) { + maxStringValueLength = length; + } + } diff --git a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverterConfig.java b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverterConfig.java index ac3f4dc580..857c595787 100644 --- a/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverterConfig.java +++ b/common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/adaptor/JsonConverterConfig.java @@ -28,4 +28,10 @@ public class JsonConverterConfig { JsonConverter.setTypeCastEnabled(jsonTypeCastEnabled); log.info("JSON type cast enabled = {}", jsonTypeCastEnabled); } + + @Value("${transport.json.max_string_value_length:0}") + public void setMaxStringValueLength(int maxStringValueLength) { + JsonConverter.setMaxStringValueLength(maxStringValueLength); + log.info("JSON max string value length = {}", maxStringValueLength); + } }