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

Merge branch 'master' of github.com:thingsboard/thingsboard

Browse Source
This commit is contained in:
Igor Kulikov 2017-03-14 15:23:24 +02:00
commit 062c8b30a0
1 changed files with 32 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -87,30 +87,26 @@ public class TelemetryRestMsgHandler extends DefaultRestMsgHandler {
} else if (method.equals("values")) { } else if (method.equals("values")) {
if ("timeseries".equals(entity)) { if ("timeseries".equals(entity)) {
String keysStr = request.getParameter("keys"); String keysStr = request.getParameter("keys");
List<String> keys = Arrays.asList(keysStr.split(","));
Optional<Long> startTs = request.getLongParamValue("startTs"); Optional<Long> startTs = request.getLongParamValue("startTs");
Optional<Long> endTs = request.getLongParamValue("endTs"); Optional<Long> endTs = request.getLongParamValue("endTs");
Optional<Long> interval = request.getLongParamValue("interval"); Optional<Long> interval = request.getLongParamValue("interval");
Optional<Integer> limit = request.getIntParamValue("limit"); Optional<Integer> limit = request.getIntParamValue("limit");
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())); 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))
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()); .collect(Collectors.toList());
ctx.loadTimeseries(deviceId, queries, new PluginCallback<List<TsKvEntry>>() { ctx.loadTimeseries(deviceId, queries, getTsKvListCallback(msg));
@Override } else {
public void onSuccess(PluginContext ctx, List<TsKvEntry> data) { ctx.loadLatestTimeseries(deviceId, keys, getTsKvListCallback(msg));
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));
}
@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));
}
});
} else if ("attributes".equals(entity)) { } else if ("attributes".equals(entity)) {
String keys = request.getParameter("keys", ""); 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 @Override
public void handleHttpPostRequest(PluginContext ctx, PluginRestMsg msg) throws ServletException { public void handleHttpPostRequest(PluginContext ctx, PluginRestMsg msg) throws ServletException {
RestRequest request = msg.getRequest(); RestRequest request = msg.getRequest();