From 11d97b8a5dcf8618aecd6e73f721e972d9173d92 Mon Sep 17 00:00:00 2001 From: Andrii Landiak Date: Thu, 15 Feb 2024 15:39:02 +0200 Subject: [PATCH] Sync oauth settings in realtime --- .../edge/DefaultEdgeNotificationService.java | 7 +++++++ .../edge/EdgeEventSourcingListener.java | 5 +++++ .../edge/rpc/processor/BaseEdgeProcessor.java | 6 +++--- .../processor/oauth2/OAuth2EdgeProcessor.java | 18 ++++++++++++++++++ .../server/dao/oauth2/OAuth2ServiceImpl.java | 2 ++ 5 files changed, 35 insertions(+), 3 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java index 03fb982d6d..1136900a33 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java @@ -42,6 +42,7 @@ import org.thingsboard.server.service.edge.rpc.processor.device.DeviceEdgeProces import org.thingsboard.server.service.edge.rpc.processor.device.profile.DeviceProfileEdgeProcessor; import org.thingsboard.server.service.edge.rpc.processor.edge.EdgeProcessor; import org.thingsboard.server.service.edge.rpc.processor.entityview.EntityViewEdgeProcessor; +import org.thingsboard.server.service.edge.rpc.processor.oauth2.OAuth2EdgeProcessor; import org.thingsboard.server.service.edge.rpc.processor.ota.OtaPackageEdgeProcessor; import org.thingsboard.server.service.edge.rpc.processor.queue.QueueEdgeProcessor; import org.thingsboard.server.service.edge.rpc.processor.relation.RelationEdgeProcessor; @@ -126,6 +127,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { @Autowired private ResourceEdgeProcessor resourceEdgeProcessor; + @Autowired + private OAuth2EdgeProcessor oAuth2EdgeProcessor; + @Autowired protected ApplicationEventPublisher eventPublisher; @@ -231,6 +235,9 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService { case TB_RESOURCE: resourceEdgeProcessor.processEntityNotification(tenantId, edgeNotificationMsg); break; + case OAUTH2: + oAuth2EdgeProcessor.processOAuth2Notification(tenantId, edgeNotificationMsg); + break; default: log.warn("[{}] Edge event type [{}] is not designed to be pushed to edge", tenantId, type); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java index ca7048c7e9..f2dac98d47 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/EdgeEventSourcingListener.java @@ -31,6 +31,7 @@ import org.thingsboard.server.common.data.alarm.AlarmApiCallResult; import org.thingsboard.server.common.data.alarm.AlarmComment; import org.thingsboard.server.common.data.edge.EdgeEventActionType; import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.oauth2.OAuth2Info; import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChain; @@ -189,6 +190,8 @@ public class EdgeEventSourcingListener { private EdgeEventType getEdgeEventTypeForEntityEvent(Object entity) { if (entity instanceof AlarmComment) { return EdgeEventType.ALARM_COMMENT; + } else if (entity instanceof OAuth2Info) { + return EdgeEventType.OAUTH2; } return null; } @@ -196,6 +199,8 @@ public class EdgeEventSourcingListener { private String getBodyMsgForEntityEvent(Object entity) { if (entity instanceof AlarmComment) { return JacksonUtil.toString(entity); + } else if (entity instanceof OAuth2Info) { + return JacksonUtil.toString(entity); } return null; } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java index ec994e01d8..cd1876cd34 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java @@ -402,7 +402,7 @@ public abstract class BaseEdgeProcessor { protected ListenableFuture processActionForAllEdges(TenantId tenantId, EdgeEventType type, EdgeEventActionType actionType, EntityId entityId, - EdgeId sourceEdgeId) { + JsonNode body, EdgeId sourceEdgeId) { List> futures = new ArrayList<>(); if (TenantId.SYS_TENANT_ID.equals(tenantId)) { PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE); @@ -410,7 +410,7 @@ public abstract class BaseEdgeProcessor { do { tenantsIds = tenantService.findTenantsIds(pageLink); for (TenantId tenantId1 : tenantsIds.getData()) { - futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, null, sourceEdgeId)); + futures.addAll(processActionForAllEdgesByTenantId(tenantId1, type, actionType, entityId, body, sourceEdgeId)); } pageLink = pageLink.nextPageLink(); } while (tenantsIds.hasNext()); @@ -586,7 +586,7 @@ public abstract class BaseEdgeProcessor { case UPDATED: case DELETED: case CREDENTIALS_UPDATED: // used by USER entity - return processActionForAllEdges(tenantId, type, actionType, entityId, sourceEdgeId); + return processActionForAllEdges(tenantId, type, actionType, entityId, null, sourceEdgeId); default: return Futures.immediateFuture(null); } diff --git a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/oauth2/OAuth2EdgeProcessor.java b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/oauth2/OAuth2EdgeProcessor.java index 394b59c46a..1c62ad5de8 100644 --- a/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/oauth2/OAuth2EdgeProcessor.java +++ b/application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/oauth2/OAuth2EdgeProcessor.java @@ -15,14 +15,20 @@ */ package org.thingsboard.server.service.edge.rpc.processor.oauth2; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.server.common.data.EdgeUtils; import org.thingsboard.server.common.data.edge.EdgeEvent; +import org.thingsboard.server.common.data.edge.EdgeEventActionType; +import org.thingsboard.server.common.data.edge.EdgeEventType; +import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.oauth2.OAuth2Info; import org.thingsboard.server.gen.edge.v1.DownlinkMsg; import org.thingsboard.server.gen.edge.v1.OAuth2UpdateMsg; +import org.thingsboard.server.gen.transport.TransportProtos; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor; @@ -43,4 +49,16 @@ public class OAuth2EdgeProcessor extends BaseEdgeProcessor { } return downlinkMsg; } + + public ListenableFuture processOAuth2Notification(TenantId tenantId, TransportProtos.EdgeNotificationMsgProto edgeNotificationMsg) { + OAuth2Info oAuth2Info = JacksonUtil.fromString(edgeNotificationMsg.getBody(), OAuth2Info.class); + if (oAuth2Info == null) { + return Futures.immediateFuture(null); + } + + EdgeEventType type = EdgeEventType.valueOf(edgeNotificationMsg.getType()); + EdgeEventActionType actionType = EdgeEventActionType.valueOf(edgeNotificationMsg.getAction()); + return processActionForAllEdges(tenantId, type, actionType, null, JacksonUtil.toJsonNode(edgeNotificationMsg.getBody()), null); + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/oauth2/OAuth2ServiceImpl.java b/dao/src/main/java/org/thingsboard/server/dao/oauth2/OAuth2ServiceImpl.java index 8d031e78c2..f85a684e72 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/oauth2/OAuth2ServiceImpl.java +++ b/dao/src/main/java/org/thingsboard/server/dao/oauth2/OAuth2ServiceImpl.java @@ -40,6 +40,7 @@ import org.thingsboard.server.common.data.oauth2.PlatformType; import org.thingsboard.server.common.data.oauth2.SchemeType; import org.thingsboard.server.common.data.oauth2.TenantNameStrategyType; import org.thingsboard.server.dao.entity.AbstractEntityService; +import org.thingsboard.server.dao.eventsourcing.SaveEntityEvent; import org.thingsboard.server.dao.exception.DataValidationException; import org.thingsboard.server.dao.exception.IncorrectParameterException; @@ -117,6 +118,7 @@ public class OAuth2ServiceImpl extends AbstractEntityService implements OAuth2Se }); } }); + eventPublisher.publishEvent(SaveEntityEvent.builder().tenantId(TenantId.SYS_TENANT_ID).entity(oauth2Info).build()); } @Override