You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by su...@apache.org on 2024/06/11 02:01:09 UTC
(shardingsphere) branch master updated: Add InternalEventSubscriberRegistry (#31628)
This is an automated email from the ASF dual-hosted git repository.
sunnianjun pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new c7648695a52 Add InternalEventSubscriberRegistry (#31628)
c7648695a52 is described below
commit c7648695a521dd16cfdf7601fe137bb11679c557
Author: Haoran Meng <me...@gmail.com>
AuthorDate: Tue Jun 11 10:01:02 2024 +0800
Add InternalEventSubscriberRegistry (#31628)
---
.../cluster/ClusterContextManagerBuilder.java | 11 +++------
... => InternalQualifiedDataSourceSubscriber.java} | 2 +-
.../InternalEventSubscriberRegistry.java} | 27 +++++++---------------
.../QualifiedDataSourceStateSubscriberTest.java | 4 ++--
4 files changed, 14 insertions(+), 30 deletions(-)
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
index 67ffdc0701b..f536bc4b84e 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/ClusterContextManagerBuilder.java
@@ -31,13 +31,13 @@ import org.apache.shardingsphere.mode.lock.GlobalLockContext;
import org.apache.shardingsphere.mode.manager.ContextManager;
import org.apache.shardingsphere.mode.manager.ContextManagerBuilder;
import org.apache.shardingsphere.mode.manager.ContextManagerBuilderParameter;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.listener.MetaDataChangedListener;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.listener.MetaDataWatchListenerManager;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.lock.GlobalLockPersistService;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.GovernanceWatcherFactory;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.listener.MetaDataChangedListener;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.nodes.storage.subscriber.internal.QualifiedDataSourceStatusSubscriber;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.workerid.generator.ClusterWorkerIdGenerator;
import org.apache.shardingsphere.mode.manager.cluster.coordinator.subscriber.ClusterEventSubscriberRegistry;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.subscriber.InternalEventSubscriberRegistry;
import org.apache.shardingsphere.mode.manager.cluster.exception.MissingRequiredClusterRepositoryConfigurationException;
import org.apache.shardingsphere.mode.metadata.MetaDataContexts;
import org.apache.shardingsphere.mode.metadata.MetaDataContextsFactory;
@@ -64,7 +64,6 @@ public final class ClusterContextManagerBuilder implements ContextManagerBuilder
MetaDataContexts metaDataContexts = MetaDataContextsFactory.create(metaDataPersistService, param, computeNodeInstanceContext,
new QualifiedDataSourceStatePersistService(repository).loadStates());
ContextManager result = new ContextManager(metaDataContexts, computeNodeInstanceContext, repository);
- createSubscribers(eventBusContext, repository);
registerOnline(eventBusContext, computeNodeInstanceContext, repository, param, result);
return result;
}
@@ -81,17 +80,13 @@ public final class ClusterContextManagerBuilder implements ContextManagerBuilder
new GlobalLockContext(new GlobalLockPersistService(repository)), eventBusContext);
}
- // TODO remove the method, only keep ZooKeeper's events, remove all decouple events
- private void createSubscribers(final EventBusContext eventBusContext, final ClusterPersistRepository repository) {
- eventBusContext.register(new QualifiedDataSourceStatusSubscriber(repository));
- }
-
private void registerOnline(final EventBusContext eventBusContext, final ComputeNodeInstanceContext computeNodeInstanceContext,
final ClusterPersistRepository repository, final ContextManagerBuilderParameter param, final ContextManager contextManager) {
contextManager.getPersistServiceFacade().getComputeNodePersistService().registerOnline(computeNodeInstanceContext.getInstance());
new GovernanceWatcherFactory(repository, eventBusContext).watchListeners();
watchDatabaseMetaDataListener(param, contextManager.getPersistServiceFacade().getMetaDataPersistService(), eventBusContext);
contextManager.getComputeNodeInstanceContext().getAllClusterInstances().addAll(contextManager.getPersistServiceFacade().getComputeNodePersistService().loadAllComputeNodeInstances());
+ new InternalEventSubscriberRegistry(contextManager, repository).register();
new ClusterEventSubscriberRegistry(contextManager, repository).register();
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/InternalQualifiedDataSourceSubscriber.java
similarity index 95%
copy from mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java
copy to mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/InternalQualifiedDataSourceSubscriber.java
index 21ff04409c5..50450cdd9b9 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/InternalQualifiedDataSourceSubscriber.java
@@ -28,7 +28,7 @@ import org.apache.shardingsphere.mode.storage.node.QualifiedDataSourceNode;
* Qualified data source status subscriber.
*/
@RequiredArgsConstructor
-public final class QualifiedDataSourceStatusSubscriber implements EventSubscriber {
+public final class InternalQualifiedDataSourceSubscriber implements EventSubscriber {
private final ClusterPersistRepository repository;
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/InternalEventSubscriberRegistry.java
similarity index 52%
rename from mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java
rename to mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/InternalEventSubscriberRegistry.java
index 21ff04409c5..eac9e3b9da6 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/nodes/storage/subscriber/internal/QualifiedDataSourceStatusSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/InternalEventSubscriberRegistry.java
@@ -15,30 +15,19 @@
* limitations under the License.
*/
-package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.nodes.storage.subscriber.internal;
+package org.apache.shardingsphere.mode.manager.cluster.coordinator.subscriber;
-import com.google.common.eventbus.Subscribe;
-import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.util.eventbus.EventSubscriber;
-import org.apache.shardingsphere.mode.event.node.QualifiedDataSourceDeletedEvent;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.nodes.storage.subscriber.internal.InternalQualifiedDataSourceSubscriber;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
-import org.apache.shardingsphere.mode.storage.node.QualifiedDataSourceNode;
+import org.apache.shardingsphere.mode.subsciber.EventSubscriberRegistry;
/**
- * Qualified data source status subscriber.
+ * Internal event subscriber registry.
*/
-@RequiredArgsConstructor
-public final class QualifiedDataSourceStatusSubscriber implements EventSubscriber {
+public class InternalEventSubscriberRegistry extends EventSubscriberRegistry {
- private final ClusterPersistRepository repository;
-
- /**
- * Delete qualified data source.
- *
- * @param event qualified data source deleted event
- */
- @Subscribe
- public void delete(final QualifiedDataSourceDeletedEvent event) {
- repository.delete(QualifiedDataSourceNode.getQualifiedDataSourceNodePath(event.getQualifiedDataSource()));
+ public InternalEventSubscriberRegistry(final ContextManager contextManager, final ClusterPersistRepository repository) {
+ super(contextManager, new InternalQualifiedDataSourceSubscriber(repository));
}
}
diff --git a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/storage/subscriber/QualifiedDataSourceStateSubscriberTest.java b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/storage/subscriber/QualifiedDataSourceStateSubscriberTest.java
index 4a680d40f28..c0d6fc37a13 100644
--- a/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/storage/subscriber/QualifiedDataSourceStateSubscriberTest.java
+++ b/mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/storage/subscriber/QualifiedDataSourceStateSubscriberTest.java
@@ -19,7 +19,7 @@ package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.stat
import org.apache.shardingsphere.infra.metadata.database.schema.QualifiedDataSource;
import org.apache.shardingsphere.mode.event.node.QualifiedDataSourceDeletedEvent;
-import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.nodes.storage.subscriber.internal.QualifiedDataSourceStatusSubscriber;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.nodes.storage.subscriber.internal.InternalQualifiedDataSourceSubscriber;
import org.apache.shardingsphere.mode.repository.cluster.ClusterPersistRepository;
import org.apache.shardingsphere.mode.storage.node.QualifiedDataSourceNode;
import org.junit.jupiter.api.Test;
@@ -41,7 +41,7 @@ class QualifiedDataSourceStateSubscriberTest {
String groupName = "readwrite_ds";
String dataSourceName = "replica_ds_0";
QualifiedDataSourceDeletedEvent event = new QualifiedDataSourceDeletedEvent(new QualifiedDataSource(databaseName, groupName, dataSourceName));
- new QualifiedDataSourceStatusSubscriber(repository).delete(event);
+ new InternalQualifiedDataSourceSubscriber(repository).delete(event);
verify(repository).delete(QualifiedDataSourceNode.getQualifiedDataSourceNodePath(new QualifiedDataSource(databaseName, groupName, dataSourceName)));
}
}