2 posts tagged with "http"

/**
 * Http sync data configuration for spring boot.
 */
@Configuration
@ConditionalOnClass(HttpSyncDataService.class)
@ConditionalOnProperty(prefix = "shenyu.sync.http", name = "url")
@EnableConfigurationProperties(value = HttpConfig.class)
public class HttpSyncDataConfiguration {

    private static final Logger LOGGER = LoggerFactory.getLogger(HttpSyncDataConfiguration.class);

    /**
     * Rest template.
     *
     * @param httpConfig the http config
     * @return the rest template
     */
    @Bean
    public RestTemplate restTemplate(final HttpConfig httpConfig) {
        OkHttp3ClientHttpRequestFactory factory = new OkHttp3ClientHttpRequestFactory();
        factory.setConnectTimeout(Objects.isNull(httpConfig.getConnectionTimeout()) ? (int) HttpConstants.CLIENT_POLLING_CONNECT_TIMEOUT : httpConfig.getConnectionTimeout());
        factory.setReadTimeout(Objects.isNull(httpConfig.getReadTimeout()) ? (int) HttpConstants.CLIENT_POLLING_READ_TIMEOUT : httpConfig.getReadTimeout());
        factory.setWriteTimeout(Objects.isNull(httpConfig.getWriteTimeout()) ? (int) HttpConstants.CLIENT_POLLING_WRITE_TIMEOUT : httpConfig.getWriteTimeout());
        return new RestTemplate(factory);
    }

    /**
     * AccessTokenManager.
     *
     * @param httpConfig   the http config.
     * @param restTemplate the rest template.
     * @return the access token manager.
     */
    @Bean
    public AccessTokenManager accessTokenManager(final HttpConfig httpConfig, final RestTemplate restTemplate) {
        return new AccessTokenManager(restTemplate, httpConfig);
    }

    /**
     * Http sync data service.
     *
     * @param httpConfig         the http config
     * @param pluginSubscriber   the plugin subscriber
     * @param restTemplate       the rest template
     * @param metaSubscribers    the meta subscribers
     * @param authSubscribers    the auth subscribers
     * @param accessTokenManager the access token manager
     * @return the sync data service
     */
    @Bean
    public SyncDataService httpSyncDataService(final ObjectProvider<HttpConfig> httpConfig,
                                               final ObjectProvider<PluginDataSubscriber> pluginSubscriber,
                                               final ObjectProvider<RestTemplate> restTemplate,
                                               final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers,
                                               final ObjectProvider<List<AuthDataSubscriber>> authSubscribers,
                                               final ObjectProvider<AccessTokenManager> accessTokenManager) {
        LOGGER.info("you use http long pull sync shenyu data");
        return new HttpSyncDataService(
                Objects.requireNonNull(httpConfig.getIfAvailable()),
                Objects.requireNonNull(pluginSubscriber.getIfAvailable()),
                Objects.requireNonNull(restTemplate.getIfAvailable()),
                metaSubscribers.getIfAvailable(Collections::emptyList),
                authSubscribers.getIfAvailable(Collections::emptyList),
                Objects.requireNonNull(accessTokenManager.getIfAvailable())
        );
    }
}

public class HttpSyncDataService implements SyncDataService {

    // omitted attribute field ......

    public HttpSyncDataService(final HttpConfig httpConfig,
                               final PluginDataSubscriber pluginDataSubscriber,
                               final RestTemplate restTemplate,
                               final List<MetaDataSubscriber> metaDataSubscribers,
                               final List<AuthDataSubscriber> authDataSubscribers,
                               final AccessTokenManager accessTokenManager) {
          // 1. accessTokenManager
          this.accessTokenManager = accessTokenManager;
          // 2. create data refresh factory
          this.factory = new DataRefreshFactory(pluginDataSubscriber, metaDataSubscribers, authDataSubscribers);
          // 3. shenyu-admin url
          this.serverList = Lists.newArrayList(Splitter.on(",").split(httpConfig.getUrl()));
          // 4. restTemplate
          this.restTemplate = restTemplate;
          // 5. start a long polling task
          this.start();
    }

    //......
}

public class HttpSyncDataService implements SyncDataService {

    // ......

    private void start() {
        // It could be initialized multiple times, so you need to control that.
        if (RUNNING.compareAndSet(false, true)) {
            // fetch all group configs.
            // Initial startup, get full data
            this.fetchGroupConfig(ConfigGroupEnum.values());
            // one backend service, one thread
            int threadSize = serverList.size();
            // ThreadPoolExecutor
            this.executor = new ThreadPoolExecutor(threadSize, threadSize, 60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(),
                    ShenyuThreadFactory.create("http-long-polling", true));
            // start long polling, each server creates a thread to listen for changes.
            this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
        } else {
            LOG.info("shenyu http long polling was started, executor=[{}]", executor);
        }
    }

    // ......
}

public enum ConfigGroupEnum {
    APP_AUTH, // app auth data
    PLUGIN, // plugin data
    RULE, // rule data
    SELECTOR, // selector data
    META_DATA; // meta data
}

public class HttpSyncDataService implements SyncDataService {

    // ......

    private void fetchGroupConfig(final ConfigGroupEnum... groups) throws ShenyuException {
        // It is possible that admins are clustered, and here requests are made to each admin by means of a loop.
        for (int index = 0; index < this.serverList.size(); index++) {
            String server = serverList.get(index);
            try {
                // do execute
                this.doFetchGroupConfig(server, groups);
                // If you have a success, you are successful and can exit the loop
                break;
            } catch (ShenyuException e) {
                // An exception occurs, try executing the next
                // The last one also failed to execute, throwing an exception
                // no available server, throw exception.
                if (index >= serverList.size() - 1) {
                    throw e;
                }
                LOG.warn("fetch config fail, try another one: {}", serverList.get(index + 1));
            }
        }
    }

    // ......
}

public class HttpSyncDataService implements SyncDataService {

    // ......

    // Launch a request to the admin backend management system to get all synchronized data
    private void doFetchGroupConfig(final String server, final ConfigGroupEnum... groups) {
        // 1. build request parameters, all grouped enumeration types
        StringBuilder params = new StringBuilder();
        for (ConfigGroupEnum groupKey : groups) {
            params.append("groupKeys").append("=").append(groupKey.name()).append("&");
        }
        // admin url:  /configs/fetch
        String url = server + Constants.SHENYU_ADMIN_PATH_CONFIGS_FETCH + "?" + StringUtils.removeEnd(params.toString(), "&");
        LOG.info("request configs: [{}]", url);
        String json;
        try {
            HttpHeaders headers = new HttpHeaders();
            // set accessToken
            headers.set(Constants.X_ACCESS_TOKEN, this.accessTokenManager.getAccessToken());
            HttpEntity<String> httpEntity = new HttpEntity<>(headers);
            // 2. get a request for change data
            json = this.restTemplate.exchange(url, HttpMethod.GET, httpEntity, String.class).getBody();
        } catch (RestClientException e) {
            String message = String.format("fetch config fail from server[%s], %s", url, e.getMessage());
            LOG.warn(message);
            throw new ShenyuException(message, e);
        }
        // update local cache
        // 3. Update data in gateway memory
        boolean updated = this.updateCacheWithJson(json);
        if (updated) {
            LOG.debug("get latest configs: [{}]", json);
            return;
        }
        // not updated. it is likely that the current config server has not been updated yet. wait a moment.
        LOG.info("The config of the server[{}] has not been updated or is out of date. Wait for 30s to listen for changes again.", server);
        // No data update on the server side, just wait 30s
        ThreadUtils.sleep(TimeUnit.SECONDS, 30);
    }

    // ......
}

public class HttpSyncDataService implements SyncDataService {

    // ......

    private boolean updateCacheWithJson(final String json) {
        // Using GSON for deserialization
        JsonObject jsonObject = GSON.fromJson(json, JsonObject.class);
        // if the config cache will be updated?
        return factory.executor(jsonObject.getAsJsonObject("data"));
    }

    // ......
}

public final class DataRefreshFactory {
    
    // ......
    
    public boolean executor(final JsonObject data) {
        // update data
        List<Boolean> result = ENUM_MAP.values().parallelStream()
                .map(dataRefresh -> dataRefresh.refresh(data))
                .collect(Collectors.toList());
        // one of the data types is updated, which means that the operation has been updated.
        return result.stream().anyMatch(Boolean.TRUE::equals);
    }
    
    // ......
}

public abstract class AbstractDataRefresh<T> implements DataRefresh {

    // ......

    @Override
    public Boolean refresh(final JsonObject data) {
        // convert data
        JsonObject jsonObject = convert(data);
        if (Objects.isNull(jsonObject)) {
            return false;
        }

        boolean updated = false;
        // get data
        ConfigData<T> result = fromJson(jsonObject);
        // does it need to be updated
        if (this.updateCacheIfNeed(result)) {
            updated = true;
            // real update logic, data refresh operation
            refresh(result.getData());
        }

        return updated;
    }

    // ......
}

public abstract class AbstractDataRefresh<T> implements DataRefresh {

    // ......

    // result is data
    protected abstract boolean updateCacheIfNeed(ConfigData<T> result);

    // newVal is the latest value obtained
    // What kind of data type is groupEnum
    protected boolean updateCacheIfNeed(final ConfigData<T> newVal, final ConfigGroupEnum groupEnum) {
        // If it is the first time, then it is put directly into the cache and returns true, indicating that the update was made this time
        if (GROUP_CACHE.putIfAbsent(groupEnum, newVal) == null) {
            return true;
        }
        ResultHolder holder = new ResultHolder(false);
        GROUP_CACHE.merge(groupEnum, newVal, (oldVal, value) -> {
            // md5 value is the same, no need to update
            if (StringUtils.equals(oldVal.getMd5(), newVal.getMd5())) {
                LOG.info("Get the same config, the [{}] config cache will not be updated, md5:{}", groupEnum, oldVal.getMd5());
                return oldVal;
            }

            // The current cached data has been modified for a longer period than the new data and does not need to be updated.
            // must compare the last update time
            if (oldVal.getLastModifyTime() >= newVal.getLastModifyTime()) {
                LOG.info("Last update time earlier than the current configuration, the [{}] config cache will not be updated", groupEnum);
                return oldVal;
            }
            LOG.info("update {} config: {}", groupEnum, newVal);
            holder.result = true;
            return newVal;
        });
        return holder.result;
    }

    // ......
}

public class HttpSyncDataService implements SyncDataService {

    // ......

    private void start() {
        // It could be initialized multiple times, so you need to control that.
        if (RUNNING.compareAndSet(false, true)) {
            // fetch all group configs.
            // Initial startup, get full data
            this.fetchGroupConfig(ConfigGroupEnum.values());
            // one backend service, one thread
            int threadSize = serverList.size();
            // ThreadPoolExecutor
            this.executor = new ThreadPoolExecutor(threadSize, threadSize, 60L, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(),
                    ShenyuThreadFactory.create("http-long-polling", true));
            // start long polling, each server creates a thread to listen for changes.
            this.serverList.forEach(server -> this.executor.execute(new HttpLongPollingTask(server)));
        } else {
            LOG.info("shenyu http long polling was started, executor=[{}]", executor);
        }
    }

    // ......
}

class HttpLongPollingTask implements Runnable {

    private final String server;

    HttpLongPollingTask(final String server) {
        this.server = server;
    }

    @Override
    public void run() {
        // long polling
        while (RUNNING.get()) {
            // Default retry 3 times
            int retryTimes = 3;
            for (int time = 1; time <= retryTimes; time++) {
                try {
                    doLongPolling(server);
                } catch (Exception e) {
                    if (time < retryTimes) {
                        LOG.warn("Long polling failed, tried {} times, {} times left, will be suspended for a while! {}",
                                time, retryTimes - time, e.getMessage());
                        // long polling failed, wait 5s and continue
                        ThreadUtils.sleep(TimeUnit.SECONDS, 5);
                        continue;
                    }
                    // print error, then suspended for a while.
                    LOG.error("Long polling failed, try again after 5 minutes!", e);
                    // 3 次都失败了，等 5 分钟再试
                    ThreadUtils.sleep(TimeUnit.MINUTES, 5);
                }
            }
        }
        LOG.warn("Stop http long polling.");
    }
}

public class HttpSyncDataService implements SyncDataService {
    private void doLongPolling(final String server) {
        // build request params: md5 and lastModifyTime
        MultiValueMap<String, String> params = new LinkedMultiValueMap<>(8);
        for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
            ConfigData<?> cacheConfig = factory.cacheConfigData(group);
            if (cacheConfig != null) {
                String value = String.join(",", cacheConfig.getMd5(), String.valueOf(cacheConfig.getLastModifyTime()));
                params.put(group.name(), Lists.newArrayList(value));
            }
        }
        // build request head and body
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
        // set accessToken
        headers.set(Constants.X_ACCESS_TOKEN, this.accessTokenManager.getAccessToken());
        HttpEntity<MultiValueMap<String, String>> httpEntity = new HttpEntity<>(params, headers);
        String listenerUrl = server + Constants.SHENYU_ADMIN_PATH_CONFIGS_LISTENER;

        JsonArray groupJson;
        //Initiate a request to admin to determine if the group data has changed
        //Here it just determines whether a group has changed or not
        try {
            String json = this.restTemplate.postForEntity(listenerUrl, httpEntity, String.class).getBody();
            LOG.info("listener result: [{}]", json);
            JsonObject responseFromServer = GsonUtils.getGson().fromJson(json, JsonObject.class);
            groupJson = responseFromServer.getAsJsonArray("data");
        } catch (RestClientException e) {
            String message = String.format("listener configs fail, server:[%s], %s", server, e.getMessage());
            throw new ShenyuException(message, e);
        }
        // Depending on the group where the change occurred, go back and get the data
        /**
         * The official website explains here.
         * After the gateway receives the response message, it only knows which Group has made the configuration change, and it still needs to request the configuration data of that Group again.
         * There may be a question here: why not write out the changed data directly?
         * We also discussed this issue in depth during development, because the http long polling mechanism can only guarantee quasi-real time, if the processing at the gateway layer is not timely, * or the administrator frequently updates the configuration, it is very difficult to get the information from the gateway layer.
         * If it is not processed in time at the gateway level, or if the administrator updates the configuration frequently, it is very likely to miss the push of a configuration change, so for security reasons, we only inform a group that the information has changed.
         *For security reasons, we only notify a group of changes.
         * Personal understanding.
         * If the change data is written out directly, when the administrator frequently updates the configuration, the first update will remove the client from the blocking queue and return the response information to the gateway.
         * If a second update is made at this time, the current client is not in the blocking queue, so this time the change is missed.
         * The same is true for untimely processing by the gateway layer.
         * This is a long polling, one gateway one synchronization thread, there may be time consuming process.
         * If the admin has data changes, the current gateway client is not in the blocking queue and will not get the data.
         */
        if (groupJson != null) {
            // fetch group configuration async.
            ConfigGroupEnum[] changedGroups = GSON.fromJson(groupJson, ConfigGroupEnum[].class);
            if (ArrayUtils.isNotEmpty(changedGroups)) {
                log.info("Group config changed: {}", Arrays.toString(changedGroups));
                // Proactively get the changed data from admin, depending on the grouping, and take the data in full
                this.doFetchGroupConfig(server, changedGroups);
            }
        }
        if (Objects.nonNull(groupJson) && groupJson.size() > 0) {
            // fetch group configuration async.
            ConfigGroupEnum[] changedGroups = GsonUtils.getGson().fromJson(groupJson, ConfigGroupEnum[].class);
            LOG.info("Group config changed: {}", Arrays.toString(changedGroups));
            // Proactively get the changed data from admin, depending on the grouping, and take the data in full
            this.doFetchGroupConfig(server, changedGroups);
        }
    }
}

/**
 * Data synchronization configuration class
 * Conditional assembly via springboot
 * The type Data sync configuration.
 */
@Configuration
public class DataSyncConfiguration {

    /**
     * http long polling.
     */
    @Configuration
    @ConditionalOnProperty(name = "shenyu.sync.http.enabled", havingValue = "true")
    @EnableConfigurationProperties(HttpSyncProperties.class)
    static class HttpLongPollingListener {

        @Bean
        @ConditionalOnMissingBean(HttpLongPollingDataChangedListener.class)
        public HttpLongPollingDataChangedListener httpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {
            return new HttpLongPollingDataChangedListener(httpSyncProperties);
        }
    }
}

    public HttpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {
        // default client (here is the gateway) 1024
        this.clients = new ArrayBlockingQueue<>(1024);
        // create thread pool
        // ScheduledThreadPoolExecutor can perform delayed tasks, periodic tasks, and normal tasks
        this.scheduler = new ScheduledThreadPoolExecutor(1,
                ShenyuThreadFactory.create("long-polling", true));
        // http sync properties
        this.httpSyncProperties = httpSyncProperties;
    }

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {

    // ......

    /**
     * is called in the afterPropertiesSet() method of the InitializingBean interface, which is executed during the initialization of the bean
     */
    @Override
    protected void afterInitialize() {
        long syncInterval = httpSyncProperties.getRefreshInterval().toMillis();
        // Periodically check the data for changes and update the cache

        // Execution cycle task: Update data in memory (CACHE) is executed every 5 minutes and starts after 5 minutes
        // Prevent the admin from starting up first for a while and then generating data; then the gateway doesn't get the full amount of data when it first connects
        scheduler.scheduleWithFixedDelay(() -> {
            LOG.info("http sync strategy refresh config start.");
            try {
                // Read data from database to local cache (in this case, memory)
                this.refreshLocalCache();
                LOG.info("http sync strategy refresh config success.");
            } catch (Exception e) {
                LOG.error("http sync strategy refresh config error!", e);
            }
        }, syncInterval, syncInterval, TimeUnit.MILLISECONDS);
        LOG.info("http sync strategy refresh interval: {}ms", syncInterval);
    }

    // ......
}

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {

    // ......

    // Read data from database to local cache (in this case, memory)
    private void refreshLocalCache() {
        //update app auth data
        this.updateAppAuthCache();
        //update plugin data
        this.updatePluginCache();
        //update rule data
        this.updateRuleCache();
        //update selector data
        this.updateSelectorCache();
        //update meta data
        this.updateMetaDataCache();
    }

    // ......
}

    /**
     * Update rule cache.
     */
    protected void updateRuleCache() {
        this.updateCache(ConfigGroupEnum.RULE, ruleService.listAll());
    }

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {

    // ......

    // cache Map
    protected static final ConcurrentMap<String, ConfigDataCache> CACHE = new ConcurrentHashMap<>();

    /**
     * if md5 is not the same as the original, then update lcoal cache.
     * @param group ConfigGroupEnum
     * @param <T> the type of class
     * @param data the new config data
     */
    protected <T> void updateCache(final ConfigGroupEnum group, final List<T> data) {
        // data serialization
        String json = GsonUtils.getInstance().toJson(data);
        // pass in md5 value and modification time
        ConfigDataCache newVal = new ConfigDataCache(group.name(), json, Md5Utils.md5(json), System.currentTimeMillis());
        // update group data
        ConfigDataCache oldVal = CACHE.put(newVal.getGroup(), newVal);
        log.info("update config cache[{}], old: {}, updated: {}", group, oldVal, newVal);
    }

    // ......
}

   
/**
 * This Controller only when HttpLongPollingDataChangedListener exist, will take effect.
 */
@ConditionalOnBean(HttpLongPollingDataChangedListener.class)
@RestController
@RequestMapping("/configs")
public class ConfigController {

    private final HttpLongPollingDataChangedListener longPollingListener;

    public ConfigController(final HttpLongPollingDataChangedListener longPollingListener) {
        this.longPollingListener = longPollingListener;
    }
    
    // Omit other logic

    /**
     * Listener.
     * Listen for data changes and perform long polling
     * @param request  the request
     * @param response the response
     */
    @PostMapping(value = "/listener")
    public void listener(final HttpServletRequest request, final HttpServletResponse response) {
        longPollingListener.doLongPolling(request, response);
    }

}

public class HttpLongPollingDataChangedListener extends AbstractDataChangedListener {

    // ......

    /**
     * Execute long polling: If there is a data change, it will be responded to the client (here is the gateway side) immediately.
     * Otherwise, the client will otherwise remain blocked until there is a data change or a timeout.
     * @param request
     * @param response
     */
    public void doLongPolling(final HttpServletRequest request, final HttpServletResponse response) {
        // compare group md5
        // Compare the md5, determine whether the data of the gateway and the data of the admin side are consistent, and get the data group that has changed
        List<ConfigGroupEnum> changedGroup = compareChangedGroup(request);
        String clientIp = getRemoteIp(request);
        // response immediately.
        // Immediate response to the gateway if there is changed data
        if (CollectionUtils.isNotEmpty(changedGroup)) {
            this.generateResponse(response, changedGroup);
            Log.info("send response with the changed group, ip={}, group={}", clientIp, changedGroup);
            return;
        }

        // No change, then the client (in this case the gateway) is put into the blocking queue
        // listen for configuration changed.
        final AsyncContext asyncContext = request.startAsync();
        // AsyncContext.settimeout() does not timeout properly, so you have to control it yourself
        asyncContext.setTimeout(0L);
        // block client's thread.
        scheduler.execute(new LongPollingClient(asyncContext, clientIp, HttpConstants.SERVER_MAX_HOLD_TIMEOUT));
    }
}

 /**
     * Determine if the group data has changed
     * @param request
     * @return
     */
    private List<ConfigGroupEnum> compareChangedGroup(final HttpServletRequest request) {
        List<ConfigGroupEnum> changedGroup = new ArrayList<>(ConfigGroupEnum.values().length);
        for (ConfigGroupEnum group : ConfigGroupEnum.values()) {
            // The md5 value and lastModifyTime of the data on the gateway side
            String[] params = StringUtils.split(request.getParameter(group.name()), ',');
            if (params == null || params.length != 2) {
                throw new ShenyuException("group param invalid:" + request.getParameter(group.name()));
            }
            String clientMd5 = params[0];
            long clientModifyTime = NumberUtils.toLong(params[1]);
            ConfigDataCache serverCache = CACHE.get(group.name());
            // do check. determine if the group data has changed
            if (this.checkCacheDelayAndUpdate(serverCache, clientMd5, clientModifyTime)) {
                changedGroup.add(group);
            }
        }
        return changedGroup;
    }

class LongPollingClient implements Runnable {
      // omitted other logic
    
        @Override
        public void run() {
            try {
                // Removal after 60 seconds and response to the client
                this.asyncTimeoutFuture = scheduler.schedule(() -> {
                    clients.remove(LongPollingClient.this);
                    List<ConfigGroupEnum> changedGroups = compareChangedGroup((HttpServletRequest) asyncContext.getRequest());
                    sendResponse(changedGroups);
                }, timeoutTime, TimeUnit.MILLISECONDS);

                // Add to blocking queue
                clients.add(this);

            } catch (Exception ex) {
                log.error("add long polling client error", ex);
            }
        }

        /**
         * Send response.
         *
         * @param changedGroups the changed groups
         */
        void sendResponse(final List<ConfigGroupEnum> changedGroups) {
            // cancel scheduler
            if (null != asyncTimeoutFuture) {
                asyncTimeoutFuture.cancel(false);
            }
            // Groups responding to changes
            generateResponse((HttpServletResponse) asyncContext.getResponse(), changedGroups);
            asyncContext.complete();
        }
    }

@ConditionalOnBean(HttpLongPollingDataChangedListener.class)
@RestController
@RequestMapping("/configs")
public class ConfigController {

    private final HttpLongPollingDataChangedListener longPollingListener;

    public ConfigController(final HttpLongPollingDataChangedListener longPollingListener) {
        this.longPollingListener = longPollingListener;
    }

    /**
     * Fetch configs shenyu result.
     * @param groupKeys the group keys
     * @return the shenyu result
     */
    @GetMapping("/fetch")
    public ShenyuAdminResult fetchConfigs(@NotNull final String[] groupKeys) {
        Map<String, ConfigData<?>> result = Maps.newHashMap();
        for (String groupKey : groupKeys) {
            ConfigData<?> data = longPollingListener.fetchConfig(ConfigGroupEnum.valueOf(groupKey));
            result.put(groupKey, data);
        }
        return ShenyuAdminResult.success(ShenyuResultMessage.SUCCESS, result);
    }
    
  // Other interfaces are omitted

}

public abstract class AbstractDataChangedListener implements DataChangedListener, InitializingBean {
    
    // ......
    
    /**
     * fetch configuration from cache.
     * @param groupKey the group key
     * @return the configuration data
     */
    public ConfigData<?> fetchConfig(final ConfigGroupEnum groupKey) {
        // get data from CACHE
        ConfigDataCache config = CACHE.get(groupKey.name());
        switch (groupKey) {
            case APP_AUTH: // app auth data
                return buildConfigData(config, AppAuthData.class);
            case PLUGIN: // plugin data
                return buildConfigData(config, PluginData.class);
            case RULE:   // rule data
                return buildConfigData(config, RuleData.class);
            case SELECTOR:  // selector data
                return buildConfigData(config, SelectorData.class);
            case META_DATA: // meta data 
                return buildConfigData(config, MetaData.class);
            default:  // other data type, throw exception
                throw new IllegalStateException("Unexpected groupKey: " + groupKey);
        }
    }
    
    // ......
}

/**
 * Event forwarders, which forward the changed events to each ConfigEventListener.
 * Data change event distributor: synchronize the change data to ShenYu gateway when there is a data change in admin side
 * Data changes rely on Spring's event-listening mechanism: ApplicationEventPublisher --> ApplicationEvent --> ApplicationListener
 *
 */
@Component
public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {

  // other logic omitted

    /**
     * Call this method when there are data changes
     * @param event
     */
    @Override
    @SuppressWarnings("unchecked")
    public void onApplicationEvent(final DataChangedEvent event) {
        // Iterate through the data change listeners (it's generally good to use a kind of data synchronization)
        for (DataChangedListener listener : listeners) {
            // What kind of data has changed
            switch (event.getGroupKey()) {
                case APP_AUTH: // app auth data
                    listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
                    break;
                case PLUGIN:  // plugin data
                    listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
                    break;
                case RULE:    // rule data
                    listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
                    break;
                case SELECTOR:   // selector data
                    listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                    // pull and save API document on seletor changed
                    applicationContext.getBean(LoadServiceDocEntry.class).loadDocOnSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
                    break;
                case META_DATA:  // meta data
                    listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
                    break;
                default:  // other data type, throw exception
                    throw new IllegalStateException("Unexpected value: " + event.getGroupKey());
            }
        }
    }
}

    /**
     * In the operation of the admin, there is an update of the plugin occurred
     * @param changed   the changed
     * @param eventType the event type
     */
    @Override
    public void onPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
        if (CollectionUtils.isEmpty(changed)) {
            return;
        }
        // update CACHE
        this.updatePluginCache();
        // execute change task
        this.afterPluginChanged(changed, eventType);
    }

    @Override
    protected void afterPluginChanged(final List<PluginData> changed, final DataEventTypeEnum eventType) {
        // execute by thread pool
        scheduler.execute(new DataChangeTask(ConfigGroupEnum.PLUGIN));
    }

class DataChangeTask implements Runnable {
		//other logic omitted
  
        @Override
        public void run() {
            // If the client in the blocking queue exceeds the given value of 100, it is executed in batches
            if (clients.size() > httpSyncProperties.getNotifyBatchSize()) {
                List<LongPollingClient> targetClients = new ArrayList<>(clients.size());
                clients.drainTo(targetClients);
                List<List<LongPollingClient>> partitionClients = Lists.partition(targetClients, httpSyncProperties.getNotifyBatchSize());
               // batch execution
                partitionClients.forEach(item -> scheduler.execute(() -> doRun(item)));
            } else {
                // execute task
                doRun(clients);
            }
        }

        private void doRun(final Collection<LongPollingClient> clients) {
            // Notify all clients that a data change has occurred
            for (Iterator<LongPollingClient> iter = clients.iterator(); iter.hasNext();) {
                LongPollingClient client = iter.next();
                iter.remove();
                // send response to client
                client.sendResponse(Collections.singletonList(groupKey));
                Log.info("send response with the changed group,ip={}, group={}, changeTime={}", client.ip, groupKey, changeTime);
            }
        }
    }