The Soul gateway learns WebSocket data synchronization parsing
Authors: Fan Jinpeng, Zhu Ming
1. Previous episodes
The soul-admin interface will be called after the HTTP user service system is connected to the Soul gateway. The soul-admin interface will register all the interface information that needs the gateway proxy with the soul-admin. The soul-admin interface is synchronized to the Soul-bootstrap gateway. The soul-bootstrap interface is synchronized to the soul-bootstrap gateway.
4.HTTP user access Soul call /soul-client/ springMVC-register interface logic analysis
2. Soul-admin synchronizes data with soul-bootstrap
In order to verify the data synchronization process, it is not necessary to start the business system. You can just start the soul-admin and soul-bootstrap systems. You can open or close the plug-in on the page and see how the process works.
Data synchronization policy official website dromara.org/zh-cn/docs/…
2.1 Starting two Systems
All are started by default without modifying any configuration files.
2.2 Page Operation Search interface
Let’s start Divide, F12, and see which interface the foreground calls soul-admin.
Can see the foreground to background sends a PUT request: http://localhost:9095/plugin/5
2.3 Background Interfaces
Search for this interface in the project
// PluginController.java
@RestController
@RequestMapping("/plugin")
public class PluginController {.../**
* update plugin.
*
* @param id primary key.
* @param pluginDTO plugin.
* @return {@linkplain SoulAdminResult}
*/
@PutMapping("/{id}")
public SoulAdminResult updatePlugin(@PathVariable("id") final String id, @RequestBody final PluginDTO pluginDTO) {
Objects.requireNonNull(pluginDTO);
pluginDTO.setId(id);
final String result = pluginService.createOrUpdate(pluginDTO);
if (StringUtils.isNoneBlank(result)) {
return SoulAdminResult.error(result);
}
returnSoulAdminResult.success(SoulResultMessage.UPDATE_SUCCESS); }... }Copy the codeGo into the implementation class
// PluginServiceImpl.java
/**
* create or update plugin.
*
* @param pluginDTO {@linkplain PluginDTO}
* @return rows
*/
@Override
@Transactional(rollbackFor = Exception.class)
public String createOrUpdate(final PluginDTO pluginDTO) {
final String msg = checkData(pluginDTO);
if (StringUtils.isNoneBlank(msg)) {
return msg;
}
PluginDO pluginDO = PluginDO.buildPluginDO(pluginDTO);
DataEventTypeEnum eventType = DataEventTypeEnum.CREATE;
if (StringUtils.isBlank(pluginDTO.getId())) {
pluginMapper.insertSelective(pluginDO);
} else {
eventType = DataEventTypeEnum.UPDATE;
pluginMapper.updateSelective(pluginDO);
}
// publish change event.
eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.PLUGIN, eventType,
Collections.singletonList(PluginTransfer.INSTANCE.mapToData(pluginDO))));
return StringUtils.EMPTY;
}
Copy the codeAs you can see here, the first half is all about manipulating the database, persisting relevant information; The second half is publishing an event.
2.4 Release Event
The events published here are wrapped in a DataChangedEvent layer, which contains an enumeration of various types:
/**
* configuration group.
*
* @author huangxiaofeng
*/
public enum ConfigGroupEnum { APP_AUTH, PLUGIN, RULE, SELECTOR, META_DATA; . }Copy the codeLooking at these types, if you remember from Chapter 4, the types of events sent were SELECTOR and RULE, and now PLUGIN, but they’re different types, so let’s move on to the logic. Let’s move on.
The other eventType is also an enumeration, with DELETE, CREATE, UPDATE, REFRESH, and MYSELF, in this case UPDATE.
/**
* The enum Data event type.
*
* @author xiaoyu
*/
public enum DataEventTypeEnum {
/** * delete event. */
DELETE,
/** * insert event. */
CREATE,
/** * update event. */
UPDATE,
/** * REFRESH data event type enum. */
REFRESH,
/** * Myself data event type enum. */MYSELF; . }Copy the code2.5 Listening Events
Find the code that listens for events:
// DataChangedEventDispatcher.java
@Component
public class DataChangedEventDispatcher implements ApplicationListener<DataChangedEvent>, InitializingBean {
private ApplicationContext applicationContext;
private List<DataChangedListener> listeners;
public DataChangedEventDispatcher(final ApplicationContext applicationContext) {
this.applicationContext = applicationContext;
}
@Override
@SuppressWarnings("unchecked")
public void onApplicationEvent(final DataChangedEvent event) {
for (DataChangedListener listener : listeners) {
switch (event.getGroupKey()) {
case APP_AUTH:
listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
break;
case PLUGIN:
listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
break;
case RULE:
listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
break;
case SELECTOR:
listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
break;
case META_DATA:
listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
break;
default:
throw new IllegalStateException("Unexpected value: "+ event.getGroupKey()); }}}@Override
public void afterPropertiesSet(a) {
Collection<DataChangedListener> listenerBeans = applicationContext.getBeansOfType(DataChangedListener.class).values();
this.listeners = Collections.unmodifiableList(newArrayList<>(listenerBeans)); }}Copy the code2.5.1 Listener Injection
Can see DataChangedEventDispatcher implements InitializingBean interface, overwrite the afterPropertiesSet method, and use the @ Component, when the Spring starts, The override method is called after the container has finished loading. AfterPropertiesSet fetches all beans of type DataChangedListener and places them on class properties listeners.
The question is, when are these listeners injected into the container?
Take a look at the DataChangedListener interface definition:
/**
* Event listener, used to send notification of event changes,
* used to support HTTP, websocket, zookeeper and other event notifications.
*
* @author huangxiaofeng
* @author xiaoyu
*/
public interface DataChangedListener {
/**
* invoke this method when AppAuth was received.
*
* @param changed the changed
* @param eventType the event type
*/
default void onAppAuthChanged(List<AppAuthData> changed, DataEventTypeEnum eventType) {}/**
* invoke this method when Plugin was received.
*
* @param changed the changed
* @param eventType the event type
*/
default void onPluginChanged(List<PluginData> changed, DataEventTypeEnum eventType) {}/**
* invoke this method when Selector was received.
*
* @param changed the changed
* @param eventType the event type
*/
default void onSelectorChanged(List<SelectorData> changed, DataEventTypeEnum eventType) {}/**
* On meta data changed.
*
* @param changed the changed
* @param eventType the event type
*/
default void onMetaDataChanged(List<MetaData> changed, DataEventTypeEnum eventType) {}/**
* invoke this method when Rule was received.
*
* @param changed the changed
* @param eventType the event type
*/
default void onRuleChanged(List<RuleData> changed, DataEventTypeEnum eventType) {}}Copy the codeAs you can see, there are five methods defined in the interface, each for handling changes in appAuth, Plugin, Selector, metaData, and rule data.
Its inheritance relationship:
Because the default is to use the websocket, here is the listener of the corresponding WebsocketDataChangedListener, Alt + F7, to search the class instantiation place, is the configuration class is as follows:
// DataSyncConfiguration.java
@Configuration
public class DataSyncConfiguration {
/** * http long polling. */
@Configuration
@ConditionalOnProperty(name = "soul.sync.http.enabled", havingValue = "true")
@EnableConfigurationProperties(HttpSyncProperties.class)
static class HttpLongPollingListener {
@Bean
@ConditionalOnMissingBean(HttpLongPollingDataChangedListener.class)
public HttpLongPollingDataChangedListener httpLongPollingDataChangedListener(final HttpSyncProperties httpSyncProperties) {
return newHttpLongPollingDataChangedListener(httpSyncProperties); }}/** * The type Zookeeper listener. */
@Configuration
@ConditionalOnProperty(prefix = "soul.sync.zookeeper", name = "url")
@Import(ZookeeperConfiguration.class)
static class ZookeeperListener {
@Bean
@ConditionalOnMissingBean(ZookeeperDataChangedListener.class)
public DataChangedListener zookeeperDataChangedListener(final ZkClient zkClient) {
return new ZookeeperDataChangedListener(zkClient);
}
@Bean
@ConditionalOnMissingBean(ZookeeperDataInit.class)
public ZookeeperDataInit zookeeperDataInit(final ZkClient zkClient, final SyncDataService syncDataService) {
return newZookeeperDataInit(zkClient, syncDataService); }}/** * The type Nacos listener. */
@Configuration
@ConditionalOnProperty(prefix = "soul.sync.nacos", name = "url")
@Import(NacosConfiguration.class)
static class NacosListener {
@Bean
@ConditionalOnMissingBean(NacosDataChangedListener.class)
public DataChangedListener nacosDataChangedListener(final ConfigService configService) {
return newNacosDataChangedListener(configService); }}/** * The WebsocketListener(default strategy). */
@Configuration
@ConditionalOnProperty(name = "soul.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)
@EnableConfigurationProperties(WebsocketSyncProperties.class)
static class WebsocketListener {
@Bean
@ConditionalOnMissingBean(WebsocketDataChangedListener.class)
public DataChangedListener websocketDataChangedListener(a) {
return new WebsocketDataChangedListener();
}
@Bean
@ConditionalOnMissingBean(WebsocketCollector.class)
public WebsocketCollector websocketCollector(a) {
return new WebsocketCollector();
}
@Bean
@ConditionalOnMissingBean(ServerEndpointExporter.class)
public ServerEndpointExporter serverEndpointExporter(a) {
return newServerEndpointExporter(); }}}Copy the codeThere are four data synchronization strategies: HTTP long polling, ZooKeeper, NacOS, and WebSocket (the default).
See the websocket annotation @ ConditionalOnProperty (name = “soul. Sync. Websocket. Enabled”, havingValue = “true”, MatchIfMissing = true), go to the configuration file to find the following configuration:
soul:
sync:
websocket:
enabled: true
Copy the codeThis is where the truth comes out.
If you do not want to use webSocket’s default synchronization policy, write the corresponding configuration in the configuration file.
2.5.2 Listening Event Processing logic
In order to prevent you from looking back, I will post the processing logic code here:
// DataChangedEventDispatcher.java
@Override
@SuppressWarnings("unchecked")
public void onApplicationEvent(final DataChangedEvent event) {
for (DataChangedListener listener : listeners) {
switch (event.getGroupKey()) {
case APP_AUTH:
listener.onAppAuthChanged((List<AppAuthData>) event.getSource(), event.getEventType());
break;
case PLUGIN:
listener.onPluginChanged((List<PluginData>) event.getSource(), event.getEventType());
break;
case RULE:
listener.onRuleChanged((List<RuleData>) event.getSource(), event.getEventType());
break;
case SELECTOR:
listener.onSelectorChanged((List<SelectorData>) event.getSource(), event.getEventType());
break;
case META_DATA:
listener.onMetaDataChanged((List<MetaData>) event.getSource(), event.getEventType());
break;
default:
throw new IllegalStateException("Unexpected value: "+ event.getGroupKey()); }}}Copy the codeFor the current websocket, there will only be one listener. For the other multiple cases, I don’t know when it will appear. I doubt it.
The logic varies according to the type of published event, which corresponds to the methods defined in the DataChangedListener interface.
The listener is WebsocketDataChangedListener instance, will into the corresponding method to the class:
// WebsocketDataChangedListener.java
public class WebsocketDataChangedListener implements DataChangedListener {
@Override
public void onPluginChanged(final List<PluginData> pluginDataList, final DataEventTypeEnum eventType) {
WebsocketData<PluginData> websocketData =
new WebsocketData<>(ConfigGroupEnum.PLUGIN.name(), eventType.name(), pluginDataList);
WebsocketCollector.send(GsonUtils.getInstance().toJson(websocketData), eventType);
}
@Override
public void onSelectorChanged(final List<SelectorData> selectorDataList, final DataEventTypeEnum eventType) {
WebsocketData<SelectorData> websocketData =
new WebsocketData<>(ConfigGroupEnum.SELECTOR.name(), eventType.name(), selectorDataList);
WebsocketCollector.send(GsonUtils.getInstance().toJson(websocketData), eventType);
}
@Override
public void onRuleChanged(final List<RuleData> ruleDataList, final DataEventTypeEnum eventType) {
WebsocketData<RuleData> configData =
new WebsocketData<>(ConfigGroupEnum.RULE.name(), eventType.name(), ruleDataList);
WebsocketCollector.send(GsonUtils.getInstance().toJson(configData), eventType);
}
@Override
public void onAppAuthChanged(final List<AppAuthData> appAuthDataList, final DataEventTypeEnum eventType) {
WebsocketData<AppAuthData> configData =
new WebsocketData<>(ConfigGroupEnum.APP_AUTH.name(), eventType.name(), appAuthDataList);
WebsocketCollector.send(GsonUtils.getInstance().toJson(configData), eventType);
}
@Override
public void onMetaDataChanged(final List<MetaData> metaDataList, final DataEventTypeEnum eventType) {
WebsocketData<MetaData> configData =
newWebsocketData<>(ConfigGroupEnum.META_DATA.name(), eventType.name(), metaDataList); WebsocketCollector.send(GsonUtils.getInstance().toJson(configData), eventType); }}Copy the codeWhen you see the code that encapsulates the data as WebsocketData, it sends it out using the webSocketController.send method.
2.6 Synchronize data to soul-bootstrap
// WebsocketCollector.java
@Slf4j
@ServerEndpoint("/websocket")
public class WebsocketCollector {
private static final Set<Session> SESSION_SET = new CopyOnWriteArraySet<>();
private static final String SESSION_KEY = "sessionKey";
/**
* On open.
*
* @param session the session
*/
@OnOpen
public void onOpen(final Session session) {
log.info("websocket on open successful....");
SESSION_SET.add(session);
}
/**
* On message.
*
* @param message the message
* @param session the session
*/
@OnMessage
public void onMessage(final String message, final Session session) {
if (message.equals(DataEventTypeEnum.MYSELF.name())) {
try {
ThreadLocalUtil.put(SESSION_KEY, session);
SpringBeanUtils.getInstance().getBean(SyncDataService.class).syncAll(DataEventTypeEnum.MYSELF);
} finally{ ThreadLocalUtil.clear(); }}}/**
* On close.
*
* @param session the session
*/
@OnClose
public void onClose(final Session session) {
SESSION_SET.remove(session);
ThreadLocalUtil.clear();
}
/**
* On error.
*
* @param session the session
* @param error the error
*/
@OnError
public void onError(final Session session, final Throwable error) {
SESSION_SET.remove(session);
ThreadLocalUtil.clear();
log.error("websocket collection error: ", error);
}
/**
* Send.
*
* @param message the message
* @param type the type
*/
public static void send(final String message, final DataEventTypeEnum type) {
if (StringUtils.isNotBlank(message)) {
if (DataEventTypeEnum.MYSELF == type) {
try {
Session session = (Session) ThreadLocalUtil.get(SESSION_KEY);
if(session ! =null) { session.getBasicRemote().sendText(message); }}catch (IOException e) {
log.error("websocket send result is exception: ", e);
}
return;
}
for (Session session : SESSION_SET) {
try {
session.getBasicRemote().sendText(message);
} catch (IOException e) {
log.error("websocket send result is exception: ", e);
}
}
}
}
}
Copy the codeThe WebsocketController uses the @Serverendpoint (“/ webSocket “) annotation to open a WebSocket service interface waiting for a connection.
When soul-bootstrap is started, the websocket is connected, and the onOpen method is triggered to store the Session of the connection information in the SESSION_SET Set.
In the send method, the DataEventTypeEnum type is checked first to determine whether it is MYSELF. This type can be traced back to 2.3-2.4. This is UPDATE, and it needs to be added later.
The following for loop iterates through all webSocket connections to the Session, sending the changed data.
At this point, the default WebSocket synchronization data policy is analyzed.
Mr.
Data Synchronization between background and Gateway (Websocket)
How to set up Websocket in background?
DataSyncConfiguration: As a configuration factory for Spring beans, you can build various listeners based on configuration information, including HTTP long polling, Zookeeper, Nacos, and Websocket methods.
@Configuration
public class DataSyncConfiguration {
/ / soul - admin in configuration information of the project, use soul. Sync. Websocket. Enabled the websocket being opened or closed
@Configuration
@ConditionalOnProperty(name = "soul.sync.websocket.enabled", havingValue = "true", matchIfMissing = true)
@EnableConfigurationProperties(WebsocketSyncProperties.class)
static class WebsocketListener {
@Bean
@ConditionalOnMissingBean(WebsocketCollector.class)
public WebsocketCollector websocketCollector(a) {
return newWebsocketCollector(); }}}Copy the codeWebsocketListener: An internal class of DataSyncConfiguration that initializes webSocket listeners. WebsocketCollector: Monitors websocket connections and receives information, maintains all session sessions connected to the background, and provides the send() method to notify session information.
How does a gateway set up a Websocket?
WebsocketSyncDataConfiguration: As the configuration factory of Spring Bean, it is the gateway to build Websocket communication.
@Configuration
@ConditionalOnClass(WebsocketSyncDataService.class)
@ConditionalOnProperty(prefix = "soul.sync.websocket", name = "urls")
@Slf4j
public class WebsocketSyncDataConfiguration {
// Collect all subscribers registered as beans, such as PluginDataSubscriber, MetaDataSubscriber, AuthDataSubscriber
@Bean
public SyncDataService websocketSyncDataService(final ObjectProvider<WebsocketConfig> websocketConfig, final ObjectProvider<PluginDataSubscriber> pluginSubscriber, final ObjectProvider<List<MetaDataSubscriber>> metaSubscribers, final ObjectProvider<List<AuthDataSubscriber>> authSubscribers) {
log.info("you use websocket sync soul data.......");
return new WebsocketSyncDataService(websocketConfig.getIfAvailable(WebsocketConfig::new), pluginSubscriber.getIfAvailable(), metaSubscribers.getIfAvailable(Collections::emptyList), authSubscribers.getIfAvailable(Collections::emptyList));
}
// In the configuration information for the soul-bootstrap project, use soul.sync.websocket to configure the background path to establish the connection
@Bean
@ConfigurationProperties(prefix = "soul.sync.websocket")
public WebsocketConfig websocketConfig(a) {
return newWebsocketConfig(); }}Copy the codeWebsocketSyncDataService: Get all the WebsocketConfig registered as beans, as well as the various DataSubscriber subscribers, and build a SoulWebsocketClient list that implements WebsocketClient
SoulWebsocketClient: Websocket communication class. It monitors Websocket connections and receives information, and notifies each subscriber when receiving information from the background.
public final class SoulWebsocketClient extends WebSocketClient {
private final WebsocketDataHandler websocketDataHandler;
private void handleResult(final String result) {
WebsocketData websocketData = GsonUtils.getInstance().fromJson(result, WebsocketData.class);
ConfigGroupEnum groupEnum = ConfigGroupEnum.acquireByName(websocketData.getGroupType());
// Get the event type of the data change based on the incoming information, such as refresh, update, delete, etcString eventType = websocketData.getEventType(); String json = GsonUtils.getInstance().toJson(websocketData.getData()); websocketDataHandler.executor(groupEnum, json, eventType); }}Copy the codeWebsocketDataHandler: Various data handling classes that implement AbstractDataHandler are built and cached during initialization.
public class WebsocketDataHandler {
// Cache all DataHandler data change classes
private static final EnumMap<ConfigGroupEnum, DataHandler> ENUM_MAP = new EnumMap<>(ConfigGroupEnum.class);
public WebsocketDataHandler(final PluginDataSubscriber pluginDataSubscriber,
final List<MetaDataSubscriber> metaDataSubscribers,
final List<AuthDataSubscriber> authDataSubscribers) {
ENUM_MAP.put(ConfigGroupEnum.PLUGIN, new PluginDataHandler(pluginDataSubscriber));
ENUM_MAP.put(ConfigGroupEnum.SELECTOR, new SelectorDataHandler(pluginDataSubscriber));
ENUM_MAP.put(ConfigGroupEnum.RULE, new RuleDataHandler(pluginDataSubscriber));
ENUM_MAP.put(ConfigGroupEnum.APP_AUTH, new AuthDataHandler(authDataSubscribers));
ENUM_MAP.put(ConfigGroupEnum.META_DATA, new MetaDataHandler(metaDataSubscribers));
}
public void executor(final ConfigGroupEnum type, final String json, final String eventType) {
// Call the corresponding DataHandler class based on the data change event typeENUM_MAP.get(type).handle(json, eventType); }}Copy the codeGateway data change call chain
SoulWebsocketClient, the portal class that implements Websocket communication, calls the Executor () method of the WebsocketDataHandler to match the information type. Handler () of the corresponding DataHandler is called to process the information.
AbstractDataHandler: Implements the handler() method that calls the corresponding event abstraction method based on the event type (such as refresh, update, create, delete, etc.).
public abstract class AbstractDataHandler<T> implements DataHandler {
The called methods are implemented by subclasses, since different types of metadata handling classes are handled differently
@Override
public void handle(final String json, final String eventType) {
List<T> dataList = convert(json);
if (CollectionUtils.isNotEmpty(dataList)) {
DataEventTypeEnum eventTypeEnum = DataEventTypeEnum.acquireByName(eventType);
switch (eventTypeEnum) {
case REFRESH:
case MYSELF:
doRefresh(dataList);
break;
case UPDATE:
case CREATE:
doUpdate(dataList);
break;
case DELETE:
doDelete(dataList);
break;
default:
break; }}}}Copy the codeXXXDataHandler: This refers to the various implementation classes of AbstractDataHandler (e.g. PluginDataHandler, etc.) that call their subscribers.
Different DataHandlers call different subscription methods:
PluginDataHandlerWill be calledonSubscribe()Notify plug-ins of metadata changesSelectorDataHandlerWill be calledonSelectorSubscribe()Notifies the selector of metadata changesRuleDataHandlerWill be calledonRuleSubscribe()Notify rule metadata changes
@RequiredArgsConstructor
public class PluginDataHandler extends AbstractDataHandler<PluginData> {
private final PluginDataSubscriber pluginDataSubscriber;
@Override
protected void doUpdate(final List<PluginData> dataList) {
// Call onSubscribe() to send the data object PluginData
dataList.forEach(pluginDataSubscriber::onSubscribe);
}
// ...
}
Copy the codeCommonPluginDataSubscriber: The onSubscribe() method of the subscriber notifies all PluginDataHandler classes injected into beans (not to be confused with the previous class of the same name, which is an interface under soul-plugin-base and implements classes in various pluggable plug-in packages).
public class CommonPluginDataSubscriber implements PluginDataSubscriber {
Collect and cache all data handlers registered as beans, such as the DividePluginDataHandler under DIVIDE, the HTTP plug-in
private final Map<String, PluginDataHandler> handlerMap;
// Plug-in metadata changes call
@Override
public void onSubscribe(final PluginData pluginData) {
BaseDataCache.getInstance().cachePluginData(pluginData);
Optional.ofNullable(handlerMap.get(pluginData.getName())).ifPresent(handler -> handler.handlerPlugin(pluginData));
}
// Selector metadata changes call
@Override
public void onSelectorSubscribe(final SelectorData selectorData) {
BaseDataCache.getInstance().cacheSelectData(selectorData);
Optional.ofNullable(handlerMap.get(selectorData.getPluginName())).ifPresent(handler -> handler.handlerSelector(selectorData));
}
// Rule metadata change calls
@Override
public void onRuleSubscribe(final RuleData ruleData) { BaseDataCache.getInstance().cacheRuleData(ruleData); Optional.ofNullable(handlerMap.get(ruleData.getPluginName())).ifPresent(handler -> handler.handlerRule(ruleData)); }}Copy the codeTIPS
There are two classes named PluginDataHandler under the project soul-sync-data-webSocket, which is used to notify plug-ins of metadata changes, and soul-plugin-base, which is used to notify plug-ins of metadata changes. Each type of metadata update used to define each plug-in.
Soul-sync-data-websocket class name “plugin” refers to the metadata type of the plug-in class, soul-plugin-base class name “plugin” refers to the subclass derived from each interpotable plug-in. Divide, Dubbo plug-ins, etc