简柏林 / thingskit · Commits

Commit ccd4f4b09692c7b1f673e28238007f19182f744a

Authored by Andrew Shvayka
1 parent 15f442fb

Improvements

Inline Side-by-side
Showing 12 changed files with 72 additions and 47 deletions
@@ -18,6 +18,8 @@ package org.thingsboard.server.common.msg.core; @@ -18,6 +18,8 @@ package org.thingsboard.server.common.msg.core;
18 import lombok.ToString; 18 import lombok.ToString;
19 import org.thingsboard.server.common.msg.session.MsgType; 19 import org.thingsboard.server.common.msg.session.MsgType;
20 20
  21 +import java.util.Collections;
  22 +import java.util.Optional;
21 import java.util.Set; 23 import java.util.Set;
22 24
23 @ToString 25 @ToString
@@ -28,6 +30,10 @@ public class BasicGetAttributesRequest extends BasicRequest implements GetAttrib @@ -28,6 +30,10 @@ public class BasicGetAttributesRequest extends BasicRequest implements GetAttrib
28 private final Set<String> clientKeys; 30 private final Set<String> clientKeys;
29 private final Set<String> sharedKeys; 31 private final Set<String> sharedKeys;
30 32
  33 + public BasicGetAttributesRequest(Integer requestId) {
  34 + this(requestId, Collections.emptySet(), Collections.emptySet());
  35 + }
  36 +
31 public BasicGetAttributesRequest(Integer requestId, Set<String> clientKeys, Set<String> sharedKeys) { 37 public BasicGetAttributesRequest(Integer requestId, Set<String> clientKeys, Set<String> sharedKeys) {
32 super(requestId); 38 super(requestId);
33 this.clientKeys = clientKeys; 39 this.clientKeys = clientKeys;
@@ -40,13 +46,13 @@ public class BasicGetAttributesRequest extends BasicRequest implements GetAttrib @@ -40,13 +46,13 @@ public class BasicGetAttributesRequest extends BasicRequest implements GetAttrib
40 } 46 }
41 47
42 @Override 48 @Override
43 - public Set<String> getClientAttributeNames() {  
44 - return clientKeys; 49 + public Optional<Set<String>> getClientAttributeNames() {
  50 + return Optional.of(clientKeys);
45 } 51 }
46 52
47 @Override 53 @Override
48 - public Set<String> getSharedAttributeNames() {  
49 - return sharedKeys; 54 + public Optional<Set<String>> getSharedAttributeNames() {
  55 + return Optional.ofNullable(sharedKeys);
50 } 56 }
51 57
52 } 58 }
@@ -15,6 +15,7 @@ @@ -15,6 +15,7 @@
15 */ 15 */
16 package org.thingsboard.server.common.msg.core; 16 package org.thingsboard.server.common.msg.core;
17 17
  18 +import java.util.Optional;
18 import java.util.Set; 19 import java.util.Set;
19 20
20 import org.thingsboard.server.common.msg.session.FromDeviceMsg; 21 import org.thingsboard.server.common.msg.session.FromDeviceMsg;
@@ -22,7 +23,7 @@ import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg; @@ -22,7 +23,7 @@ import org.thingsboard.server.common.msg.session.FromDeviceRequestMsg;
22 23
23 public interface GetAttributesRequest extends FromDeviceRequestMsg { 24 public interface GetAttributesRequest extends FromDeviceRequestMsg {
24 25
25 - Set<String> getClientAttributeNames();  
26 - Set<String> getSharedAttributeNames(); 26 + Optional<Set<String>> getClientAttributeNames();
  27 + Optional<Set<String>> getSharedAttributeNames();
27 28
28 } 29 }
@@ -24,10 +24,6 @@ import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionT @@ -24,10 +24,6 @@ import org.thingsboard.server.extensions.core.plugin.telemetry.sub.SubscriptionT
24 @NoArgsConstructor 24 @NoArgsConstructor
25 public class AttributesSubscriptionCmd extends SubscriptionCmd { 25 public class AttributesSubscriptionCmd extends SubscriptionCmd {
26 26
27 - public AttributesSubscriptionCmd(int cmdId, String deviceId, String keys, boolean unsubscribe) {  
28 - super(cmdId, deviceId, keys, unsubscribe);  
29 - }  
30 -  
31 @Override 27 @Override
32 public SubscriptionType getType() { 28 public SubscriptionType getType() {
33 return SubscriptionType.ATTRIBUTES; 29 return SubscriptionType.ATTRIBUTES;
@@ -26,6 +26,7 @@ public abstract class SubscriptionCmd implements TelemetryPluginCmd { @@ -26,6 +26,7 @@ public abstract class SubscriptionCmd implements TelemetryPluginCmd {
26 private int cmdId; 26 private int cmdId;
27 private String deviceId; 27 private String deviceId;
28 private String keys; 28 private String keys;
  29 + private String scope;
29 private boolean unsubscribe; 30 private boolean unsubscribe;
30 31
31 public abstract SubscriptionType getType(); 32 public abstract SubscriptionType getType();
@@ -62,6 +63,14 @@ public abstract class SubscriptionCmd implements TelemetryPluginCmd { @@ -62,6 +63,14 @@ public abstract class SubscriptionCmd implements TelemetryPluginCmd {
62 this.unsubscribe = unsubscribe; 63 this.unsubscribe = unsubscribe;
63 } 64 }
64 65
  66 + public String getScope() {
  67 + return scope;
  68 + }
  69 +
  70 + public void setKeys(String keys) {
  71 + this.keys = keys;
  72 + }
  73 +
65 @Override 74 @Override
66 public String toString() { 75 public String toString() {
67 return "SubscriptionCmd [deviceId=" + deviceId + ", tags=" + keys + ", unsubscribe=" + unsubscribe + "]"; 76 return "SubscriptionCmd [deviceId=" + deviceId + ", tags=" + keys + ", unsubscribe=" + unsubscribe + "]";
@@ -26,11 +26,6 @@ public class TimeseriesSubscriptionCmd extends SubscriptionCmd { @@ -26,11 +26,6 @@ public class TimeseriesSubscriptionCmd extends SubscriptionCmd {
26 26
27 private long timeWindow; 27 private long timeWindow;
28 28
29 - public TimeseriesSubscriptionCmd(int cmdId, String deviceId, String keys, boolean unsubscribe, long timeWindow) {  
30 - super(cmdId, deviceId, keys, unsubscribe);  
31 - this.timeWindow = timeWindow;  
32 - }  
33 -  
34 public long getTimeWindow() { 29 public long getTimeWindow() {
35 return timeWindow; 30 return timeWindow;
36 } 31 }
@@ -58,10 +58,14 @@ public class TelemetryRuleMsgHandler extends DefaultRuleMsgHandler { @@ -58,10 +58,14 @@ public class TelemetryRuleMsgHandler extends DefaultRuleMsgHandler {
58 ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId, response)); 58 ctx.reply(new ResponsePluginToRuleMsg(msg.getUid(), tenantId, ruleId, response));
59 } 59 }
60 60
61 - private List<AttributeKvEntry> getAttributeKvEntries(PluginContext ctx, DeviceId deviceId, String scope, Set<String> names) { 61 + private List<AttributeKvEntry> getAttributeKvEntries(PluginContext ctx, DeviceId deviceId, String scope, Optional<Set<String>> names) {
62 List<AttributeKvEntry> attributes; 62 List<AttributeKvEntry> attributes;
63 - if (!names.isEmpty()) {  
64 - attributes = ctx.loadAttributes(deviceId, scope, new ArrayList<>(names)); 63 + if (names.isPresent()) {
  64 + if (!names.get().isEmpty()) {
  65 + attributes = ctx.loadAttributes(deviceId, scope, new ArrayList<>(names.get()));
  66 + } else {
  67 + attributes = ctx.loadAttributes(deviceId, scope);
  68 + }
65 } else { 69 } else {
66 attributes = Collections.emptyList(); 70 attributes = Collections.emptyList();
67 } 71 }
@@ -105,7 +105,12 @@ public class TelemetryWebsocketMsgHandler extends DefaultWebsocketMsgHandler { @@ -105,7 +105,12 @@ public class TelemetryWebsocketMsgHandler extends DefaultWebsocketMsgHandler {
105 if (keysOptional.isPresent()) { 105 if (keysOptional.isPresent()) {
106 List<String> keys = new ArrayList<>(keysOptional.get()); 106 List<String> keys = new ArrayList<>(keysOptional.get());
107 List<AttributeKvEntry> data = new ArrayList<>(); 107 List<AttributeKvEntry> data = new ArrayList<>();
108 - Arrays.stream(DataConstants.ALL_SCOPES).forEach(s -> data.addAll(ctx.loadAttributes(deviceId, s, keys))); 108 + if (StringUtils.isEmpty(cmd.getScope())) {
  109 + Arrays.stream(DataConstants.ALL_SCOPES).forEach(s -> data.addAll(ctx.loadAttributes(deviceId, s, keys)));
  110 + } else {
  111 + data.addAll(ctx.loadAttributes(deviceId, cmd.getScope(), keys));
  112 + }
  113 +
109 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); 114 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList());
110 sendWsMsg(ctx, sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData)); 115 sendWsMsg(ctx, sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData));
111 116
@@ -116,7 +121,11 @@ public class TelemetryWebsocketMsgHandler extends DefaultWebsocketMsgHandler { @@ -116,7 +121,11 @@ public class TelemetryWebsocketMsgHandler extends DefaultWebsocketMsgHandler {
116 sub = new SubscriptionState(sessionId, cmd.getCmdId(), deviceId, SubscriptionType.ATTRIBUTES, false, subState); 121 sub = new SubscriptionState(sessionId, cmd.getCmdId(), deviceId, SubscriptionType.ATTRIBUTES, false, subState);
117 } else { 122 } else {
118 List<AttributeKvEntry> data = new ArrayList<>(); 123 List<AttributeKvEntry> data = new ArrayList<>();
119 - Arrays.stream(DataConstants.ALL_SCOPES).forEach(s -> data.addAll(ctx.loadAttributes(deviceId, s))); 124 + if (StringUtils.isEmpty(cmd.getScope())) {
  125 + Arrays.stream(DataConstants.ALL_SCOPES).forEach(s -> data.addAll(ctx.loadAttributes(deviceId, s)));
  126 + } else {
  127 + data.addAll(ctx.loadAttributes(deviceId, cmd.getScope()));
  128 + }
120 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList()); 129 List<TsKvEntry> attributesData = data.stream().map(d -> new BasicTsKvEntry(d.getLastUpdateTs(), d)).collect(Collectors.toList());
121 sendWsMsg(ctx, sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData)); 130 sendWsMsg(ctx, sessionRef, new SubscriptionUpdate(cmd.getCmdId(), attributesData));
122 131
@@ -167,17 +167,13 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { @@ -167,17 +167,13 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
167 167
168 private FromDeviceMsg convertToGetAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException { 168 private FromDeviceMsg convertToGetAttributesRequest(SessionContext ctx, Request inbound) throws AdaptorException {
169 List<String> queryElements = inbound.getOptions().getUriQuery(); 169 List<String> queryElements = inbound.getOptions().getUriQuery();
170 - if (queryElements == null || queryElements.size() == 0) {  
171 - log.warn("[{}] Query is empty!", ctx.getSessionId());  
172 - throw new AdaptorException(new IllegalArgumentException("Query is empty!"));  
173 - }  
174 -  
175 - Set<String> clientKeys = toKeys(ctx, queryElements, "clientKeys");  
176 - Set<String> sharedKeys = toKeys(ctx, queryElements, "sharedKeys");  
177 - if (clientKeys.isEmpty() && sharedKeys.isEmpty()) {  
178 - throw new AdaptorException("No clientKeys and serverKeys parameters!"); 170 + if (queryElements != null || queryElements.size() > 0) {
  171 + Set<String> clientKeys = toKeys(ctx, queryElements, "clientKeys");
  172 + Set<String> sharedKeys = toKeys(ctx, queryElements, "sharedKeys");
  173 + return new BasicGetAttributesRequest(0, clientKeys, sharedKeys);
  174 + } else {
  175 + return new BasicGetAttributesRequest(0);
179 } 176 }
180 - return new BasicGetAttributesRequest(0, clientKeys, sharedKeys);  
181 } 177 }
182 178
183 private Set<String> toKeys(SessionContext ctx, List<String> queryElements, String attributeName) throws AdaptorException { 179 private Set<String> toKeys(SessionContext ctx, List<String> queryElements, String attributeName) throws AdaptorException {
@@ -191,7 +187,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor { @@ -191,7 +187,7 @@ public class JsonCoapAdaptor implements CoapTransportAdaptor {
191 if (!StringUtils.isEmpty(keys)) { 187 if (!StringUtils.isEmpty(keys)) {
192 return new HashSet<>(Arrays.asList(keys.split(","))); 188 return new HashSet<>(Arrays.asList(keys.split(",")));
193 } else { 189 } else {
194 - return Collections.emptySet(); 190 + return null;
195 } 191 }
196 } 192 }
197 193
@@ -182,7 +182,7 @@ public class CoapServerTest { @@ -182,7 +182,7 @@ public class CoapServerTest {
182 public void testNoKeysAttributesGetRequest() { 182 public void testNoKeysAttributesGetRequest() {
183 CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.ATTRIBUTES.name().toLowerCase() + "?data=key1,key2"); 183 CoapClient client = new CoapClient(getBaseTestUrl() + DEVICE1_TOKEN + "/" + FeatureType.ATTRIBUTES.name().toLowerCase() + "?data=key1,key2");
184 CoapResponse response = client.setTimeout(6000).get(); 184 CoapResponse response = client.setTimeout(6000).get();
185 - Assert.assertEquals(ResponseCode.BAD_REQUEST, response.getCode()); 185 + Assert.assertEquals(ResponseCode.CONTENT, response.getCode());
186 } 186 }
187 187
188 @Test 188 @Test
@@ -38,6 +38,7 @@ import org.thingsboard.server.common.transport.auth.DeviceAuthService; @@ -38,6 +38,7 @@ import org.thingsboard.server.common.transport.auth.DeviceAuthService;
38 import org.thingsboard.server.transport.http.session.HttpSessionCtx; 38 import org.thingsboard.server.transport.http.session.HttpSessionCtx;
39 39
40 import java.util.Arrays; 40 import java.util.Arrays;
  41 +import java.util.Collections;
41 import java.util.HashSet; 42 import java.util.HashSet;
42 import java.util.Set; 43 import java.util.Set;
43 44
@@ -60,20 +61,22 @@ public class DeviceApiController { @@ -60,20 +61,22 @@ public class DeviceApiController {
60 61
61 @RequestMapping(value = "/{deviceToken}/attributes", method = RequestMethod.GET, produces = "application/json") 62 @RequestMapping(value = "/{deviceToken}/attributes", method = RequestMethod.GET, produces = "application/json")
62 public DeferredResult<ResponseEntity> getDeviceAttributes(@PathVariable("deviceToken") String deviceToken, 63 public DeferredResult<ResponseEntity> getDeviceAttributes(@PathVariable("deviceToken") String deviceToken,
63 - @RequestParam(value = "clientKeys", required = false) String clientKeys,  
64 - @RequestParam(value = "sharedKeys", required = false) String sharedKeys) { 64 + @RequestParam(value = "clientKeys", required = false, defaultValue = "") String clientKeys,
  65 + @RequestParam(value = "sharedKeys", required = false, defaultValue = "") String sharedKeys) {
65 DeferredResult<ResponseEntity> responseWriter = new DeferredResult<ResponseEntity>(); 66 DeferredResult<ResponseEntity> responseWriter = new DeferredResult<ResponseEntity>();
66 - if (StringUtils.isEmpty(clientKeys) && StringUtils.isEmpty(sharedKeys)) {  
67 - responseWriter.setResult(new ResponseEntity<>(HttpStatus.BAD_REQUEST));  
68 - } else {  
69 - HttpSessionCtx ctx = getHttpSessionCtx(responseWriter);  
70 - if (ctx.login(new DeviceTokenCredentials(deviceToken))) {  
71 - Set<String> clientKeySet = new HashSet<>(Arrays.asList(clientKeys.split(",")));  
72 - Set<String> sharedKeySet = new HashSet<>(Arrays.asList(clientKeys.split(",")));  
73 - process(ctx, new BasicGetAttributesRequest(0, clientKeySet, sharedKeySet)); 67 + HttpSessionCtx ctx = getHttpSessionCtx(responseWriter);
  68 + if (ctx.login(new DeviceTokenCredentials(deviceToken))) {
  69 + GetAttributesRequest request;
  70 + if (StringUtils.isEmpty(clientKeys) && StringUtils.isEmpty(sharedKeys)) {
  71 + request = new BasicGetAttributesRequest(0);
74 } else { 72 } else {
75 - responseWriter.setResult(new ResponseEntity<>(HttpStatus.UNAUTHORIZED)); 73 + Set<String> clientKeySet = !StringUtils.isEmpty(clientKeys) ? new HashSet<>(Arrays.asList(clientKeys.split(","))) : null;
  74 + Set<String> sharedKeySet = !StringUtils.isEmpty(sharedKeys) ? new HashSet<>(Arrays.asList(sharedKeys.split(","))) : null;
  75 + request = new BasicGetAttributesRequest(0, clientKeySet, sharedKeySet);
76 } 76 }
  77 + process(ctx, request);
  78 + } else {
  79 + responseWriter.setResult(new ResponseEntity<>(HttpStatus.UNAUTHORIZED));
77 } 80 }
78 81
79 return responseWriter; 82 return responseWriter;
@@ -162,8 +162,13 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { @@ -162,8 +162,13 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
162 Integer requestId = Integer.valueOf(topicName.substring(MqttTransportHandler.ATTRIBUTES_REQUEST_TOPIC_PREFIX.length())); 162 Integer requestId = Integer.valueOf(topicName.substring(MqttTransportHandler.ATTRIBUTES_REQUEST_TOPIC_PREFIX.length()));
163 String payload = inbound.payload().toString(UTF8); 163 String payload = inbound.payload().toString(UTF8);
164 JsonElement requestBody = new JsonParser().parse(payload); 164 JsonElement requestBody = new JsonParser().parse(payload);
165 - return new BasicGetAttributesRequest(requestId,  
166 - toStringSet(requestBody, "clientKeys"), toStringSet(requestBody, "sharedKeys")); 165 + Set<String> clientKeys = toStringSet(requestBody, "clientKeys");
  166 + Set<String> sharedKeys = toStringSet(requestBody, "sharedKeys");
  167 + if (clientKeys == null && sharedKeys == null) {
  168 + return new BasicGetAttributesRequest(requestId);
  169 + } else {
  170 + return new BasicGetAttributesRequest(requestId, clientKeys, sharedKeys);
  171 + }
167 } catch (RuntimeException e) { 172 } catch (RuntimeException e) {
168 log.warn("Failed to decode get attributes request", e); 173 log.warn("Failed to decode get attributes request", e);
169 throw new AdaptorException(e); 174 throw new AdaptorException(e);
@@ -189,7 +194,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor { @@ -189,7 +194,7 @@ public class JsonMqttAdaptor implements MqttTransportAdaptor {
189 if (element != null) { 194 if (element != null) {
190 return new HashSet<>(Arrays.asList(element.getAsString().split(","))); 195 return new HashSet<>(Arrays.asList(element.getAsString().split(",")));
191 } else { 196 } else {
192 - return Collections.emptySet(); 197 + return null;
193 } 198 }
194 } 199 }
195 200
@@ -293,7 +293,8 @@ function DeviceService($http, $q, $filter, telemetryWebsocketService, types) { @@ -293,7 +293,8 @@ function DeviceService($http, $q, $filter, telemetryWebsocketService, types) {
293 var deviceAttributesSubscription = deviceAttributesSubscriptionMap[subscriptionId]; 293 var deviceAttributesSubscription = deviceAttributesSubscriptionMap[subscriptionId];
294 if (!deviceAttributesSubscription) { 294 if (!deviceAttributesSubscription) {
295 var subscriptionCommand = { 295 var subscriptionCommand = {
296 - deviceId: deviceId 296 + deviceId: deviceId,
  297 + scope: attributeScope
297 }; 298 };
298 299
299 var type = attributeScope === types.latestTelemetry.value ? 300 var type = attributeScope === types.latestTelemetry.value ?