Add max string value lenght parameter for attributes/timeseries. Improve WS updates handling.

@@ -24,6 +24,7 @@ import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
 import com.google.gson.JsonParser;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.prepost.PreAuthorize;
 import org.springframework.security.access.prepost.PreAuthorize;
@@ -99,6 +100,9 @@ public class TelemetryController extends BaseController {
     @Autowired
     @Autowired
     private AccessValidator accessValidator;
     private AccessValidator accessValidator;
 
 
+    @Value("${transport.json.max_string_value_length:0}")
+    private int maxStringValueLength;
+
     private ExecutorService executor;
     private ExecutorService executor;
 
 
     @PostConstruct
     @PostConstruct
@@ -628,6 +632,10 @@ public class TelemetryController extends BaseController {
             String key = entry.getKey();
             String key = entry.getKey();
             JsonNode value = entry.getValue();
             JsonNode value = entry.getValue();
             if (entry.getValue().isTextual()) {
             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));
                 attributes.add(new BaseAttributeKvEntry(new StringDataEntry(key, value.textValue()), ts));
             } else if (entry.getValue().isBoolean()) {
             } else if (entry.getValue().isBoolean()) {
                 attributes.add(new BaseAttributeKvEntry(new BooleanDataEntry(key, value.booleanValue()), ts));
                 attributes.add(new BaseAttributeKvEntry(new BooleanDataEntry(key, value.booleanValue()), ts));

@@ -200,7 +200,12 @@ public class TbWebSocketHandler extends TextWebSocketHandler implements Telemetr
                     }
                     }
                 }
                 }
                 synchronized (sessionMd) {
                 synchronized (sessionMd) {
+                    long start = System.currentTimeMillis();
                     sessionMd.session.sendMessage(new TextMessage(msg));
                     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 {
             } else {
                 log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId);
                 log.warn("[{}][{}] Failed to find session by internal id", externalId, internalId);

@@ -581,13 +581,15 @@ public class DefaultTelemetryWebSocketService implements TelemetryWebSocketServi
     }
     }
 
 
     private void sendWsMsg(TelemetryWebSocketSessionRef sessionRef, SubscriptionUpdate update) {
     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<Set<String>> getKeys(TelemetryPluginCmd cmd) {
     private static Optional<Set<String>> getKeys(TelemetryPluginCmd cmd) {

@@ -439,6 +439,8 @@ transport:
   json:
   json:
     # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
     # Cast String data types to Numeric if possible when processing Telemetry/Attributes JSON
     type_cast_enabled: "${JSON_TYPE_CAST_ENABLED:true}"
     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
   # Local HTTP transport parameters
   http:
   http:
     enabled: "${HTTP_ENABLED:true}"
     enabled: "${HTTP_ENABLED:true}"

@@ -61,6 +61,8 @@ public class JsonConverter {
 
 
     private static boolean isTypeCastEnabled = true;
     private static boolean isTypeCastEnabled = true;
 
 
+    private static int maxStringValueLength = 0;
+
     public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonObject) throws JsonSyntaxException {
     public static PostTelemetryMsg convertToTelemetryProto(JsonElement jsonObject) throws JsonSyntaxException {
         long systemTs = System.currentTimeMillis();
         long systemTs = System.currentTimeMillis();
         PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder();
         PostTelemetryMsg.Builder builder = PostTelemetryMsg.newBuilder();
@@ -131,6 +133,10 @@ public class JsonConverter {
             if (element.isJsonPrimitive()) {
             if (element.isJsonPrimitive()) {
                 JsonPrimitive value = element.getAsJsonPrimitive();
                 JsonPrimitive value = element.getAsJsonPrimitive();
                 if (value.isString()) {
                 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())) {
                     if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
                         try {
                         try {
                             result.add(buildNumericKeyValueProto(value, valueEntry.getKey()));
                             result.add(buildNumericKeyValueProto(value, valueEntry.getKey()));
@@ -389,6 +395,10 @@ public class JsonConverter {
             if (element.isJsonPrimitive()) {
             if (element.isJsonPrimitive()) {
                 JsonPrimitive value = element.getAsJsonPrimitive();
                 JsonPrimitive value = element.getAsJsonPrimitive();
                 if (value.isString()) {
                 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())) {
                     if(isTypeCastEnabled && NumberUtils.isParsable(value.getAsString())) {
                         try {
                         try {
                             parseNumericValue(result, valueEntry, value);
                             parseNumericValue(result, valueEntry, value);
@@ -456,4 +466,9 @@ public class JsonConverter {
     public static void setTypeCastEnabled(boolean enabled) {
     public static void setTypeCastEnabled(boolean enabled) {
         isTypeCastEnabled = enabled;
         isTypeCastEnabled = enabled;
     }
     }
+
+    public static void setMaxStringValueLength(int length) {
+        maxStringValueLength = length;
+    }
+
 }
 }

@@ -28,4 +28,10 @@ public class JsonConverterConfig {
         JsonConverter.setTypeCastEnabled(jsonTypeCastEnabled);
         JsonConverter.setTypeCastEnabled(jsonTypeCastEnabled);
         log.info("JSON type cast enabled = {}", 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);
+    }
 }
 }