简柏林 / thingskit · Commits

Commit afa3d5c3757dd001a812a493c2aaaa85d5762956

Authored by Volodymyr Babak
1 parent 17cb573e

Fix entity view timeseries

Inline Side-by-side
Showing 2 changed files with 102 additions and 19 deletions
... ... @@ -85,11 +85,11 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
85 85 testDevice = doPost("/api/device", device, Device.class);
86 86
87 87 telemetry = new TelemetryEntityView(
88   - Arrays.asList("109", "209", "309"),
  88 + Arrays.asList("tsKey1", "tsKey2", "tsKey3"),
89 89 new AttributesEntityView(
90   - Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"),
91   - Arrays.asList("saValue1", "saValue2", "saValue3", "saValue4"),
92   - Arrays.asList("shValue1", "shValue2", "shValue3", "shValue4")));
  90 + Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"),
  91 + Arrays.asList("saKey1", "saKey2", "saKey3", "saKey4"),
  92 + Arrays.asList("shKey1", "shKey2", "shKey3", "shKey4")));
93 93 }
94 94
95 95 @After
... ... @@ -324,10 +324,10 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
324 324 @Test
325 325 public void testTheCopyOfAttrsIntoTSForTheView() throws Exception {
326 326 Set<String> actualAttributesSet =
327   - getAttributesByKeys("{\"caValue1\":\"value1\", \"caValue2\":true, \"caValue3\":42.0, \"caValue4\":73}");
  327 + getAttributesByKeys("{\"caKey1\":\"value1\", \"caKey2\":true, \"caKey3\":42.0, \"caKey4\":73}");
328 328
329 329 Set<String> expectedActualAttributesSet =
330   - new HashSet<>(Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"));
  330 + new HashSet<>(Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"));
331 331 assertTrue(actualAttributesSet.containsAll(expectedActualAttributesSet));
332 332 Thread.sleep(1000);
333 333
... ... @@ -335,18 +335,18 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
335 335 List<Map<String, Object>> values = doGetAsync("/api/plugins/telemetry/ENTITY_VIEW/" + savedView.getId().getId().toString() +
336 336 "/values/attributes?keys=" + String.join(",", actualAttributesSet), List.class);
337 337
338   - assertEquals("value1", getValue(values, "caValue1"));
339   - assertEquals(true, getValue(values, "caValue2"));
340   - assertEquals(42.0, getValue(values, "caValue3"));
341   - assertEquals(73, getValue(values, "caValue4"));
  338 + assertEquals("value1", getValue(values, "caKey1"));
  339 + assertEquals(true, getValue(values, "caKey2"));
  340 + assertEquals(42.0, getValue(values, "caKey3"));
  341 + assertEquals(73, getValue(values, "caKey4"));
342 342 }
343 343
344 344 @Test
345 345 public void testTheCopyOfAttrsOutOfTSForTheView() throws Exception {
346 346 Set<String> actualAttributesSet =
347   - getAttributesByKeys("{\"caValue1\":\"value1\", \"caValue2\":true, \"caValue3\":42.0, \"caValue4\":73}");
  347 + getAttributesByKeys("{\"caKey1\":\"value1\", \"caKey2\":true, \"caKey3\":42.0, \"caKey4\":73}");
348 348
349   - Set<String> expectedActualAttributesSet = new HashSet<>(Arrays.asList("caValue1", "caValue2", "caValue3", "caValue4"));
  349 + Set<String> expectedActualAttributesSet = new HashSet<>(Arrays.asList("caKey1", "caKey2", "caKey3", "caKey4"));
350 350 assertTrue(actualAttributesSet.containsAll(expectedActualAttributesSet));
351 351 Thread.sleep(1000);
352 352
... ... @@ -368,6 +368,69 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
368 368 assertEquals(0, values.size());
369 369 }
370 370
  371 +
  372 + @Test
  373 + public void testGetTelemetryWhenEntityViewTimeRangeInsideTimestampRange() throws Exception {
  374 + uploadTelemetry("{\"tsKey1\":\"value1\", \"tsKey2\":true, \"tsKey3\":40.0}");
  375 + Thread.sleep(1000);
  376 + long startTimeMs = System.currentTimeMillis();
  377 + uploadTelemetry("{\"tsKey1\":\"value2\", \"tsKey2\":false, \"tsKey3\":80.0}");
  378 + Thread.sleep(1000);
  379 + uploadTelemetry("{\"tsKey1\":\"value3\", \"tsKey2\":false, \"tsKey3\":120.0}");
  380 + long endTimeMs = System.currentTimeMillis();
  381 + uploadTelemetry("{\"tsKey1\":\"value4\", \"tsKey2\":true, \"tsKey3\":160.0}");
  382 +
  383 + String deviceId = testDevice.getId().getId().toString();
  384 + Set<String> keys = getTelemetryKeys("DEVICE", deviceId);
  385 + Thread.sleep(1000);
  386 +
  387 + EntityView view = createEntityView("Test entity view", startTimeMs, endTimeMs);
  388 + EntityView savedView = doPost("/api/entityView", view, EntityView.class);
  389 + String entityViewId = savedView.getId().getId().toString();
  390 +
  391 + Map<String, List<Map<String, String>>> expectedValues = getTelemetryValues("DEVICE", deviceId, keys, 0L, (startTimeMs + endTimeMs) / 2);
  392 + Assert.assertEquals(2, expectedValues.get("tsKey1").size());
  393 + Assert.assertEquals(2, expectedValues.get("tsKey2").size());
  394 + Assert.assertEquals(2, expectedValues.get("tsKey3").size());
  395 +
  396 + Map<String, List<Map<String, String>>> actualValues = getTelemetryValues("ENTITY_VIEW", entityViewId, keys, 0L, (startTimeMs + endTimeMs) / 2);
  397 + Assert.assertEquals(1, actualValues.get("tsKey1").size());
  398 + Assert.assertEquals(1, actualValues.get("tsKey2").size());
  399 + Assert.assertEquals(1, actualValues.get("tsKey3").size());
  400 + }
  401 +
  402 + private void uploadTelemetry(String strKvs) throws Exception {
  403 + String viewDeviceId = testDevice.getId().getId().toString();
  404 + DeviceCredentials deviceCredentials =
  405 + doGet("/api/device/" + viewDeviceId + "/credentials", DeviceCredentials.class);
  406 + assertEquals(testDevice.getId(), deviceCredentials.getDeviceId());
  407 +
  408 + String accessToken = deviceCredentials.getCredentialsId();
  409 + assertNotNull(accessToken);
  410 +
  411 + String clientId = MqttAsyncClient.generateClientId();
  412 + MqttAsyncClient client = new MqttAsyncClient("tcp://localhost:1883", clientId);
  413 +
  414 + MqttConnectOptions options = new MqttConnectOptions();
  415 + options.setUserName(accessToken);
  416 + client.connect(options);
  417 + Thread.sleep(3000);
  418 +
  419 + MqttMessage message = new MqttMessage();
  420 + message.setPayload(strKvs.getBytes());
  421 + client.publish("v1/devices/me/telemetry", message);
  422 + Thread.sleep(1000);
  423 + }
  424 +
  425 + private Set<String> getTelemetryKeys(String type, String id) throws Exception {
  426 + return new HashSet<>(doGetAsync("/api/plugins/telemetry/" + type + "/" + id + "/keys/timeseries", List.class));
  427 + }
  428 +
  429 + private Map<String, List<Map<String, String>>> getTelemetryValues(String type, String id, Set<String> keys, Long startTs, Long endTs) throws Exception {
  430 + return doGetAsync("/api/plugins/telemetry/" + type + "/" + id +
  431 + "/values/timeseries?keys=" + String.join(",", keys) + "&startTs=" + startTs + "&endTs=" + endTs, Map.class);
  432 + }
  433 +
371 434 private Set<String> getAttributesByKeys(String stringKV) throws Exception {
372 435 String viewDeviceId = testDevice.getId().getId().toString();
373 436 DeviceCredentials deviceCredentials =
... ... @@ -390,7 +453,7 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
390 453 client.publish("v1/devices/me/attributes", message);
391 454 Thread.sleep(1000);
392 455
393   - return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", List.class));
  456 + return new HashSet<>(doGetAsync("/api/plugins/telemetry/DEVICE/" + viewDeviceId + "/keys/attributes", List.class));
394 457 }
395 458
396 459 private Object getValue(List<Map<String, Object>> values, String stringValue) {
... ... @@ -401,13 +464,20 @@ public abstract class BaseEntityViewControllerTest extends AbstractControllerTes
401 464 }
402 465
403 466 private EntityView getNewSavedEntityView(String name) throws Exception {
  467 + EntityView view = createEntityView(name, 0, 0);
  468 + return doPost("/api/entityView", view, EntityView.class);
  469 + }
  470 +
  471 + private EntityView createEntityView(String name, long startTimeMs, long endTimeMs) {
404 472 EntityView view = new EntityView();
405 473 view.setEntityId(testDevice.getId());
406 474 view.setTenantId(savedTenant.getId());
407 475 view.setName(name);
408 476 view.setType("default");
409 477 view.setKeys(telemetry);
410   - return doPost("/api/entityView", view, EntityView.class);
  478 + view.setStartTimeMs(startTimeMs);
  479 + view.setEndTimeMs(endTimeMs);
  480 + return view;
411 481 }
412 482
413 483 private Customer getNewCustomer(String title) {
... ...
... ... @@ -87,7 +87,11 @@ public class BaseTimeseriesService implements TimeseriesService {
87 87 .map(key -> new BaseReadTsKvQuery(key, entityView.getStartTimeMs(), entityView.getEndTimeMs(), 1, "ASC"))
88 88 .collect(Collectors.toList());
89 89
90   - return timeseriesDao.findAllAsync(entityView.getEntityId(), updateQueriesForEntityView(entityView, queries));
  90 + if (queries.size() > 0) {
  91 + return timeseriesDao.findAllAsync(entityView.getEntityId(), queries);
  92 + } else {
  93 + return Futures.immediateFuture(new ArrayList<>());
  94 + }
91 95 }
92 96 keys.forEach(key -> futures.add(timeseriesDao.findLatest(entityId, key)));
93 97 return Futures.allAsList(futures);
... ... @@ -133,11 +137,20 @@ public class BaseTimeseriesService implements TimeseriesService {
133 137
134 138 private List<ReadTsKvQuery> updateQueriesForEntityView(EntityView entityView, List<ReadTsKvQuery> queries) {
135 139 return queries.stream().map(query -> {
136   - long startTs = entityView.getStartTimeMs() == 0 ? query.getStartTs() : entityView.getStartTimeMs();
137   - long endTs = entityView.getEndTimeMs() == 0 ? query.getEndTs() : entityView.getEndTimeMs();
  140 + long startTs;
  141 + if (entityView.getStartTimeMs() != 0 && entityView.getStartTimeMs() > query.getStartTs()) {
  142 + startTs = entityView.getStartTimeMs();
  143 + } else {
  144 + startTs = query.getStartTs();
  145 + }
138 146
139   - return startTs <= query.getStartTs() && endTs >= query.getEndTs() ? query :
140   - new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation());
  147 + long endTs;
  148 + if (entityView.getEndTimeMs() != 0 && entityView.getEndTimeMs() < query.getEndTs()) {
  149 + endTs = entityView.getEndTimeMs();
  150 + } else {
  151 + endTs = query.getEndTs();
  152 + }
  153 + return new BaseReadTsKvQuery(query.getKey(), startTs, endTs, query.getInterval(), query.getLimit(), query.getAggregation());
141 154 }).collect(Collectors.toList());
142 155 }
143 156
... ...