handleRegistration(TransportProtos.LwM2MRegistrationRequestMsg msg) {
+ TenantId tenantId = new TenantId(UUID.fromString(msg.getTenantId()));
+ String deviceName = msg.getEndpoint();
+ Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(deviceName, id -> new ReentrantLock());
+ deviceCreationLock.lock();
+ try {
+ Device device = deviceService.findDeviceByTenantIdAndName(tenantId, deviceName);
+ if (device == null) {
+ device = new Device();
+ device.setTenantId(tenantId);
+ device.setName(deviceName);
+ device.setType("LwM2M");
+ device = deviceService.saveDevice(device);
+ deviceStateService.onDeviceAdded(device);
+ }
+ TransportProtos.LwM2MRegistrationResponseMsg registrationResponseMsg =
+ TransportProtos.LwM2MRegistrationResponseMsg.newBuilder()
+ .setDeviceInfo(getDeviceInfoProto(device)).build();
+ TransportProtos.LwM2MResponseMsg responseMsg = TransportProtos.LwM2MResponseMsg.newBuilder().setRegistrationMsg(registrationResponseMsg).build();
+ return Futures.immediateFuture(TransportApiResponseMsg.newBuilder().setLwM2MResponseMsg(responseMsg).build());
+ } catch (JsonProcessingException e) {
+ log.warn("[{}][{}] Failed to lookup device by gateway id and name", tenantId, deviceName, e);
+ throw new RuntimeException(e);
+ } finally {
+ deviceCreationLock.unlock();
+ }
+ }
}
diff --git a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java
index d542bdbe61..74d81c0307 100644
--- a/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java
+++ b/application/src/main/java/org/thingsboard/server/service/transport/msg/TransportToDeviceActorMsgWrapper.java
@@ -34,6 +34,8 @@ import java.util.UUID;
@Data
public class TransportToDeviceActorMsgWrapper implements TbActorMsg, DeviceAwareMsg, TenantAwareMsg, Serializable {
+ private static final long serialVersionUID = 7191333353202935941L;
+
private final TenantId tenantId;
private final DeviceId deviceId;
private final TransportToDeviceActorMsg msg;
diff --git a/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java b/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java
new file mode 100644
index 0000000000..4ce0d2f3d5
--- /dev/null
+++ b/application/src/main/java/org/thingsboard/server/utils/EventDeduplicationExecutor.java
@@ -0,0 +1,85 @@
+/**
+ * Copyright © 2016-2021 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.utils;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+
+/**
+ * This class deduplicate executions of the specified function.
+ * Useful in cluster mode, when you get event about partition change multiple times.
+ * Assuming that the function execution is expensive, we should execute it immediately when first time event occurs and
+ * later, once the processing of first event is done, process last pending task.
+ *
+ * @param parameters of the function
+ */
+@Slf4j
+public class EventDeduplicationExecutor
{
+ private final String name;
+ private final ExecutorService executor;
+ private final Consumer
function;
+ private P pendingTask;
+ private boolean busy;
+
+ public EventDeduplicationExecutor(String name, ExecutorService executor, Consumer
function) {
+ this.name = name;
+ this.executor = executor;
+ this.function = function;
+ }
+
+ public void submit(P params) {
+ log.info("[{}] Going to submit: {}", name, params);
+ synchronized (EventDeduplicationExecutor.this) {
+ if (!busy) {
+ busy = true;
+ pendingTask = null;
+ try {
+ log.info("[{}] Submitting task: {}", name, params);
+ executor.submit(() -> {
+ try {
+ log.info("[{}] Executing task: {}", name, params);
+ function.accept(params);
+ } catch (Throwable e) {
+ log.warn("[{}] Failed to process task with parameters: {}", name, params, e);
+ throw e;
+ } finally {
+ unlockAndProcessIfAny();
+ }
+ });
+ } catch (Throwable e) {
+ log.warn("[{}] Failed to submit task with parameters: {}", name, params, e);
+ unlockAndProcessIfAny();
+ throw e;
+ }
+ } else {
+ log.info("[{}] Task is already in progress. {} pending task: {}", name, pendingTask == null ? "adding" : "updating", params);
+ pendingTask = params;
+ }
+ }
+ }
+
+ private void unlockAndProcessIfAny() {
+ synchronized (EventDeduplicationExecutor.this) {
+ busy = false;
+ if (pendingTask != null) {
+ submit(pendingTask);
+ }
+ }
+ }
+}
diff --git a/application/src/main/java/org/thingsboard/server/utils/MiscUtils.java b/application/src/main/java/org/thingsboard/server/utils/MiscUtils.java
index 6ca82aa062..8ee5a5ebc7 100644
--- a/application/src/main/java/org/thingsboard/server/utils/MiscUtils.java
+++ b/application/src/main/java/org/thingsboard/server/utils/MiscUtils.java
@@ -33,6 +33,7 @@ public class MiscUtils {
return "The " + propertyName + " property need to be set!";
}
+ @SuppressWarnings("deprecation")
public static HashFunction forName(String name) {
switch (name) {
case "murmur3_32":
diff --git a/application/src/main/resources/banner.txt b/application/src/main/resources/banner.txt
index 791f878366..111465dd8a 100644
--- a/application/src/main/resources/banner.txt
+++ b/application/src/main/resources/banner.txt
@@ -1,3 +1,10 @@
+ ______ __ _ __ __
+ /_ __/ / /_ (_) ____ ____ _ _____ / /_ ____ ____ _ _____ ____/ /
+ / / / __ \ / / / __ \ / __ `/ / ___/ / __ \ / __ \ / __ `/ / ___/ / __ /
+ / / / / / / / / / / / / / /_/ / (__ ) / /_/ // /_/ // /_/ / / / / /_/ /
+/_/ /_/ /_/ /_/ /_/ /_/ \__, / /____/ /_.___/ \____/ \__,_/ /_/ \__,_/
+ /____/
+
===================================================
:: ${application.title} :: ${application.formatted-version}
===================================================
diff --git a/application/src/main/resources/logback.xml b/application/src/main/resources/logback.xml
index 6d10a74854..f8085ec985 100644
--- a/application/src/main/resources/logback.xml
+++ b/application/src/main/resources/logback.xml
@@ -30,10 +30,13 @@
-
+
+
+
+
@@ -41,4 +44,4 @@
-
\ No newline at end of file
+
diff --git a/application/src/main/resources/thingsboard.yml b/application/src/main/resources/thingsboard.yml
index 437a672c54..3f7ef449ed 100644
--- a/application/src/main/resources/thingsboard.yml
+++ b/application/src/main/resources/thingsboard.yml
@@ -34,6 +34,7 @@ server:
log_controller_error_stack_trace: "${HTTP_LOG_CONTROLLER_ERROR_STACK_TRACE:false}"
ws:
send_timeout: "${TB_SERVER_WS_SEND_TIMEOUT:5000}"
+ ping_timeout: "${TB_SERVER_WS_PING_TIMEOUT:30000}"
limits:
# Limit the amount of sessions and subscriptions available on each server. Put values to zero to disable particular limitation
max_sessions_per_tenant: "${TB_SERVER_WS_TENANT_RATE_LIMITS_MAX_SESSIONS_PER_TENANT:0}"
@@ -118,6 +119,15 @@ security:
githubMapper:
emailUrl: "${SECURITY_OAUTH2_GITHUB_MAPPER_EMAIL_URL_KEY:https://api.github.com/user/emails}"
+# Usage statistics parameters
+usage:
+ stats:
+ report:
+ enabled: "${USAGE_STATS_REPORT_ENABLED:true}"
+ interval: "${USAGE_STATS_REPORT_INTERVAL:10}"
+ check:
+ cycle: "${USAGE_STATS_CHECK_CYCLE:60000}"
+
# Dashboard parameters
dashboard:
# Maximum allowed datapoints fetched by widgets
@@ -313,6 +323,9 @@ actors:
cache:
# caffeine or redis
type: "${CACHE_TYPE:caffeine}"
+ attributes:
+ # make sure that if cache.type is 'redis' and cache.attributes.enabled is 'true' that you change 'maxmemory-policy' Redis config property to 'allkeys-lru', 'allkeys-lfu' or 'allkeys-random'
+ enabled: "${CACHE_ATTRIBUTES_ENABLED:true}"
caffeine:
specs:
@@ -346,6 +359,9 @@ caffeine:
deviceProfiles:
timeToLiveInMinutes: 1440
maxSize: 0
+ attributes:
+ timeToLiveInMinutes: 1440
+ maxSize: 100000
redis:
# standalone or cluster
@@ -489,7 +505,7 @@ js:
# Built-in JVM JavaScript environment properties
local:
# Use Sandboxed (secured) JVM JavaScript environment
- use_js_sandbox: "${USE_LOCAL_JS_SANDBOX:true}"
+ use_js_sandbox: "${USE_LOCAL_JS_SANDBOX:false}"
# Specify thread pool size for JavaScript sandbox resource monitor
monitor_thread_pool_size: "${LOCAL_JS_SANDBOX_MONITOR_THREAD_POOL_SIZE:4}"
# Maximum CPU time in milliseconds allowed for script execution
@@ -530,6 +546,7 @@ transport:
http:
enabled: "${HTTP_ENABLED:true}"
request_timeout: "${HTTP_REQUEST_TIMEOUT:60000}"
+ max_request_timeout: "${HTTP_MAX_REQUEST_TIMEOUT:300000}"
# Local MQTT transport parameters
mqtt:
# Enable/disable mqtt transport protocol.
@@ -566,6 +583,68 @@ transport:
bind_address: "${COAP_BIND_ADDRESS:0.0.0.0}"
bind_port: "${COAP_BIND_PORT:5683}"
timeout: "${COAP_TIMEOUT:10000}"
+ # Local LwM2M transport parameters
+ lwm2m:
+ # Enable/disable lvm2m transport protocol.
+ enabled: "${LWM2M_ENABLED:true}"
+ # We choose a default timeout a bit higher to the MAX_TRANSMIT_WAIT(62-93s) which is the time from starting to
+ # send a Confirmable message to the time when an acknowledgement is no longer expected.
+ # DEFAULT_TIMEOUT = 2 * 60 * 1000l; 2 min in ms
+ timeout: "${LWM2M_TIMEOUT:120000}"
+# model_path_file: "${LWM2M_MODEL_PATH_FILE:./common/transport/lwm2m/src/main/resources/models/}"
+ model_path_file: "${LWM2M_MODEL_PATH_FILE:}"
+ recommended_ciphers: "${LWM2M_RECOMMENDED_CIPHERS:false}"
+ recommended_supported_groups: "${LWM2M_RECOMMENDED_SUPPORTED_GROUPS:true}"
+ request_pool_size: "${LWM2M_REQUEST_POOL_SIZE:100}"
+ request_error_pool_size: "${LWM2M_REQUEST_ERROR_POOL_SIZE:10}"
+ registered_pool_size: "${LWM2M_REGISTERED_POOL_SIZE:10}"
+ update_registered_pool_size: "${LWM2M_UPDATE_REGISTERED_POOL_SIZE:10}"
+ un_registered_pool_size: "${LWM2M_UN_REGISTERED_POOL_SIZE:10}"
+ secure:
+ # Certificate_x509:
+ # To get helps about files format and how to generate it, see: https://github.com/eclipse/leshan/wiki/Credential-files-format
+ # Create new X509 Certificates: common/transport/lwm2m/src/main/resources/credentials/shell/lwM2M_credentials.sh
+ key_store_type: "${LWM2M_KEYSTORE_TYPE:JKS}"
+ # key_store_type: "${LWM2M_KEYSTORE_TYPE:PKCS12}"
+# key_store_path_file: "${KEY_STORE_PATH_FILE:/usr/share/thingsboard/conf/credentials/serverKeyStore.jks}"
+ key_store_path_file: "${KEY_STORE_PATH_FILE:}"
+ key_store_password: "${LWM2M_KEYSTORE_PASSWORD_SERVER:server_ks_password}"
+ root_alias: "${LWM2M_SERVER_ROOT_CA:rootca}"
+ enable_gen_new_key_psk_rpk: "${ENABLE_GEN_NEW_KEY_PSK_RPK:false}"
+ server:
+ id: "${LWM2M_SERVER_ID:123}"
+ bind_address: "${LWM2M_BIND_ADDRESS:0.0.0.0}"
+ bind_port_no_sec: "${LWM2M_BIND_PORT_NO_SEC:5685}"
+ secure:
+ bind_address_security: "${LWM2M_BIND_ADDRESS_SECURITY:0.0.0.0}"
+ bind_port_security: "${LWM2M_BIND_PORT_SECURITY:5686}"
+ # create_rpk: "${CREATE_RPK:}"
+ # Only for RPK: Public & Private Key. If the keystore file is missing or not working
+ # - Public Key (Hex): [3059301306072a8648ce3d020106082a8648ce3d0301070342000405064b9e6762dd8d8b8a52355d7b4d8b9a3d64e6d2ee277d76c248861353f3585eeb1838e4f9e37b31fa347aef5ce3431eb54e0a2506910c5e0298817445721b]
+ # - Private Key (Hex): [308193020100301306072a8648ce3d020106082a8648ce3d030107047930770201010420dc774b309e547ceb48fee547e104ce201a9c48c449dc5414cd04e7f5cf05f67ba00a06082a8648ce3d030107a1440342000405064b9e6762dd8d8b8a52355d7b4d8b9a3d64e6d2ee277d76c248861353f3585eeb1838e4f9e37b31fa347aef5ce3431eb54e0a2506910c5e0298817445721b],
+ # - Elliptic Curve parameters : [secp256r1 [NIST P-256, X9.62 prime256v1] (1.2.840.10045.3.1.7)]
+ public_x: "${LWM2M_SERVER_PUBLIC_X:05064b9e6762dd8d8b8a52355d7b4d8b9a3d64e6d2ee277d76c248861353f358}"
+ public_y: "${LWM2M_SERVER_PUBLIC_Y:5eeb1838e4f9e37b31fa347aef5ce3431eb54e0a2506910c5e0298817445721b}"
+ private_encoded: "${LWM2M_SERVER_PRIVATE_ENCODED:308193020100301306072a8648ce3d020106082a8648ce3d030107047930770201010420dc774b309e547ceb48fee547e104ce201a9c48c449dc5414cd04e7f5cf05f67ba00a06082a8648ce3d030107a1440342000405064b9e6762dd8d8b8a52355d7b4d8b9a3d64e6d2ee277d76c248861353f3585eeb1838e4f9e37b31fa347aef5ce3431eb54e0a2506910c5e0298817445721b}" # Only Certificate_x509:
+ alias: "${LWM2M_KEYSTORE_ALIAS_SERVER:server}"
+ bootstrap:
+ enable: "${LWM2M_BOOTSTRAP_ENABLED:true}"
+ id: "${LWM2M_SERVER_ID:111}"
+ bind_address: "${LWM2M_BIND_ADDRESS_BS:0.0.0.0}"
+ bind_port_no_sec: "${LWM2M_BIND_PORT_NO_SEC_BS:5687}"
+ secure:
+ bind_address_security: "${LWM2M_BIND_ADDRESS_BS:0.0.0.0}"
+ bind_port_security: "${LWM2M_BIND_PORT_SEC_BS:5688}"
+ # Only for RPK: Public & Private Key. If the keystore file is missing or not working
+ # - Elliptic Curve parameters : [secp256r1 [NIST P-256, X9.62 prime256v1] (1.2.840.10045.3.1.7)]
+ # - Public Key (Hex): [3059301306072a8648ce3d020106082a8648ce3d030107034200045017c87a1c1768264656b3b355434b0def6edb8b9bf166a4762d9930cd730f913fc4e61bcd8901ec27c424114c3e887ed372497f0c2cf85839b8443e76988b34]
+ # - Private Key (Hex): [308193020100301306072a8648ce3d020106082a8648ce3d0301070479307702010104205ecafd90caa7be45c42e1f3f32571632b8409e6e6249d7124f4ba56fab3c8083a00a06082a8648ce3d030107a144034200045017c87a1c1768264656b3b355434b0def6edb8b9bf166a4762d9930cd730f913fc4e61bcd8901ec27c424114c3e887ed372497f0c2cf85839b8443e76988b34],
+ public_x: "${LWM2M_SERVER_PUBLIC_X_BS:5017c87a1c1768264656b3b355434b0def6edb8b9bf166a4762d9930cd730f91}"
+ public_y: "${LWM2M_SERVER_PUBLIC_Y_BS:3fc4e61bcd8901ec27c424114c3e887ed372497f0c2cf85839b8443e76988b34}"
+ private_encoded: "${LWM2M_SERVER_PRIVATE_ENCODED_BS:308193020100301306072a8648ce3d020106082a8648ce3d0301070479307702010104205ecafd90caa7be45c42e1f3f32571632b8409e6e6249d7124f4ba56fab3c8083a00a06082a8648ce3d030107a144034200045017c87a1c1768264656b3b355434b0def6edb8b9bf166a4762d9930cd730f913fc4e61bcd8901ec27c424114c3e887ed372497f0c2cf85839b8443e76988b34}" # Only Certificate_x509:
+ alias: "${LWM2M_KEYSTORE_ALIAS_BOOTSTRAP:bootstrap}"
+ # Use redis for Security and Registration stores
+ redis.enabled: "${LWM2M_REDIS_ENABLED:false}"
snmp:
enabled: "${SNMP_ENABLED:true}"
@@ -615,6 +694,10 @@ queue:
transport-api: "${TB_QUEUE_KAFKA_TA_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}"
notifications: "${TB_QUEUE_KAFKA_NOTIFICATIONS_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:1048576000;partitions:1}"
js-executor: "${TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES:retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600;partitions:100}"
+ consumer-stats:
+ enabled: "${TB_QUEUE_KAFKA_CONSUMER_STATS_ENABLED:true}"
+ print-interval-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_MIN_PRINT_INTERVAL_MS:60000}"
+ kafka-response-timeout-ms: "${TB_QUEUE_KAFKA_CONSUMER_STATS_RESPONSE_TIMEOUT_MS:1000}"
aws_sqs:
use_default_credential_provider_chain: "${TB_QUEUE_AWS_SQS_USE_DEFAULT_CREDENTIAL_PROVIDER_CHAIN:false}"
access_key_id: "${TB_QUEUE_AWS_SQS_ACCESS_KEY_ID:YOUR_KEY}"
diff --git a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
index 12f4e5fced..e25592e35e 100644
--- a/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
@@ -375,6 +375,10 @@ public abstract class AbstractWebTest {
return readResponse(doGetAsync(urlTemplate, urlVariables).andExpect(status().isOk()), responseClass);
}
+ protected T doGetAsyncTyped(String urlTemplate, TypeReference responseType, Object... urlVariables) throws Exception {
+ return readResponse(doGetAsync(urlTemplate, urlVariables).andExpect(status().isOk()), responseType);
+ }
+
protected ResultActions doGetAsync(String urlTemplate, Object... urlVariables) throws Exception {
MockHttpServletRequestBuilder getRequest;
getRequest = get(urlTemplate, urlVariables);
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java
index d4e4b4e0d6..5d234d2691 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseEntityQueryControllerTest.java
@@ -44,6 +44,7 @@ import org.thingsboard.server.common.data.query.EntityDataSortOrder;
import org.thingsboard.server.common.data.query.EntityKey;
import org.thingsboard.server.common.data.query.EntityKeyType;
import org.thingsboard.server.common.data.query.EntityListFilter;
+import org.thingsboard.server.common.data.query.EntityTypeFilter;
import org.thingsboard.server.common.data.query.FilterPredicateValue;
import org.thingsboard.server.common.data.query.KeyFilter;
import org.thingsboard.server.common.data.query.NumericFilterPredicate;
@@ -132,6 +133,14 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
count = doPostWithResponse("/api/entitiesQuery/count", countQuery, Long.class);
Assert.assertEquals(97, count.longValue());
+
+ EntityTypeFilter filter2 = new EntityTypeFilter();
+ filter2.setEntityType(EntityType.DEVICE);
+
+ EntityCountQuery countQuery2 = new EntityCountQuery(filter2);
+
+ Long count2 = doPostWithResponse("/api/entitiesQuery/count", countQuery2, Long.class);
+ Assert.assertEquals(97, count2.longValue());
}
@Test
@@ -198,11 +207,31 @@ public abstract class BaseEntityQueryControllerTest extends AbstractControllerTe
Assert.assertEquals(11, data.getTotalElements());
Assert.assertEquals("Device19", data.getData().get(0).getLatest().get(EntityKeyType.ENTITY_FIELD).get("name").getValue());
+
+ EntityTypeFilter filter2 = new EntityTypeFilter();
+ filter2.setEntityType(EntityType.DEVICE);
+
+ EntityDataSortOrder sortOrder2 = new EntityDataSortOrder(
+ new EntityKey(EntityKeyType.ENTITY_FIELD, "createdTime"), EntityDataSortOrder.Direction.ASC
+ );
+ EntityDataPageLink pageLink2 = new EntityDataPageLink(10, 0, null, sortOrder2);
+ List entityFields2 = Collections.singletonList(new EntityKey(EntityKeyType.ENTITY_FIELD, "name"));
+
+ EntityDataQuery query2 = new EntityDataQuery(filter2, pageLink2, entityFields2, null, null);
+
+ PageData data2 =
+ doPostWithTypedResponse("/api/entitiesQuery/find", query2, new TypeReference>() {
+ });
+
+ Assert.assertEquals(97, data2.getTotalElements());
+ Assert.assertEquals(10, data2.getTotalPages());
+ Assert.assertTrue(data2.hasNext());
+ Assert.assertEquals(10, data2.getData().size());
+
}
@Test
public void testFindEntityDataByQueryWithAttributes() throws Exception {
-
List devices = new ArrayList<>();
List temperatures = new ArrayList<>();
List highTemperatures = new ArrayList<>();
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java
index 35ff3703c0..c6d0073d83 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseEntityViewControllerTest.java
@@ -347,8 +347,8 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
Thread.sleep(1000);
- List> values = doGetAsync("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
- "/values/attributes?keys=" + String.join(",", actualAttributesSet), List.class);
+ List> values = doGetAsyncTyped("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
+ "/values/attributes?keys=" + String.join(",", actualAttributesSet), new TypeReference<>() {});
assertEquals("value1", getValue(values, "caKey1"));
assertEquals(true, getValue(values, "caKey2"));
@@ -364,8 +364,8 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
Set expectedActualAttributesSet = new HashSet<>(Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"));
assertTrue(actualAttributesSet.containsAll(expectedActualAttributesSet));
- List> valueTelemetryOfDevices = doGetAsync("/api/plugins/telemetry/DEVICE/" + testDevice.getId().getId().toString() +
- "/values/attributes?keys=" + String.join(",", actualAttributesSet), List.class);
+ List> valueTelemetryOfDevices = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + testDevice.getId().getId().toString() +
+ "/values/attributes?keys=" + String.join(",", actualAttributesSet), new TypeReference<>() {});
EntityView view = new EntityView();
view.setEntityId(testDevice.getId());
@@ -379,8 +379,8 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
Thread.sleep(1000);
- List> values = doGetAsync("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
- "/values/attributes?keys=" + String.join(",", actualAttributesSet), List.class);
+ List> values = doGetAsyncTyped("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
+ "/values/attributes?keys=" + String.join(",", actualAttributesSet), new TypeReference<>() {});
assertEquals(0, values.size());
}
@@ -449,12 +449,12 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
}
private Set getTelemetryKeys(String type, String id) throws Exception {
- return new HashSet<>(doGetAsync("/api/plugins/telemetry/" + type + "/" + id + "/keys/timeseries", List.class));
+ return new HashSet<>(doGetAsyncTyped("/api/plugins/telemetry/" + type + "/" + id + "/keys/timeseries", new TypeReference<>() {}));
}
private Map>> getTelemetryValues(String type, String id, Set keys, Long startTs, Long endTs) throws Exception {
- return doGetAsync("/api/plugins/telemetry/" + type + "/" + id +
- "/values/timeseries?keys=" + String.join(",", keys) + "&startTs=" + startTs + "&endTs=" + endTs, Map.class);
+ return doGetAsyncTyped("/api/plugins/telemetry/" + type + "/" + id +
+ "/values/timeseries?keys=" + String.join(",", keys) + "&startTs=" + startTs + "&endTs=" + endTs, new TypeReference<>() {});
}
private Set getAttributesByKeys(String stringKV) throws Exception {
@@ -479,7 +479,7 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
client.publish("v1/devices/me/attributes", message);
Thread.sleep(1000);
client.disconnect();
- return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", List.class));
+ return new HashSet<>(doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", new TypeReference<>() {}));
}
private Object getValue(List> values, String stringValue) {
diff --git a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
index 299413d9c9..2249940c48 100644
--- a/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
+++ b/application/src/test/java/org/thingsboard/server/controller/BaseWebsocketApiTest.java
@@ -35,16 +35,23 @@ import org.thingsboard.server.common.data.kv.LongDataEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.data.page.PageData;
import org.thingsboard.server.common.data.query.DeviceTypeFilter;
+import org.thingsboard.server.common.data.query.EntityCountQuery;
import org.thingsboard.server.common.data.query.EntityData;
import org.thingsboard.server.common.data.query.EntityDataPageLink;
import org.thingsboard.server.common.data.query.EntityDataQuery;
import org.thingsboard.server.common.data.query.EntityKey;
import org.thingsboard.server.common.data.query.EntityKeyType;
+import org.thingsboard.server.common.data.query.EntityKeyValueType;
+import org.thingsboard.server.common.data.query.FilterPredicateValue;
+import org.thingsboard.server.common.data.query.KeyFilter;
+import org.thingsboard.server.common.data.query.NumericFilterPredicate;
import org.thingsboard.server.common.data.query.TsValue;
import org.thingsboard.server.common.data.security.Authority;
import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService;
import org.thingsboard.server.service.telemetry.cmd.TelemetryPluginCmdsWrapper;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountCmd;
+import org.thingsboard.server.service.telemetry.cmd.v2.EntityCountUpdate;
import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataCmd;
import org.thingsboard.server.service.telemetry.cmd.v2.EntityDataUpdate;
import org.thingsboard.server.service.telemetry.cmd.v2.EntityHistoryCmd;
@@ -243,6 +250,98 @@ public class BaseWebsocketApiTest extends AbstractWebsocketTest {
Assert.assertEquals(new TsValue(dataPoint4.getTs(), dataPoint4.getValueAsString()), tsValues[0]);
}
+ @Test
+ public void testEntityCountWsCmd() throws Exception {
+ Device device = new Device();
+ device.setName("Device");
+ device.setType("default");
+ device.setLabel("testLabel" + (int) (Math.random() * 1000));
+ device = doPost("/api/device", device, Device.class);
+
+ AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(System.currentTimeMillis(), new LongDataEntry("temperature", 42L));
+ sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, Collections.singletonList(dataPoint1));
+
+ DeviceTypeFilter dtf1 = new DeviceTypeFilter();
+ dtf1.setDeviceNameFilter("D");
+ dtf1.setDeviceType("default");
+ EntityCountQuery edq1 = new EntityCountQuery(dtf1, Collections.emptyList());
+
+ EntityCountCmd cmd1 = new EntityCountCmd(1, edq1);
+
+ TelemetryPluginCmdsWrapper wrapper1 = new TelemetryPluginCmdsWrapper();
+ wrapper1.setEntityCountCmds(Collections.singletonList(cmd1));
+
+ wsClient.send(mapper.writeValueAsString(wrapper1));
+ String msg1 = wsClient.waitForReply();
+ EntityCountUpdate update1 = mapper.readValue(msg1, EntityCountUpdate.class);
+ Assert.assertEquals(1, update1.getCmdId());
+ Assert.assertEquals(1, update1.getCount());
+
+ DeviceTypeFilter dtf2 = new DeviceTypeFilter();
+ dtf2.setDeviceNameFilter("D");
+ dtf2.setDeviceType("non-existing-device-type");
+ EntityCountQuery edq2 = new EntityCountQuery(dtf2, Collections.emptyList());
+
+ EntityCountCmd cmd2 = new EntityCountCmd(2, edq2);
+
+ TelemetryPluginCmdsWrapper wrapper2 = new TelemetryPluginCmdsWrapper();
+ wrapper2.setEntityCountCmds(Collections.singletonList(cmd2));
+ wsClient.send(mapper.writeValueAsString(wrapper2));
+
+ String msg2 = wsClient.waitForReply();
+ EntityCountUpdate update2 = mapper.readValue(msg2, EntityCountUpdate.class);
+ Assert.assertEquals(2, update2.getCmdId());
+ Assert.assertEquals(0, update2.getCount());
+
+ KeyFilter highTemperatureFilter = new KeyFilter();
+ highTemperatureFilter.setKey(new EntityKey(EntityKeyType.ATTRIBUTE, "temperature"));
+ NumericFilterPredicate predicate = new NumericFilterPredicate();
+ predicate.setValue(FilterPredicateValue.fromDouble(40));
+ predicate.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
+ highTemperatureFilter.setPredicate(predicate);
+ highTemperatureFilter.setValueType(EntityKeyValueType.NUMERIC);
+
+ DeviceTypeFilter dtf3 = new DeviceTypeFilter();
+ dtf3.setDeviceNameFilter("D");
+ dtf3.setDeviceType("default");
+ EntityCountQuery edq3 = new EntityCountQuery(dtf3, Collections.singletonList(highTemperatureFilter));
+
+ EntityCountCmd cmd3 = new EntityCountCmd(3, edq3);
+
+ TelemetryPluginCmdsWrapper wrapper3 = new TelemetryPluginCmdsWrapper();
+ wrapper3.setEntityCountCmds(Collections.singletonList(cmd3));
+ wsClient.send(mapper.writeValueAsString(wrapper3));
+
+ String msg3 = wsClient.waitForReply();
+ EntityCountUpdate update3 = mapper.readValue(msg3, EntityCountUpdate.class);
+ Assert.assertEquals(3, update3.getCmdId());
+ Assert.assertEquals(1, update3.getCount());
+
+ KeyFilter highTemperatureFilter2 = new KeyFilter();
+ highTemperatureFilter2.setKey(new EntityKey(EntityKeyType.ATTRIBUTE, "temperature"));
+ NumericFilterPredicate predicate2 = new NumericFilterPredicate();
+ predicate2.setValue(FilterPredicateValue.fromDouble(50));
+ predicate2.setOperation(NumericFilterPredicate.NumericOperation.GREATER);
+ highTemperatureFilter2.setPredicate(predicate2);
+ highTemperatureFilter2.setValueType(EntityKeyValueType.NUMERIC);
+
+ DeviceTypeFilter dtf4 = new DeviceTypeFilter();
+ dtf4.setDeviceNameFilter("D");
+ dtf4.setDeviceType("default");
+ EntityCountQuery edq4 = new EntityCountQuery(dtf4, Collections.singletonList(highTemperatureFilter2));
+
+ EntityCountCmd cmd4 = new EntityCountCmd(4, edq4);
+
+ TelemetryPluginCmdsWrapper wrapper4 = new TelemetryPluginCmdsWrapper();
+ wrapper4.setEntityCountCmds(Collections.singletonList(cmd4));
+ wsClient.send(mapper.writeValueAsString(wrapper4));
+
+ String msg4 = wsClient.waitForReply();
+ EntityCountUpdate update4 = mapper.readValue(msg4, EntityCountUpdate.class);
+ Assert.assertEquals(4, update4.getCmdId());
+ Assert.assertEquals(0, update4.getCount());
+ }
+
@Test
public void testEntityDataLatestWidgetFlow() throws Exception {
Device device = new Device();
diff --git a/application/src/test/java/org/thingsboard/server/controller/ControllerSqlTestSuite.java b/application/src/test/java/org/thingsboard/server/controller/ControllerSqlTestSuite.java
index 3186765130..0f969a848f 100644
--- a/application/src/test/java/org/thingsboard/server/controller/ControllerSqlTestSuite.java
+++ b/application/src/test/java/org/thingsboard/server/controller/ControllerSqlTestSuite.java
@@ -27,7 +27,7 @@ import java.util.Arrays;
@RunWith(ClasspathSuite.class)
@ClasspathSuite.ClassnameFilters({
// "org.thingsboard.server.controller.sql.WebsocketApiSqlTest",
-// "org.thingsboard.server.controller.sql.TenantProfileControllerSqlTest",
+// "org.thingsboard.server.controller.sql.EntityQueryControllerSqlTest",
"org.thingsboard.server.controller.sql.*Test",
})
public class ControllerSqlTestSuite {
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/attributes/request/AbstractMqttAttributesRequestIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/attributes/request/AbstractMqttAttributesRequestIntegrationTest.java
index 002474c59d..6ad5db9aff 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/attributes/request/AbstractMqttAttributesRequestIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/attributes/request/AbstractMqttAttributesRequestIntegrationTest.java
@@ -26,7 +26,7 @@ import org.junit.Before;
import org.junit.Test;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
-import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.mqtt.attributes.AbstractMqttAttributesIntegrationTest;
import java.nio.charset.StandardCharsets;
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/attributes/updates/AbstractMqttAttributesUpdatesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/attributes/updates/AbstractMqttAttributesUpdatesIntegrationTest.java
index 7045e28f62..9810c4fe3a 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/attributes/updates/AbstractMqttAttributesUpdatesIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/attributes/updates/AbstractMqttAttributesUpdatesIntegrationTest.java
@@ -25,7 +25,7 @@ import org.junit.Test;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
-import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.mqtt.attributes.AbstractMqttAttributesIntegrationTest;
import java.nio.charset.StandardCharsets;
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java b/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java
index 0fce47fc84..1011c86b7f 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionJsonDeviceTest.java
@@ -37,7 +37,7 @@ import org.thingsboard.server.common.transport.util.JsonUtils;
import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.device.provision.ProvisionResponseStatus;
-import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.mqtt.AbstractMqttIntegrationTest;
import java.util.concurrent.CountDownLatch;
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java b/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java
index 0478a402d9..476b4add4c 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/provision/AbstractMqttProvisionProtoDeviceTest.java
@@ -36,7 +36,7 @@ import org.thingsboard.server.common.msg.EncryptionUtil;
import org.thingsboard.server.dao.device.DeviceCredentialsService;
import org.thingsboard.server.dao.device.DeviceService;
import org.thingsboard.server.dao.device.provision.ProvisionResponseStatus;
-import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.gen.transport.TransportProtos.CredentialsDataProto;
import org.thingsboard.server.gen.transport.TransportProtos.CredentialsType;
import org.thingsboard.server.gen.transport.TransportProtos.ProvisionDeviceCredentialsMsg;
@@ -47,7 +47,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceTokenR
import org.thingsboard.server.gen.transport.TransportProtos.ValidateDeviceX509CertRequestMsg;
import org.thingsboard.server.mqtt.AbstractMqttIntegrationTest;
-import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
index 4ae0072029..19e74b6fdc 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/rpc/AbstractMqttServerSideRpcIntegrationTest.java
@@ -30,7 +30,7 @@ import org.junit.Assert;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.TransportPayloadType;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
-import org.thingsboard.server.dao.util.mapping.JacksonUtil;
+import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.server.mqtt.AbstractMqttIntegrationTest;
import java.util.Arrays;
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java
index d048546323..874d55606d 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/telemetry/attributes/AbstractMqttAttributesIntegrationTest.java
@@ -16,6 +16,7 @@
package org.thingsboard.server.mqtt.telemetry.attributes;
import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.junit.After;
@@ -80,7 +81,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
List actualKeys = null;
while (start <= end) {
- actualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/attributes/CLIENT_SCOPE", List.class);
+ actualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/attributes/CLIENT_SCOPE", new TypeReference<>() {});
if (actualKeys.size() == expectedKeys.size()) {
break;
}
@@ -96,7 +97,7 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
assertEquals(expectedKeySet, actualKeySet);
String getAttributesValuesUrl = getAttributesValuesUrl(deviceId, actualKeySet);
- List> values = doGetAsync(getAttributesValuesUrl, List.class);
+ List> values = doGetAsyncTyped(getAttributesValuesUrl, new TypeReference<>() {});
assertAttributesValues(values, expectedKeySet);
String deleteAttributesUrl = "/api/plugins/telemetry/DEVICE/" + deviceId + "/CLIENT_SCOPE?keys=" + String.join(",", actualKeySet);
doDelete(deleteAttributesUrl);
@@ -121,10 +122,10 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
Thread.sleep(2000);
- List firstDeviceActualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + firstDevice.getId() + "/keys/attributes/CLIENT_SCOPE", List.class);
+ List firstDeviceActualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + firstDevice.getId() + "/keys/attributes/CLIENT_SCOPE", new TypeReference<>() {});
Set firstDeviceActualKeySet = new HashSet<>(firstDeviceActualKeys);
- List secondDeviceActualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + secondDevice.getId() + "/keys/attributes/CLIENT_SCOPE", List.class);
+ List secondDeviceActualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + secondDevice.getId() + "/keys/attributes/CLIENT_SCOPE", new TypeReference<>() {});
Set secondDeviceActualKeySet = new HashSet<>(secondDeviceActualKeys);
Set expectedKeySet = new HashSet<>(expectedKeys);
@@ -135,14 +136,15 @@ public abstract class AbstractMqttAttributesIntegrationTest extends AbstractMqtt
String getAttributesValuesUrlFirstDevice = getAttributesValuesUrl(firstDevice.getId(), firstDeviceActualKeySet);
String getAttributesValuesUrlSecondDevice = getAttributesValuesUrl(firstDevice.getId(), secondDeviceActualKeySet);
- List> firstDeviceValues = doGetAsync(getAttributesValuesUrlFirstDevice, List.class);
- List> secondDeviceValues = doGetAsync(getAttributesValuesUrlSecondDevice, List.class);
+ List> firstDeviceValues = doGetAsyncTyped(getAttributesValuesUrlFirstDevice, new TypeReference<>() {});
+ List> secondDeviceValues = doGetAsyncTyped(getAttributesValuesUrlSecondDevice, new TypeReference<>() {});
assertAttributesValues(firstDeviceValues, expectedKeySet);
assertAttributesValues(secondDeviceValues, expectedKeySet);
}
+ @SuppressWarnings("unchecked")
protected void assertAttributesValues(List> deviceValues, Set expectedKeySet) throws JsonProcessingException {
for (Map map : deviceValues) {
String key = (String) map.get("key");
diff --git a/application/src/test/java/org/thingsboard/server/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java b/application/src/test/java/org/thingsboard/server/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java
index 9984c9faae..9ecc3415c2 100644
--- a/application/src/test/java/org/thingsboard/server/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java
+++ b/application/src/test/java/org/thingsboard/server/mqtt/telemetry/timeseries/AbstractMqttTimeseriesIntegrationTest.java
@@ -15,6 +15,7 @@
*/
package org.thingsboard.server.mqtt.telemetry.timeseries;
+import com.fasterxml.jackson.core.type.TypeReference;
import io.netty.handler.codec.mqtt.MqttQoS;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
@@ -25,6 +26,7 @@ import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.junit.After;
import org.junit.Before;
+import org.junit.Ignore;
import org.junit.Test;
import org.thingsboard.server.common.data.Device;
import org.thingsboard.server.common.data.device.profile.MqttTopics;
@@ -107,7 +109,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
List actualKeys = null;
while (start <= end) {
- actualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/timeseries", List.class);
+ actualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + deviceId + "/keys/timeseries", new TypeReference<>() {});
if (actualKeys.size() == expectedKeys.size()) {
break;
}
@@ -129,13 +131,13 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
}
start = System.currentTimeMillis();
end = System.currentTimeMillis() + 5000;
- Map>> values = null;
+ Map>> values = null;
while (start <= end) {
- values = doGetAsync(getTelemetryValuesUrl, Map.class);
+ values = doGetAsyncTyped(getTelemetryValuesUrl, new TypeReference<>() {});
boolean valid = values.size() == expectedKeys.size();
if (valid) {
for (String key : expectedKeys) {
- List> tsValues = values.get(key);
+ List> tsValues = values.get(key);
if (tsValues != null && tsValues.size() > 0) {
Object ts = tsValues.get(0).get("ts");
if (ts == null) {
@@ -181,10 +183,10 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
Thread.sleep(2000);
- List firstDeviceActualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + firstDevice.getId() + "/keys/timeseries", List.class);
+ List firstDeviceActualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + firstDevice.getId() + "/keys/timeseries", new TypeReference<>() {});
Set firstDeviceActualKeySet = new HashSet<>(firstDeviceActualKeys);
- List secondDeviceActualKeys = doGetAsync("/api/plugins/telemetry/DEVICE/" + secondDevice.getId() + "/keys/timeseries", List.class);
+ List secondDeviceActualKeys = doGetAsyncTyped("/api/plugins/telemetry/DEVICE/" + secondDevice.getId() + "/keys/timeseries", new TypeReference<>() {});
Set secondDeviceActualKeySet = new HashSet<>(secondDeviceActualKeys);
Set expectedKeySet = new HashSet<>(expectedKeys);
@@ -195,8 +197,8 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
String getTelemetryValuesUrlFirstDevice = getTelemetryValuesUrl(firstDevice.getId(), firstDeviceActualKeySet);
String getTelemetryValuesUrlSecondDevice = getTelemetryValuesUrl(firstDevice.getId(), secondDeviceActualKeySet);
- Map>> firstDeviceValues = doGetAsync(getTelemetryValuesUrlFirstDevice, Map.class);
- Map>> secondDeviceValues = doGetAsync(getTelemetryValuesUrlSecondDevice, Map.class);
+ Map>> firstDeviceValues = doGetAsyncTyped(getTelemetryValuesUrlFirstDevice, new TypeReference<>() {});
+ Map>> secondDeviceValues = doGetAsyncTyped(getTelemetryValuesUrlSecondDevice, new TypeReference<>() {});
assertGatewayDeviceData(firstDeviceValues, expectedKeys);
assertGatewayDeviceData(secondDeviceValues, expectedKeys);
@@ -212,7 +214,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
return "/api/plugins/telemetry/DEVICE/" + deviceId + "/values/timeseries?startTs=0&endTs=25000&keys=" + String.join(",", actualKeySet);
}
- private void assertGatewayDeviceData(Map>> deviceValues, List expectedKeys) {
+ private void assertGatewayDeviceData(Map>> deviceValues, List expectedKeys) {
assertEquals(2, deviceValues.get(expectedKeys.get(0)).size());
assertEquals(2, deviceValues.get(expectedKeys.get(1)).size());
@@ -228,11 +230,11 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
}
- private void assertValues(Map>> deviceValues, int arrayIndex) {
- for (Map.Entry>> entry : deviceValues.entrySet()) {
+ private void assertValues(Map>> deviceValues, int arrayIndex) {
+ for (Map.Entry>> entry : deviceValues.entrySet()) {
String key = entry.getKey();
- List> tsKv = entry.getValue();
- String value = tsKv.get(arrayIndex).get("value");
+ List> tsKv = entry.getValue();
+ String value = (String) tsKv.get(arrayIndex).get("value");
switch (key) {
case "key1":
assertEquals("value1", value);
@@ -253,7 +255,7 @@ public abstract class AbstractMqttTimeseriesIntegrationTest extends AbstractMqtt
}
}
- private void assertTs(Map>> deviceValues, List expectedKeys, int ts, int arrayIndex) {
+ private void assertTs(Map>> deviceValues, List expectedKeys, int ts, int arrayIndex) {
assertEquals(ts, deviceValues.get(expectedKeys.get(0)).get(arrayIndex).get("ts"));
assertEquals(ts, deviceValues.get(expectedKeys.get(1)).get(arrayIndex).get("ts"));
assertEquals(ts, deviceValues.get(expectedKeys.get(2)).get(arrayIndex).get("ts"));
diff --git a/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java b/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java
index 7ce23368a9..88d7796af5 100644
--- a/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/cluster/routing/HashPartitionServiceTest.java
@@ -21,12 +21,11 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
-import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.junit.MockitoJUnitRunner;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.test.util.ReflectionTestUtils;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
-import org.thingsboard.server.common.msg.queue.ServiceQueue;
import org.thingsboard.server.queue.discovery.HashPartitionService;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.queue.discovery.TbServiceInfoProvider;
diff --git a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java
index 901fc27bb1..6c4e4a9426 100644
--- a/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java
+++ b/application/src/test/java/org/thingsboard/server/service/queue/TbMsgPackProcessingContextTest.java
@@ -20,7 +20,7 @@ import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
-import org.mockito.runners.MockitoJUnitRunner;
+import org.mockito.junit.MockitoJUnitRunner;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.common.TbProtoQueueMsg;
import org.thingsboard.server.service.queue.processing.TbRuleEngineSubmitStrategy;
diff --git a/application/src/test/java/org/thingsboard/server/util/EventDeduplicationExecutorTest.java b/application/src/test/java/org/thingsboard/server/util/EventDeduplicationExecutorTest.java
new file mode 100644
index 0000000000..ac0b2342ec
--- /dev/null
+++ b/application/src/test/java/org/thingsboard/server/util/EventDeduplicationExecutorTest.java
@@ -0,0 +1,155 @@
+/**
+ * Copyright © 2016-2021 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.util;
+
+import com.google.common.util.concurrent.MoreExecutors;
+import lombok.extern.slf4j.Slf4j;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.mockito.junit.MockitoJUnitRunner;
+import org.thingsboard.server.utils.EventDeduplicationExecutor;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.function.Consumer;
+
+@Slf4j
+@RunWith(MockitoJUnitRunner.class)
+public class EventDeduplicationExecutorTest {
+
+ @Test
+ public void testSimpleFlowSameThread() throws InterruptedException {
+ simpleFlow(MoreExecutors.newDirectExecutorService());
+ }
+
+ @Test
+ public void testPeriodicFlowSameThread() throws InterruptedException {
+ periodicFlow(MoreExecutors.newDirectExecutorService());
+ }
+
+ @Test
+ public void testExceptionFlowSameThread() throws InterruptedException {
+ exceptionFlow(MoreExecutors.newDirectExecutorService());
+ }
+
+ @Test
+ public void testSimpleFlowSingleThread() throws InterruptedException {
+ simpleFlow(Executors.newSingleThreadExecutor());
+ }
+
+ @Test
+ public void testPeriodicFlowSingleThread() throws InterruptedException {
+ periodicFlow(Executors.newSingleThreadExecutor());
+ }
+
+ @Test
+ public void testExceptionFlowSingleThread() throws InterruptedException {
+ exceptionFlow(Executors.newSingleThreadExecutor());
+ }
+
+ @Test
+ public void testSimpleFlowMultiThread() throws InterruptedException {
+ simpleFlow(Executors.newFixedThreadPool(3));
+ }
+
+ @Test
+ public void testPeriodicFlowMultiThread() throws InterruptedException {
+ periodicFlow(Executors.newFixedThreadPool(3));
+ }
+
+ @Test
+ public void testExceptionFlowMultiThread() throws InterruptedException {
+ exceptionFlow(Executors.newFixedThreadPool(3));
+ }
+
+ private void simpleFlow(ExecutorService executorService) throws InterruptedException {
+ try {
+ Consumer function = Mockito.spy(StringConsumer.class);
+ EventDeduplicationExecutor executor = new EventDeduplicationExecutor<>(EventDeduplicationExecutorTest.class.getSimpleName(), executorService, function);
+
+ String params1 = "params1";
+ String params2 = "params2";
+ String params3 = "params3";
+
+ executor.submit(params1);
+ executor.submit(params2);
+ executor.submit(params3);
+ Thread.sleep(500);
+ Mockito.verify(function).accept(params1);
+ Mockito.verify(function).accept(params3);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ private void periodicFlow(ExecutorService executorService) throws InterruptedException {
+ try {
+ Consumer function = Mockito.spy(StringConsumer.class);
+ EventDeduplicationExecutor executor = new EventDeduplicationExecutor<>(EventDeduplicationExecutorTest.class.getSimpleName(), executorService, function);
+
+ String params1 = "params1";
+ String params2 = "params2";
+ String params3 = "params3";
+
+ executor.submit(params1);
+ Thread.sleep(500);
+ executor.submit(params2);
+ Thread.sleep(500);
+ executor.submit(params3);
+ Thread.sleep(500);
+ Mockito.verify(function).accept(params1);
+ Mockito.verify(function).accept(params2);
+ Mockito.verify(function).accept(params3);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ private void exceptionFlow(ExecutorService executorService) throws InterruptedException {
+ try {
+ Consumer function = Mockito.spy(StringConsumer.class);
+ EventDeduplicationExecutor executor = new EventDeduplicationExecutor<>(EventDeduplicationExecutorTest.class.getSimpleName(), executorService, function);
+
+ String params1 = "params1";
+ String params2 = "params2";
+ String params3 = "params3";
+
+ Mockito.doThrow(new RuntimeException()).when(function).accept("params1");
+ executor.submit(params1);
+ executor.submit(params2);
+ Thread.sleep(500);
+ executor.submit(params3);
+ Thread.sleep(500);
+ Mockito.verify(function).accept(params2);
+ Mockito.verify(function).accept(params3);
+ } finally {
+ executorService.shutdownNow();
+ }
+ }
+
+ public static class StringConsumer implements Consumer {
+ @Override
+ public void accept(String s) {
+ try {
+ Thread.sleep(100);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+}
diff --git a/common/actor/pom.xml b/common/actor/pom.xml
index a421b8cef4..1f4849cb79 100644
--- a/common/actor/pom.xml
+++ b/common/actor/pom.xml
@@ -20,7 +20,7 @@
4.0.0
org.thingsboard
- 3.2.1-SNAPSHOT
+ 3.3.0-SNAPSHOT
common
org.thingsboard.common
@@ -67,7 +67,7 @@