One post tagged with "apollo"

@Validated@RequiredArgsConstructor@RestController@RequestMapping("/plugin")public class PluginController {
  @PostMapping("")  @RequiresPermissions("system:plugin:add")  public ShenyuAdminResult createPlugin(@Valid @ModelAttribute final PluginDTO pluginDTO) {      // Call pluginService.createOrUpdate for processing logic      return ShenyuAdminResult.success(pluginService.createOrUpdate(pluginDTO));  }        // ......}

@RequiredArgsConstructor@Servicepublic class PluginServiceImpl implements SelectorService {    // Event publishing object pluginEventPublisher    private final PluginEventPublisher pluginEventPublisher;
   private String create(final PluginDTO pluginDTO) {      // Check if there is a corresponding plugin      Assert.isNull(pluginMapper.nameExisted(pluginDTO.getName()), AdminConstants.PLUGIN_NAME_IS_EXIST);      // check if Customized plugin jar      if (!Objects.isNull(pluginDTO.getFile())) {        Assert.isTrue(checkFile(Base64.decode(pluginDTO.getFile())), AdminConstants.THE_PLUGIN_JAR_FILE_IS_NOT_CORRECT_OR_EXCEEDS_16_MB);      }      // Create plugin object      PluginDO pluginDO = PluginDO.buildPluginDO(pluginDTO);      // Insert object into database      if (pluginMapper.insertSelective(pluginDO) > 0) {        // publish create event. init plugin data        pluginEventPublisher.onCreated(pluginDO);      }      return ShenyuResultMessage.CREATE_SUCCESS;  }            // ......    }

    @Overridepublic void onCreated(final PluginDO plugin) {        // Publish DataChangeEvent events: event grouping (plugins, selectors, rules), event types (create, delete, update), changed data        publisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, DataEventTypeEnum.CREATE,        Collections.singletonList(PluginTransfer.INSTANCE.mapToData(plugin))));        // Publish PluginCreatedEvent        publish(new PluginCreatedEvent(plugin, SessionUtil.visitorName()));}

public class DataChangedEvent extends ApplicationEvent {//......}

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
    //......    }

@Componentpublic class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
  /**     * When there is a data change, call this method     * @param event     */  @Override  @SuppressWarnings("unchecked")  public void onApplicationEvent(final DataChangedEvent event) {    // Traverse data change listeners (only ApolloDataChangedListener will be registered here)    for (DataChangedListener listener : listeners) {      // Forward according to different grouping types      switch (event.getGroupKey()) {        case APP_AUTH: // authentication information          listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());          break;        case PLUGIN: // Plugin events          // Calling the registered listener object          listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());          break;        case RULE: // Rule events          listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());          break;        case SELECTOR: // Selector event          listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());          break;        case META_DATA: // Metadata events          listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());          break;        case PROXY_SELECTOR: // Proxy selector event          listener.onProxySelectorChanged((List<ProxySelectorData>) event.getSource(), event.getEventType());          break;        case DISCOVER_UPSTREAM: // Registration discovery of downstream list events          listener.onDiscoveryUpstreamChanged((List<DiscoverySyncData>) event.getSource(), event.getEventType());          applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnUpstreamChanged((List<DiscoverySyncData>) event.getSource(), event.getEventType());          break;        default:          throw new IllegalStateException("Unexpected value: " + event.getGroupKey());      }    }  }    }

// Inheriting AbstractNodeDataChangedListenerpublic class ApolloDataChangedListener extends AbstractNodeDataChangedListener {    }

// Using key as the base class for finding storage methodspublic abstract class AbstractNodeDataChangedListener implements DataChangedListener {         protected AbstractNodeDataChangedListener(final ChangeData changeData) {      this.changeData = changeData;    }}

public class ApolloDataChangedListener extends AbstractNodeDataChangedListener {  public ApolloDataChangedListener(final ApolloClient apolloClient) {    // Configure prefixes for several types of grouped data    super(new ChangeData(ApolloPathConstants.PLUGIN_DATA_ID,            ApolloPathConstants.SELECTOR_DATA_ID,            ApolloPathConstants.RULE_DATA_ID,            ApolloPathConstants.AUTH_DATA_ID,            ApolloPathConstants.META_DATA_ID,            ApolloPathConstants.PROXY_SELECTOR_DATA_ID,            ApolloPathConstants.DISCOVERY_DATA_ID));    // Manipulating objects of Apollo    this.apolloClient = apolloClient;  }}

// Data Change Listenerpublic interface DataChangedListener {
    // Call when authorization information changes    default void onAppAuthChanged(List<AppAuthData> changed, DataEventTypeEnum eventType) {    }
    // Called when plugin information changes    default void onPluginChanged(List<PluginData> changed, DataEventTypeEnum eventType) {    }
    // Called when selector information changes    default void onSelectorChanged(List<SelectorData> changed, DataEventTypeEnum eventType) {    }         // Called when metadata information changes    default void onMetaDataChanged(List<MetaData> changed, DataEventTypeEnum eventType) {
    }
    // Call when rule information changes    default void onRuleChanged(List<RuleData> changed, DataEventTypeEnum eventType) {    }
    // Called when proxy selector changes    default void onProxySelectorChanged(List<ProxySelectorData> changed, DataEventTypeEnum eventType) {    }    // Called when downstream information changes are discovered    default void onDiscoveryUpstreamChanged(List<DiscoverySyncData> changed, DataEventTypeEnum eventType) {    }
}

public abstract class AbstractNodeDataChangedListener implements DataChangedListener {  @Override  public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {    //Configure prefix as plugin.    final String configKeyPrefix = changeData.getPluginDataId() + DefaultNodeConstants.JOIN_POINT;    this.onCommonChanged(configKeyPrefix, changed, eventType, PluginData::getName, PluginData.class);    LOG.debug("[DataChangedListener] PluginChanged {}", configKeyPrefix);  }}

private <T> void onCommonChanged(final String configKeyPrefix, final List<T> changedList,                                     final DataEventTypeEnum eventType, final Function<? super T, ? extends String> mapperToKey,                                     final Class<T> tClass) {        // Avoiding concurrent operations on list nodes        final ReentrantLock reentrantLock = listSaveLockMap.computeIfAbsent(configKeyPrefix, key -> new ReentrantLock());        try {            reentrantLock.lock();            // Current incoming plugin list            final List<String> changeNames = changedList.stream().map(mapperToKey).collect(Collectors.toList());            switch (eventType) {                // Delete Operation                case DELETE:                    // delete plugin.${pluginName}                    changedList.stream().map(mapperToKey).forEach(removeKey -> {                        delConfig(configKeyPrefix + removeKey);                    });                    // Remove the corresponding plugin name from plugin. list                    // The plugin.list records the currently enabled list                    delChangedData(configKeyPrefix, changeNames);                    break;                case REFRESH:                case MYSELF:                    // Overload logic                    // Get a list of all plugins in plugin.list                    final List<String> configDataNames = this.getConfigDataNames(configKeyPrefix);                    // Update each currently adjusted plug-in in turn                    changedList.forEach(changedData -> {                        // Publish Configuration                        publishConfig(configKeyPrefix + mapperToKey.apply(changedData), changedData);                    });                    // If there is more data in the currently stored list than what is currently being passed in, delete the excess data                    if (configDataNames != null && configDataNames.size() > changedList.size()) {                        // Kick out the currently loaded data                        configDataNames.removeAll(changeNames);                        // Delete cancelled data one by one                        configDataNames.forEach(this::delConfig);                    }                    // Update list data again                    publishConfig(configKeyPrefix + DefaultNodeConstants.LIST_STR, changeNames);                    break;                default:                    // Add or update                    changedList.forEach(changedData -> {                        publishConfig(configKeyPrefix + mapperToKey.apply(changedData), changedData);                    });                    // Update the newly added plugin                    putChangeData(configKeyPrefix, changeNames);                    break;            }        } catch (Exception e) {            LOG.error("AbstractNodeDataChangedListener onCommonMultiChanged error ", e);        } finally {            reentrantLock.unlock();        }    }

// Inheriting AbstractDataChangedInitpublic class ApolloDataChangedInit extends AbstractDataChangedInit {    // Apollo operation object    private final ApolloClient apolloClient;        public ApolloDataChangedInit(final ApolloClient apolloClient) {        this.apolloClient = apolloClient;    }        @Override    protected boolean notExist() {        // Check if nodes such as plugin, auth, meta, proxy.selector exist        // As long as one does not exist, it enters reload (these nodes will not be created, why check once?)        return Stream.of(ApolloPathConstants.PLUGIN_DATA_ID, ApolloPathConstants.AUTH_DATA_ID, ApolloPathConstants.META_DATA_ID, ApolloPathConstants.PROXY_SELECTOR_DATA_ID).allMatch(                this::dataIdNotExist);    }
    /**     * Data id not exist boolean.     *     * @param pluginDataId the plugin data id     * @return the boolean     */    private boolean dataIdNotExist(final String pluginDataId) {        return Objects.isNull(apolloClient.getItemValue(pluginDataId));    }}

@Servicepublic class SyncDataServiceImpl implements SyncDataService {    // Event Publishing    private final ApplicationEventPublisher eventPublisher;         /***     * Full data synchronization     * @param type the type     * @return     */     @Override     public boolean syncAll(final DataEventTypeEnum type) {         // Synchronize auth data         appAuthService.syncData();         // Synchronize plugin data         List<PluginData> pluginDataList = pluginService.listAll();         //Notify subscribers through the Spring publish/subscribe mechanism (publishing DataChangedEvent)         //Unified monitoring by DataChangedEventDispatcher         //DataChangedEvent comes with configuration grouping type, current operation type, and data         eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, type, pluginDataList));         // synchronizing selector         List<SelectorData> selectorDataList = selectorService.listAll();         eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, type, selectorDataList));         // Synchronization rules         List<RuleData> ruleDataList = ruleService.listAll();         eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.RULE, type, ruleDataList));         // Synchronization metadata         metaDataService.syncData();         // Synchronization Downstream List         discoveryService.syncData();         return true;     }    }

public class ApolloDataService extends AbstractNodeDataSyncService implements SyncDataService {    public ApolloDataService(final Config configService, final PluginDataSubscriber pluginDataSubscriber,                             final List<MetaDataSubscriber> metaDataSubscribers,                             final List<AuthDataSubscriber> authDataSubscribers,                             final List<ProxySelectorDataSubscriber> proxySelectorDataSubscribers,                             final List<DiscoveryUpstreamDataSubscriber> discoveryUpstreamDataSubscribers) {        // Configure the prefix for listening        super(new ChangeData(ApolloPathConstants.PLUGIN_DATA_ID,                        ApolloPathConstants.SELECTOR_DATA_ID,                        ApolloPathConstants.RULE_DATA_ID,                        ApolloPathConstants.AUTH_DATA_ID,                        ApolloPathConstants.META_DATA_ID,                        ApolloPathConstants.PROXY_SELECTOR_DATA_ID,                        ApolloPathConstants.DISCOVERY_DATA_ID),                pluginDataSubscriber, metaDataSubscribers, authDataSubscribers, proxySelectorDataSubscribers, discoveryUpstreamDataSubscribers);        this.configService = configService;        // Start listening        // Note: The Apollo method is only responsible for obtaining data from Apollo and adding it to the local cache, and does not handle listening        startWatch();        // Configure listening        apolloWatchPrefixes();    }}

private void apolloWatchPrefixes() {        // Defining Listeners        final ConfigChangeListener listener = changeEvent -> {            changeEvent.changedKeys().forEach(changeKey -> {                try {                    final ConfigChange configChange = changeEvent.getChange(changeKey);                    // Skip if not changed                    if (configChange == null) {                        LOG.error("apollo watchPrefixes error configChange is null {}", changeKey);                        return;                    }                    final String newValue = configChange.getNewValue();                    // skip last is "list"                    // If it is a Key at the end of the list, such as plugin.list, skip it because it is only a list that records the effectiveness and will not be cached locally                    final int lastListStrIndex = changeKey.length() - DefaultNodeConstants.LIST_STR.length();                    if (changeKey.lastIndexOf(DefaultNodeConstants.LIST_STR) == lastListStrIndex) {                        return;                    }                    // If it starts with plugin. => Process plugin data                    if (changeKey.indexOf(ApolloPathConstants.PLUGIN_DATA_ID) == 0) {                        // delete                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            // clear cache                            unCachePluginData(changeKey);                        } else {                            // update cache                            cachePluginData(newValue);                        }                        // If it starts with selector. => Process selector data                    } else if (changeKey.indexOf(ApolloPathConstants.SELECTOR_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheSelectorData(changeKey);                        } else {                            cacheSelectorData(newValue);                        }                        // If it starts with rule. => Process rule data                    } else if (changeKey.indexOf(ApolloPathConstants.RULE_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheRuleData(changeKey);                        } else {                            cacheRuleData(newValue);                        }                      // If it starts with auth. => Process auth data                    } else if (changeKey.indexOf(ApolloPathConstants.AUTH_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheAuthData(changeKey);                        } else {                            cacheAuthData(newValue);                        }                        // If it starts with meta. => Process meta data                    } else if (changeKey.indexOf(ApolloPathConstants.META_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheMetaData(changeKey);                        } else {                            cacheMetaData(newValue);                        }                        // If it starts with proxy.selector. => Process proxy.selector meta                    } else if (changeKey.indexOf(ApolloPathConstants.PROXY_SELECTOR_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheProxySelectorData(changeKey);                        } else {                            cacheProxySelectorData(newValue);                        }                        // If it starts with discovery. => Process discovery meta                    } else if (changeKey.indexOf(ApolloPathConstants.DISCOVERY_DATA_ID) == 0) {                        if (PropertyChangeType.DELETED.equals(configChange.getChangeType())) {                            unCacheDiscoveryUpstreamData(changeKey);                        } else {                            cacheDiscoveryUpstreamData(newValue);                        }                    }                } catch (Exception e) {                    LOG.error("apollo sync listener change key handler error", e);                }            });        };        watchConfigChangeListener = listener;        // Add listening        configService.addChangeListener(listener, Collections.emptySet(), ApolloPathConstants.pathKeySet());
    }