@ -19,13 +19,16 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode ;
import com.fasterxml.jackson.databind.node.ObjectNode ;
import com.google.common.util.concurrent.FutureCallback ;
import com.google.common.util.concurrent.Futures ;
import lombok.extern.slf4j.Slf4j ;
import org.checkerframework.checker.nullness.qual.Nullable ;
import org.junit.After ;
import org.junit.Assert ;
import org.junit.Before ;
import org.junit.Test ;
import org.mockito.Mockito ;
import org.springframework.beans.factory.annotation.Autowired ;
import org.springframework.boot.test.mock.mockito.SpyBean ;
import org.springframework.test.context.TestPropertySource ;
import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils ;
import org.thingsboard.common.util.JacksonUtil ;
@ -60,7 +63,9 @@ import org.thingsboard.server.common.data.query.NumericFilterPredicate;
import org.thingsboard.server.common.data.query.SingleEntityFilter ;
import org.thingsboard.server.common.data.query.TsValue ;
import org.thingsboard.server.common.data.relation.EntityRelation ;
import org.thingsboard.server.dao.nosql.ResultSetSizeLimitExceededException ;
import org.thingsboard.server.dao.service.DaoSqlTest ;
import org.thingsboard.server.dao.timeseries.TimeseriesService ;
import org.thingsboard.server.service.subscription.SubscriptionErrorCode ;
import org.thingsboard.server.service.subscription.TbAttributeSubscriptionScope ;
import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService ;
@ -71,13 +76,11 @@ import org.thingsboard.server.service.ws.telemetry.cmd.v2.AlarmStatusUpdate;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountCmd ;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityCountUpdate ;
import org.thingsboard.server.service.ws.telemetry.cmd.v2.EntityDataUpdate ;
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate ;
import java.util.ArrayList ;
import java.util.Arrays ;
import java.util.Collections ;
import java.util.List ;
import java.util.Map ;
import java.util.concurrent.CountDownLatch ;
import java.util.concurrent.TimeUnit ;
@ -91,10 +94,14 @@ import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.
"server.ws.alarms_per_alarm_status_subscription_cache_size=5" ,
"server.ws.dynamic_page_link.refresh_interval=15"
} )
public class WebsocketApiTest extends AbstractControllerTest {
public class WebSocketApiTest extends AbstractControllerTest {
@Autowired
private TelemetrySubscriptionService tsService ;
@SpyBean
private TimeseriesService timeseriesService ;
Device device ;
DeviceTypeFilter dtf ;
@ -128,8 +135,8 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getTimeseries ( ) . get ( "temperature" ) . length ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getTimeseries ( ) . get ( "temperature" ) . length ) ;
TsKvEntry dataPoint1 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "temperature" , 42L ) ) ;
TsKvEntry dataPoint2 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 2 ) , new LongDataEntry ( "temperature" , 42L ) ) ;
@ -144,8 +151,8 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > dataList = update . getUpdate ( ) ;
Assert . assertNotNull ( dataList ) ;
Assert . assertEquals ( 1 , dataList . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , dataList . get ( 0 ) . getEntityId ( ) ) ;
TsValue [ ] tsArray = dataList . get ( 0 ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , dataList . getFirst ( ) . getEntityId ( ) ) ;
TsValue [ ] tsArray = dataList . getFirst ( ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertEquals ( 3 , tsArray . length ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getTs ( ) , dataPoint1 . getValueAsString ( ) ) , tsArray [ 0 ] ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getTs ( ) , dataPoint2 . getValueAsString ( ) ) , tsArray [ 1 ] ) ;
@ -162,7 +169,7 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
TsKvEntry dataPoint1 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "temperature" , 42L ) ) ;
TsKvEntry dataPoint2 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 2 ) , new LongDataEntry ( "temperature" , 43L ) ) ;
@ -176,8 +183,8 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > listData = update . getUpdate ( ) ;
Assert . assertNotNull ( listData ) ;
Assert . assertEquals ( 1 , listData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , listData . get ( 0 ) . getEntityId ( ) ) ;
TsValue [ ] tsArray = listData . get ( 0 ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , listData . getFirst ( ) . getEntityId ( ) ) ;
TsValue [ ] tsArray = listData . getFirst ( ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertEquals ( 3 , tsArray . length ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getTs ( ) , dataPoint1 . getValueAsString ( ) ) , tsArray [ 0 ] ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getTs ( ) , dataPoint2 . getValueAsString ( ) ) , tsArray [ 1 ] ) ;
@ -186,7 +193,7 @@ public class WebsocketApiTest extends AbstractControllerTest {
now = System . currentTimeMillis ( ) ;
TsKvEntry dataPoint4 = new BasicTsKvEntry ( now , new LongDataEntry ( "temperature" , 45L ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint4 ) ) ;
sendTelemetry ( device , List . of ( dataPoint4 ) ) ;
String msg = getWsClient ( ) . waitForUpdate ( ) ;
update = JacksonUtil . fromString ( msg , EntityDataUpdate . class ) ;
@ -194,9 +201,9 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getTimeseries ( ) ) ;
TsValue [ ] tsValues = eData . get ( 0 ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getTimeseries ( ) ) ;
TsValue [ ] tsValues = eData . getFirst ( ) . getTimeseries ( ) . get ( "temperature" ) ;
Assert . assertNotNull ( tsValues ) ;
Assert . assertEquals ( new TsValue ( dataPoint4 . getTs ( ) , dataPoint4 . getValueAsString ( ) ) , tsValues [ 0 ] ) ;
}
@ -521,7 +528,7 @@ public class WebsocketApiTest extends AbstractControllerTest {
String msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
// check device
// check a device
AlarmStatusCmd deviceCmd = new AlarmStatusCmd ( 2 , device . getId ( ) , null , List . of ( AlarmSeverity . CRITICAL ) ) ;
getWsClient ( ) . send ( deviceCmd ) ;
@ -589,13 +596,13 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getValue ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getValue ( ) ) ;
TsKvEntry dataPoint1 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "temperature" , 42L ) ) ;
List < TsKvEntry > tsData = Arrays . asList ( dataPoint1 ) ;
List < TsKvEntry > tsData = List . of ( dataPoint1 ) ;
sendTelemetry ( device , tsData ) ;
update = getWsClient ( ) . subscribeLatestUpdate ( keys ) ;
@ -605,36 +612,36 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > listData = update . getUpdate ( ) ;
Assert . assertNotNull ( listData ) ;
Assert . assertEquals ( 1 , listData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , listData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
TsValue tsValue = listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , listData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
TsValue tsValue = listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getTs ( ) , dataPoint1 . getValueAsString ( ) ) , tsValue ) ;
now = System . currentTimeMillis ( ) ;
TsKvEntry dataPoint2 = new BasicTsKvEntry ( now , new LongDataEntry ( "temperature" , 52L ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint2 ) ) ;
sendTelemetry ( device , List . of ( dataPoint2 ) ) ;
update = getWsClient ( ) . parseDataReply ( getWsClient ( ) . waitForUpdate ( ) ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
List < EntityData > eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
tsValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
tsValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getTs ( ) , dataPoint2 . getValueAsString ( ) ) , tsValue ) ;
//Sending update from the past, while latest value has new timestamp;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint1 ) ) ;
sendTelemetry ( device , List . of ( dataPoint1 ) ) ;
String msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
//Sending duplicate update again
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint2 ) ) ;
sendTelemetry ( device , List . of ( dataPoint2 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
}
@ -677,14 +684,14 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getValue ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getValue ( ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
TsKvEntry dataPoint1 = new BasicTsKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "temperature" , 42L ) ) ;
List < TsKvEntry > tsData = Arrays . asList ( dataPoint1 ) ;
List < TsKvEntry > tsData = List . of ( dataPoint1 ) ;
sendTelemetry ( device , tsData ) ;
update = getWsClient ( ) . parseDataReply ( getWsClient ( ) . waitForUpdate ( ) ) ;
@ -694,38 +701,73 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > listData = update . getUpdate ( ) ;
Assert . assertNotNull ( listData ) ;
Assert . assertEquals ( 1 , listData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , listData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
TsValue tsValue = listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , listData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
TsValue tsValue = listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getTs ( ) , dataPoint1 . getValueAsString ( ) ) , tsValue ) ;
now = System . currentTimeMillis ( ) ;
TsKvEntry dataPoint2 = new BasicTsKvEntry ( now , new LongDataEntry ( "temperature" , 52L ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint2 ) ) ;
sendTelemetry ( device , List . of ( dataPoint2 ) ) ;
update = getWsClient ( ) . parseDataReply ( getWsClient ( ) . waitForUpdate ( ) ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
List < EntityData > eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
tsValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
tsValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getTs ( ) , dataPoint2 . getValueAsString ( ) ) , tsValue ) ;
//Sending update from the past, while latest value has new timestamp;
//Sending update from the past, while the latest value has new timestamp;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint1 ) ) ;
sendTelemetry ( device , List . of ( dataPoint1 ) ) ;
String msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
//Sending duplicate update again
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendTelemetry ( device , Arrays . asList ( dataPoint2 ) ) ;
sendTelemetry ( device , List . of ( dataPoint2 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
}
@Test
public void testShouldSendWsUpdateMessageWhenTelemetryWasDeleted ( ) throws Exception {
long now = System . currentTimeMillis ( ) - 100 ;
TsKvEntry dataPoint = new BasicTsKvEntry ( now , new LongDataEntry ( "temperature" , 42L ) ) ;
List < TsKvEntry > tsData = List . of ( dataPoint ) ;
sendTelemetry ( device , tsData ) ;
List < EntityKey > keys = List . of ( new EntityKey ( EntityKeyType . TIME_SERIES , "temperature" ) ) ;
EntityDataUpdate update = getWsClient ( ) . subscribeLatestUpdate ( keys , dtf ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ) ;
Assert . assertEquals ( now , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getTs ( ) ) ;
Assert . assertEquals ( "42" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) . getValue ( ) ) ;
// delete telemetry
getWsClient ( ) . registerWaitForUpdate ( ) ;
doDeleteAsync ( "/api/plugins/telemetry/DEVICE/" + device . getId ( ) + "/timeseries/delete?keys=temperature&deleteAllDataForKeys=true" , String . class ) ;
update = getWsClient ( ) . parseDataReply ( getWsClient ( ) . waitForUpdate ( ) ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
List < EntityData > listData = update . getUpdate ( ) ;
Assert . assertNotNull ( listData ) ;
Assert . assertEquals ( 1 , listData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , listData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) ) ;
TsValue tsValue = listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . TIME_SERIES ) . get ( "temperature" ) ;
Assert . assertEquals ( new TsValue ( 0 , "" ) , tsValue ) ;
}
@Test
public void testEntityDataLatestAttrWsCmd ( ) throws Exception {
long now = System . currentTimeMillis ( ) ;
@ -736,21 +778,21 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getValue ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getValue ( ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
// Pushing update with wrong scope and make sure it will not arrive.
// Pushing update with the wrong scope and make sure it will not arrive.
AttributeKvEntry invalidDataPoint = new BaseAttributeKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "serverAttributeKey" , 55L ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , Arrays . asList ( invalidDataPoint ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , List . of ( invalidDataPoint ) ) ;
Assert . assertNull ( getWsClient ( ) . waitForUpdate ( 3000 ) ) ;
AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "serverAttributeKey" , 42L ) ) ;
List < AttributeKvEntry > tsData = Arrays . asList ( dataPoint1 ) ;
List < AttributeKvEntry > tsData = List . of ( dataPoint1 ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , tsData ) ;
String msg = getWsClient ( ) . waitForUpdate ( ) ;
@ -761,16 +803,16 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > listData = update . getUpdate ( ) ;
Assert . assertNotNull ( listData ) ;
Assert . assertEquals ( 1 , listData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , listData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
TsValue tsValue = listData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , listData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
TsValue tsValue = listData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getLastUpdateTs ( ) , dataPoint1 . getValueAsString ( ) ) , tsValue ) ;
now = System . currentTimeMillis ( ) ;
AttributeKvEntry dataPoint2 = new BaseAttributeKvEntry ( now , new LongDataEntry ( "serverAttributeKey" , 52L ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , Arrays . asList ( dataPoint2 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , List . of ( dataPoint2 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( ) ;
Assert . assertNotNull ( msg ) ;
@ -779,20 +821,20 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
tsValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
tsValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getLastUpdateTs ( ) , dataPoint2 . getValueAsString ( ) ) , tsValue ) ;
//Sending update from the past, while latest value has new timestamp;
//Sending update from the past, while the latest value has new timestamp;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , Arrays . asList ( dataPoint1 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , List . of ( dataPoint1 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
//Sending duplicate update again
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , Arrays . asList ( dataPoint2 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , List . of ( dataPoint2 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
}
@ -812,23 +854,23 @@ public class WebsocketApiTest extends AbstractControllerTest {
PageData < EntityData > pageData = update . getData ( ) ;
Assert . assertNotNull ( pageData ) ;
Assert . assertEquals ( 1 , pageData . getData ( ) . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) . getValue ( ) ) ;
Assert . assertEquals ( device . getId ( ) , pageData . getData ( ) . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) . getValue ( ) ) ;
Assert . assertNotNull ( pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ) ;
Assert . assertEquals ( 0 , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) . getTs ( ) ) ;
Assert . assertEquals ( "" , pageData . getData ( ) . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) . getValue ( ) ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry ( now - TimeUnit . MINUTES . toMillis ( 1 ) , new LongDataEntry ( "serverAttributeKey" , 42L ) ) ;
List < AttributeKvEntry > tsData = Arrays . asList ( dataPoint1 ) ;
List < AttributeKvEntry > tsData = List . of ( dataPoint1 ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , tsData ) ;
@ -839,78 +881,78 @@ public class WebsocketApiTest extends AbstractControllerTest {
List < EntityData > eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
TsValue attrValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) ) ;
TsValue attrValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SERVER_ATTRIBUTE ) . get ( "serverAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint1 . getLastUpdateTs ( ) , dataPoint1 . getValueAsString ( ) ) , attrValue ) ;
//Sending update from the past, while latest value has new timestamp;
//Sending update from the past, while the latest value has new timestamp;
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SHARED_SCOPE , Arrays . asList ( dataPoint1 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SHARED_SCOPE , List . of ( dataPoint1 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
//Sending duplicate update again
getWsClient ( ) . registerWaitForUpdate ( ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , Arrays . asList ( dataPoint1 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , List . of ( dataPoint1 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
Assert . assertNull ( msg ) ;
//Sending update from the past, while latest value has new timestamp;
getWsClient ( ) . registerWaitForUpdate ( ) ;
AttributeKvEntry dataPoint2 = new BaseAttributeKvEntry ( now , new LongDataEntry ( "sharedAttributeKey" , 42L ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SHARED_SCOPE , Arrays . asList ( dataPoint2 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SHARED_SCOPE , List . of ( dataPoint2 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
update = JacksonUtil . fromString ( msg , EntityDataUpdate . class ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) ) ;
attrValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) ) ;
attrValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . SHARED_ATTRIBUTE ) . get ( "sharedAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint2 . getLastUpdateTs ( ) , dataPoint2 . getValueAsString ( ) ) , attrValue ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
AttributeKvEntry dataPoint3 = new BaseAttributeKvEntry ( now , new LongDataEntry ( "clientAttributeKey" , 42L ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , Arrays . asList ( dataPoint3 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , List . of ( dataPoint3 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
update = JacksonUtil . fromString ( msg , EntityDataUpdate . class ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) ) ;
attrValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) ) ;
attrValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . CLIENT_ATTRIBUTE ) . get ( "clientAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint3 . getLastUpdateTs ( ) , dataPoint3 . getValueAsString ( ) ) , attrValue ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
AttributeKvEntry dataPoint4 = new BaseAttributeKvEntry ( now , new LongDataEntry ( "anyAttributeKey" , 42L ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , Arrays . asList ( dataPoint4 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . CLIENT_SCOPE , List . of ( dataPoint4 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
update = JacksonUtil . fromString ( msg , EntityDataUpdate . class ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) ) ;
attrValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) ) ;
attrValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint4 . getLastUpdateTs ( ) , dataPoint4 . getValueAsString ( ) ) , attrValue ) ;
getWsClient ( ) . registerWaitForUpdate ( ) ;
AttributeKvEntry dataPoint5 = new BaseAttributeKvEntry ( now , new LongDataEntry ( "anyAttributeKey" , 43L ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , Arrays . asList ( dataPoint5 ) ) ;
sendAttributes ( device , TbAttributeSubscriptionScope . SERVER_SCOPE , List . of ( dataPoint5 ) ) ;
msg = getWsClient ( ) . waitForUpdate ( TimeUnit . SECONDS . toMillis ( 1 ) ) ;
update = JacksonUtil . fromString ( msg , EntityDataUpdate . class ) ;
Assert . assertEquals ( 1 , update . getCmdId ( ) ) ;
eData = update . getUpdate ( ) ;
Assert . assertNotNull ( eData ) ;
Assert . assertEquals ( 1 , eData . size ( ) ) ;
Assert . assertEquals ( device . getId ( ) , eData . get ( 0 ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) ) ;
attrValue = eData . get ( 0 ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ;
Assert . assertEquals ( device . getId ( ) , eData . getFirst ( ) . getEntityId ( ) ) ;
Assert . assertNotNull ( eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) ) ;
attrValue = eData . getFirst ( ) . getLatest ( ) . get ( EntityKeyType . ATTRIBUTE ) . get ( "anyAttributeKey" ) ;
Assert . assertEquals ( new TsValue ( dataPoint5 . getLastUpdateTs ( ) , dataPoint5 . getValueAsString ( ) ) , attrValue ) ;
}
@ -965,13 +1007,41 @@ public class WebsocketApiTest extends AbstractControllerTest {
}
@Test
public void testHistoryCmdSendsWsErrorOnResultSetSizeLimitExceeded ( ) throws Exception {
ResultSetSizeLimitExceededException exception = new ResultSetSizeLimitExceededException ( 100L , 200L ) ;
Mockito . doReturn ( Futures . immediateFailedFuture ( exception ) )
. when ( timeseriesService ) . findAllByQueries ( Mockito . any ( ) , Mockito . any ( ) , Mockito . any ( ) ) ;
List < String > keys = List . of ( "temperature" ) ;
long now = System . currentTimeMillis ( ) ;
EntityDataUpdate errorUpdate = getWsClient ( ) . sendHistoryCmd ( keys , now , TimeUnit . HOURS . toMillis ( 1 ) , dtf ) ;
assertThat ( errorUpdate . getErrorCode ( ) ) . isEqualTo ( SubscriptionErrorCode . INTERNAL_ERROR . getCode ( ) ) ;
assertThat ( errorUpdate . getErrorMsg ( ) ) . isEqualTo ( exception . getMessage ( ) ) ;
}
@Test
public void testTimeSeriesCmdSendsWsErrorOnResultSetSizeLimitExceeded ( ) throws Exception {
ResultSetSizeLimitExceededException exception = new ResultSetSizeLimitExceededException ( 100L , 200L ) ;
Mockito . doReturn ( Futures . immediateFailedFuture ( exception ) )
. when ( timeseriesService ) . findAllByQueries ( Mockito . any ( ) , Mockito . any ( ) , Mockito . any ( ) ) ;
List < String > keys = List . of ( "temperature" ) ;
long now = System . currentTimeMillis ( ) ;
EntityDataUpdate errorUpdate = getWsClient ( ) . subscribeTsUpdate ( keys , now , TimeUnit . HOURS . toMillis ( 1 ) , dtf ) ;
assertThat ( errorUpdate . getErrorCode ( ) ) . isEqualTo ( SubscriptionErrorCode . INTERNAL_ERROR . getCode ( ) ) ;
assertThat ( errorUpdate . getErrorMsg ( ) ) . isEqualTo ( exception . getMessage ( ) ) ;
}
private void sendTelemetry ( Device device , List < TsKvEntry > tsData ) throws InterruptedException {
CountDownLatch latch = new CountDownLatch ( 1 ) ;
tsService . saveTimeseries ( TimeseriesSaveRequest . builder ( )
. tenantId ( device . getTenantId ( ) )
. entityId ( device . getId ( ) )
. entries ( tsData )
. callback ( new FutureCallback < Void > ( ) {
. callback ( new FutureCallback < > ( ) {
@Override
public void onSuccess ( @Nullable Void result ) {
log . debug ( "sendTelemetry callback onSuccess" ) ;