dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix

Browse Source
This commit is contained in:
Andrew Shvayka 2017-03-14 12:49:26 +02:00
parent 696cb33156
commit 5231a33788
1 changed files with 36 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
extensions-core/src/main/java/org/thingsboard/server/extensions/core/plugin/telemetry/handlers/TelemetryRestMsgHandler.java
View File

@ -1,12 +1,12 @@
/**
* Copyright © 2016-2017 The Thingsboard Authors
*
* <p>
* 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
*
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.
@ -87,30 +87,26 @@ public class TelemetryRestMsgHandler extends DefaultRestMsgHandler {
} else if (method.equals("values")) {
if ("timeseries".equals(entity)) {
String keysStr = request.getParameter("keys");
List<String> keys = Arrays.asList(keysStr.split(","));
Optional<Long> startTs = request.getLongParamValue("startTs");
Optional<Long> endTs = request.getLongParamValue("endTs");
Optional<Long> interval = request.getLongParamValue("interval");
Optional<Integer> limit = request.getIntParamValue("limit");
Aggregation agg = Aggregation.valueOf(request.getParameter("agg", Aggregation.NONE.name()));
List<String> keys = Arrays.asList(keysStr.split(","));
List<TsKvQuery> queries = keys.stream().map(key -> new BaseTsKvQuery(key, startTs.get(), endTs.get(), interval.get(), limit.orElse(TelemetryWebsocketMsgHandler.DEFAULT_LIMIT), agg)).collect(Collectors.toList());
ctx.loadTimeseries(deviceId, queries, new PluginCallback<List<TsKvEntry>>() {
@Override
public void onSuccess(PluginContext ctx, List<TsKvEntry> data) {
Map<String, List<TsData>> result = new LinkedHashMap<>();
for (TsKvEntry entry : data) {
result.put(entry.getKey(), data.stream().map(v -> new TsData(v.getTs(), v.getValueAsString())).collect(Collectors.toList()));
}
msg.getResponseHolder().setResult(new ResponseEntity<>(data, HttpStatus.OK));
if (startTs.isPresent() || endTs.isPresent() || interval.isPresent() || limit.isPresent()) {
if (!startTs.isPresent() || !endTs.isPresent() || !interval.isPresent()) {
msg.getResponseHolder().setResult(new ResponseEntity<>(HttpStatus.BAD_REQUEST));
return;
}
Aggregation agg = Aggregation.valueOf(request.getParameter("agg", Aggregation.NONE.name()));
@Override
public void onFailure(PluginContext ctx, Exception e) {
log.error("Failed to fetch historical data", e);
msg.getResponseHolder().setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
}
});
List<TsKvQuery> queries = keys.stream().map(key -> new BaseTsKvQuery(key, startTs.get(), endTs.get(), interval.get(), limit.orElse(TelemetryWebsocketMsgHandler.DEFAULT_LIMIT), agg))
.collect(Collectors.toList());
ctx.loadTimeseries(deviceId, queries, getTsKvListCallback(msg));
} else {
ctx.loadLatestTimeseries(deviceId, keys, getTsKvListCallback(msg));
}
} else if ("attributes".equals(entity)) {
String keys = request.getParameter("keys", "");
@ -137,6 +133,25 @@ public class TelemetryRestMsgHandler extends DefaultRestMsgHandler {
}
}
private PluginCallback<List<TsKvEntry>> getTsKvListCallback(final PluginRestMsg msg) {
return new PluginCallback<List<TsKvEntry>>() {
@Override
public void onSuccess(PluginContext ctx, List<TsKvEntry> data) {
Map<String, List<TsData>> result = new LinkedHashMap<>();
for (TsKvEntry entry : data) {
result.put(entry.getKey(), data.stream().map(v -> new TsData(v.getTs(), v.getValueAsString())).collect(Collectors.toList()));
}
msg.getResponseHolder().setResult(new ResponseEntity<>(result, HttpStatus.OK));
}
@Override
public void onFailure(PluginContext ctx, Exception e) {
log.error("Failed to fetch historical data", e);
msg.getResponseHolder().setResult(new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR));
}
};
}
@Override
public void handleHttpPostRequest(PluginContext ctx, PluginRestMsg msg) throws ServletException {
RestRequest request = msg.getRequest();