shared Attributes update improvements (If the update of the shared attributes is originated by the user's REST API call - we push it to the device, if the update arrives from the device, we will not push it back to the device by default.)
This commit is contained in:
parent
6824e1c376
commit
e41d69b1a8
@ -216,6 +216,11 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback) {
|
public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback) {
|
||||||
|
onAttributesUpdate(tenantId, entityId, scope, attributes, callback, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback, boolean notifyDevice) {
|
||||||
onLocalSubUpdate(entityId,
|
onLocalSubUpdate(entityId,
|
||||||
s -> {
|
s -> {
|
||||||
if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) {
|
if (TbSubscriptionType.ATTRIBUTES.equals(s.getType())) {
|
||||||
@ -244,7 +249,7 @@ public class DefaultSubscriptionManagerService implements SubscriptionManagerSer
|
|||||||
deviceStateService.onDeviceInactivityTimeoutUpdate(new DeviceId(entityId.getId()), attribute.getLongValue().orElse(0L));
|
deviceStateService.onDeviceInactivityTimeoutUpdate(new DeviceId(entityId.getId()), attribute.getLongValue().orElse(0L));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope)) {
|
} else if (TbAttributeSubscriptionScope.SHARED_SCOPE.name().equalsIgnoreCase(scope) && notifyDevice) {
|
||||||
clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(tenantId,
|
clusterService.pushMsgToCore(DeviceAttributesEventNotificationMsg.onUpdate(tenantId,
|
||||||
new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes))
|
new DeviceId(entityId.getId()), DataConstants.SHARED_SCOPE, new ArrayList<>(attributes))
|
||||||
, null);
|
, null);
|
||||||
|
|||||||
@ -35,4 +35,6 @@ public interface SubscriptionManagerService extends ApplicationListener<Partitio
|
|||||||
|
|
||||||
void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
|
void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback);
|
||||||
|
|
||||||
|
void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, TbCallback callback, boolean notifyDevice);
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|||||||
@ -128,9 +128,14 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) {
|
public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback) {
|
||||||
|
saveAndNotify(tenantId, entityId, scope, attributes, callback, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback, boolean notifyDevice) {
|
||||||
ListenableFuture<List<Void>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);
|
ListenableFuture<List<Void>> saveFuture = attrService.save(tenantId, entityId, scope, attributes);
|
||||||
addMainCallback(saveFuture, callback);
|
addMainCallback(saveFuture, callback);
|
||||||
addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes));
|
addWsCallback(saveFuture, success -> onAttributesUpdate(tenantId, entityId, scope, attributes, notifyDevice));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
@ -157,11 +162,11 @@ public class DefaultTelemetrySubscriptionService implements TelemetrySubscriptio
|
|||||||
, System.currentTimeMillis())), callback);
|
, System.currentTimeMillis())), callback);
|
||||||
}
|
}
|
||||||
|
|
||||||
private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes) {
|
private void onAttributesUpdate(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, boolean notifyDevice) {
|
||||||
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
|
TopicPartitionInfo tpi = partitionService.resolve(ServiceType.TB_CORE, tenantId, entityId);
|
||||||
if (currentPartitions.contains(tpi)) {
|
if (currentPartitions.contains(tpi)) {
|
||||||
if (subscriptionManagerService.isPresent()) {
|
if (subscriptionManagerService.isPresent()) {
|
||||||
subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY);
|
subscriptionManagerService.get().onAttributesUpdate(tenantId, entityId, scope, attributes, TbCallback.EMPTY, notifyDevice);
|
||||||
} else {
|
} else {
|
||||||
log.warn("Possible misconfiguration because subscriptionManagerService is null!");
|
log.warn("Possible misconfiguration because subscriptionManagerService is null!");
|
||||||
}
|
}
|
||||||
|
|||||||
@ -298,6 +298,7 @@ public class DefaultTransportService implements TransportService {
|
|||||||
TbMsgMetaData metaData = new TbMsgMetaData();
|
TbMsgMetaData metaData = new TbMsgMetaData();
|
||||||
metaData.putValue("deviceName", sessionInfo.getDeviceName());
|
metaData.putValue("deviceName", sessionInfo.getDeviceName());
|
||||||
metaData.putValue("deviceType", sessionInfo.getDeviceType());
|
metaData.putValue("deviceType", sessionInfo.getDeviceType());
|
||||||
|
metaData.putValue("notifyDevice", "false");
|
||||||
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, metaData, gson.toJson(json));
|
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), deviceId, metaData, gson.toJson(json));
|
||||||
sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback));
|
sendToRuleEngine(tenantId, tbMsg, new TransportTbQueueCallback(callback));
|
||||||
}
|
}
|
||||||
|
|||||||
@ -36,6 +36,8 @@ public interface RuleEngineTelemetryService {
|
|||||||
|
|
||||||
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
|
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback);
|
||||||
|
|
||||||
|
void saveAndNotify(TenantId tenantId, EntityId entityId, String scope, List<AttributeKvEntry> attributes, FutureCallback<Void> callback, boolean notifyDevice);
|
||||||
|
|
||||||
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback);
|
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, long value, FutureCallback<Void> callback);
|
||||||
|
|
||||||
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value, FutureCallback<Void> callback);
|
void saveAttrAndNotify(TenantId tenantId, EntityId entityId, String scope, String key, String value, FutureCallback<Void> callback);
|
||||||
|
|||||||
@ -17,6 +17,7 @@ package org.thingsboard.rule.engine.telemetry;
|
|||||||
|
|
||||||
import com.google.gson.JsonParser;
|
import com.google.gson.JsonParser;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.apache.commons.lang3.StringUtils;
|
||||||
import org.thingsboard.rule.engine.api.RuleNode;
|
import org.thingsboard.rule.engine.api.RuleNode;
|
||||||
import org.thingsboard.rule.engine.api.TbContext;
|
import org.thingsboard.rule.engine.api.TbContext;
|
||||||
import org.thingsboard.rule.engine.api.TbNode;
|
import org.thingsboard.rule.engine.api.TbNode;
|
||||||
@ -63,7 +64,14 @@ public class TbMsgAttributesNode implements TbNode {
|
|||||||
}
|
}
|
||||||
String src = msg.getData();
|
String src = msg.getData();
|
||||||
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src));
|
Set<AttributeKvEntry> attributes = JsonConverter.convertToAttributes(new JsonParser().parse(src));
|
||||||
ctx.getTelemetryService().saveAndNotify(ctx.getTenantId(), msg.getOriginator(), config.getScope(), new ArrayList<>(attributes), new TelemetryNodeCallback(ctx, msg));
|
String notifyDeviceStr = msg.getMetaData().getValue("notifyDevice");
|
||||||
|
ctx.getTelemetryService().saveAndNotify(
|
||||||
|
ctx.getTenantId(),
|
||||||
|
msg.getOriginator(),
|
||||||
|
config.getScope(),
|
||||||
|
new ArrayList<>(attributes),
|
||||||
|
new TelemetryNodeCallback(ctx, msg),
|
||||||
|
StringUtils.isEmpty(notifyDeviceStr) || !notifyDeviceStr.equals("false"));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user