Sync oauth settings in realtime

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);
                     }

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;
     }

application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java

@@ -402,7 +402,7 @@ public abstract class BaseEdgeProcessor {
 
     protected ListenableFuture<Void> processActionForAllEdges(TenantId tenantId, EdgeEventType type,
                                                               EdgeEventActionType actionType, EntityId entityId,
-                                                              EdgeId sourceEdgeId) {
+                                                              JsonNode body, EdgeId sourceEdgeId) {
         List<ListenableFuture<Void>> 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);
         }

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<Void> 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);
+    }
+
 }

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