tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Attribute scope check improvement

pull/9509/head
Andrii Shvaika 3 years ago
parent
d78bb26a73
commit
e20523b502
6 changed files with 55 additions and 35 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 18
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
  2. 48
      application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
  3. 6
      application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java
  4. 11
      application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
  5. 6
      application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java
  6. 1
      common/cluster-api/src/main/proto/queue.proto

18
application/src/main/java/org/thingsboard/server/service/subscription/DefaultSubscriptionManagerService.java
View File

@ -201,7 +201,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
if (serviceId.equals(targetId)) {
localSubscriptionService.onTimeSeriesUpdate(entityId, update, TbCallback.EMPTY);
} else {
sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(true, entityId, update));
sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(entityId, update));
}
}
@ -213,7 +213,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
@Override
public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice, TbCallback callback) {
getEntityUpdatesInfo(entityId).attributesUpdateTs = System.currentTimeMillis();
processAttributesUpdate(entityId, attributes);
processAttributesUpdate(entityId, scope, attributes);
if (entityId.getEntityType() == EntityType.DEVICE) {
if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope)) {
updateDeviceInactivityTimeout(tenantId, entityId, attributes);
@ -228,7 +228,7 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
@Override
public void onAttributesDelete(TenantId tenantId, EntityId entityId, String scope, List<String> keys, boolean notifyDevice, TbCallback callback) {
processAttributesUpdate(entityId,
processAttributesUpdate(entityId, scope,
keys.stream().map(key -> new BaseAttributeKvEntry(0, new StringDataEntry(key, ""))).collect(Collectors.toList()));
if (entityId.getEntityType() == EntityType.DEVICE) {
if (TbAttributeSubscriptionScope.SERVER_SCOPE.name().equalsIgnoreCase(scope)
@ -242,17 +242,17 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
callback.onSuccess();
}
public void processAttributesUpdate(EntityId entityId, List<AttributeKvEntry> update) {
public void processAttributesUpdate(EntityId entityId, String scope, List<AttributeKvEntry> update) {
TbEntityRemoteSubsInfo subInfo = entitySubscriptions.get(entityId);
if (subInfo != null) {
log.trace("[{}] Handling attributes update: {}", entityId, update);
subInfo.getSubs().forEach((serviceId, sub) -> {
if (sub.attrAllKeys) {
processAttributesUpdate(serviceId, entityId, update);
processAttributesUpdate(serviceId, entityId, scope, update);
} else if (sub.attrKeys != null) {
List<AttributeKvEntry> tmp = getSubList(update, sub.attrKeys);
if (tmp != null) {
processAttributesUpdate(serviceId, entityId, tmp);
processAttributesUpdate(serviceId, entityId, scope, tmp);
}
}
});
@ -261,12 +261,12 @@ public class DefaultSubscriptionManagerService extends TbApplicationEventListene
}
}
private void processAttributesUpdate(String targetId, EntityId entityId, List<AttributeKvEntry> update) {
private void processAttributesUpdate(String targetId, EntityId entityId, String scope, List<AttributeKvEntry> update) {
List<TsKvEntry> tsKvEntryList = update.stream().map(attr -> new BasicTsKvEntry(attr.getLastUpdateTs(), attr)).collect(Collectors.toList());
if (serviceId.equals(targetId)) {
localSubscriptionService.onAttributesUpdate(entityId, tsKvEntryList, TbCallback.EMPTY);
localSubscriptionService.onAttributesUpdate(entityId, scope, tsKvEntryList, TbCallback.EMPTY);
} else {
sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(false, entityId, tsKvEntryList));
sendCoreNotification(targetId, entityId, TbSubscriptionUtils.toProto(scope, entityId, tsKvEntryList));
}
}

48
application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbLocalSubscriptionService.java
View File

@ -262,37 +262,39 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
@Override
public void onAttributesUpdate(TransportProtos.TbSubUpdateProto proto, TbCallback callback) {
onAttributesUpdate(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()), TbSubscriptionUtils.fromProto(proto), callback);
onAttributesUpdate(new UUID(proto.getEntityIdMSB(), proto.getEntityIdLSB()), proto.getScope(), TbSubscriptionUtils.fromProto(proto), callback);
}
@Override
public void onAttributesUpdate(EntityId entityId, List<TsKvEntry> data, TbCallback callback) {
onAttributesUpdate(entityId.getId(), data, callback);
public void onAttributesUpdate(EntityId entityId, String scope, List<TsKvEntry> data, TbCallback callback) {
onAttributesUpdate(entityId.getId(), scope, data, callback);
}
private void onAttributesUpdate(UUID entityId, List<TsKvEntry> data, TbCallback callback) {
private void onAttributesUpdate(UUID entityId, String scope, List<TsKvEntry> data, TbCallback callback) {
entityUpdates.get(entityId).attributesUpdateTs = System.currentTimeMillis();
processSubscriptionData(entityId,
sub -> TbSubscriptionType.ATTRIBUTES.equals(sub.getType()),
s -> {
TbAttributeSubscription sub = (TbAttributeSubscription) s;
List<TsKvEntry> updateData = null;
if (sub.isAllKeys()) {
updateData = data;
} else {
for (TsKvEntry kv : data) {
if (sub.getKeyStates().containsKey((kv.getKey()))) {
if (updateData == null) {
updateData = new ArrayList<>();
if (sub.getScope() == null || TbAttributeSubscriptionScope.ANY_SCOPE.equals(sub.getScope()) || sub.getScope().name().equals(scope)) {
List<TsKvEntry> updateData = null;
if (sub.isAllKeys()) {
updateData = data;
} else {
for (TsKvEntry kv : data) {
if (sub.getKeyStates().containsKey((kv.getKey()))) {
if (updateData == null) {
updateData = new ArrayList<>();
}
updateData.add(kv);
}
updateData.add(kv);
}
}
}
if (updateData != null) {
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(sub.getSubscriptionId(), updateData);
update.getLatestValues().forEach((key, value) -> sub.getKeyStates().put(key, value));
subscriptionUpdateExecutor.submit(() -> sub.getUpdateProcessor().accept(sub, update));
if (updateData != null) {
TelemetrySubscriptionUpdate update = new TelemetrySubscriptionUpdate(sub.getSubscriptionId(), updateData);
update.getLatestValues().forEach((key, value) -> sub.getKeyStates().put(key, value));
subscriptionUpdateExecutor.submit(() -> sub.getUpdateProcessor().accept(sub, update));
}
}
}, callback);
}
@ -444,7 +446,13 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
return;
}
final Map<String, Long> keyStates = subscription.getKeyStates();
DonAsynchron.withCallback(attrService.find(subscription.getTenantId(), subscription.getEntityId(), DataConstants.CLIENT_SCOPE, keyStates.keySet()), values -> {
String scope;
if (subscription.getScope() != null && !TbAttributeSubscriptionScope.ANY_SCOPE.equals(subscription.getScope())) {
scope = subscription.getScope().name();
} else {
scope = DataConstants.CLIENT_SCOPE;
}
DonAsynchron.withCallback(attrService.find(subscription.getTenantId(), subscription.getEntityId(), scope, keyStates.keySet()), values -> {
List<TsKvEntry> updates = new ArrayList<>();
values.forEach(latestEntry -> {
if (latestEntry.getLastUpdateTs() > keyStates.get(latestEntry.getKey())) {
@ -453,7 +461,7 @@ public class DefaultTbLocalSubscriptionService implements TbLocalSubscriptionSer
});
var missedUpdates = updates.stream().filter(u -> u.getValue() != null).collect(Collectors.toList());
if (!missedUpdates.isEmpty()) {
onAttributesUpdate(subscription.getEntityId(), missedUpdates, TbCallback.EMPTY);
onAttributesUpdate(subscription.getEntityId(), scope, missedUpdates, TbCallback.EMPTY);
}
},
e -> log.error("Failed to fetch missed updates.", e), tsCallBackExecutor);

6
application/src/main/java/org/thingsboard/server/service/subscription/TbLocalSubscriptionService.java
View File

@ -18,16 +18,12 @@ package org.thingsboard.server.service.subscription;
import org.thingsboard.server.common.data.alarm.AlarmInfo;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.kv.AttributeKvEntry;
import org.thingsboard.server.common.data.kv.TsKvEntry;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.gen.transport.TransportProtos;
import org.thingsboard.server.queue.discovery.event.ClusterTopologyChangeEvent;
import org.thingsboard.server.queue.discovery.event.PartitionChangeEvent;
import org.thingsboard.server.service.ws.notification.sub.NotificationRequestUpdate;
import org.thingsboard.server.service.ws.notification.sub.NotificationsSubscriptionUpdate;
import org.thingsboard.server.service.ws.telemetry.sub.AlarmSubscriptionUpdate;
import org.thingsboard.server.service.ws.telemetry.sub.TelemetrySubscriptionUpdate;
import java.util.List;
@ -49,7 +45,7 @@ public interface TbLocalSubscriptionService {
void onAttributesUpdate(TransportProtos.TbSubUpdateProto attrUpdate, TbCallback callback);
void onAttributesUpdate(EntityId entityId, List<TsKvEntry> update, TbCallback callback);
void onAttributesUpdate(EntityId entityId, String scope, List<TsKvEntry> update, TbCallback callback);
void onAlarmUpdate(EntityId entityId, AlarmInfo alarm, boolean deleted, TbCallback callback);

11
application/src/main/java/org/thingsboard/server/service/subscription/TbSubscriptionUtils.java
View File

@ -411,7 +411,15 @@ public class TbSubscriptionUtils {
return result;
}
static ToCoreNotificationMsg toProto(boolean timeSeries, EntityId entityId, List<TsKvEntry> updates) {
static ToCoreNotificationMsg toProto(EntityId entityId, List<TsKvEntry> updates) {
return toProto(true, null, entityId, updates);
}
static ToCoreNotificationMsg toProto(String scope, EntityId entityId, List<TsKvEntry> updates) {
return toProto(false, scope, entityId, updates);
}
static ToCoreNotificationMsg toProto(boolean timeSeries, String scope, EntityId entityId, List<TsKvEntry> updates) {
TransportProtos.TbSubUpdateProto.Builder builder = TransportProtos.TbSubUpdateProto.newBuilder();
builder.setEntityIdMSB(entityId.getId().getMostSignificantBits());
@ -434,6 +442,7 @@ public class TbSubscriptionUtils {
if (timeSeries) {
result.setTsUpdate(builder);
} else {
builder.setScope(scope);
result.setAttrUpdate(builder);
}
return ToCoreNotificationMsg.newBuilder().setToLocalSubscriptionServiceMsg(result).build();

6
application/src/test/java/org/thingsboard/server/controller/WebsocketApiTest.java
View File

@ -454,6 +454,12 @@ public class WebsocketApiTest extends AbstractControllerTest {
getWsClient().registerWaitForUpdate();
// Pushing update with 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));
Assert.assertNull(getWsClient().waitForUpdate(3000));
AttributeKvEntry dataPoint1 = new BaseAttributeKvEntry(now - TimeUnit.MINUTES.toMillis(1), new LongDataEntry("serverAttributeKey", 42L));
List<AttributeKvEntry> tsData = Arrays.asList(dataPoint1);
sendAttributes(device, TbAttributeSubscriptionScope.SERVER_SCOPE, tsData);

1
common/cluster-api/src/main/proto/queue.proto
View File

@ -619,6 +619,7 @@ message TbSubUpdateProto {
int32 errorCode = 3;
string errorMsg = 4;
repeated TsValueListProto data = 5;
string scope = 6;
}
message TbAlarmSubUpdateProto {

Write Preview
Loading…
Cancel
Save