Merge remote-tracking branch 'apache/master'

shardingsphere-governance/shardingsphere-governance-core/shardingsphere-governance-core-config/src/main/java/org/apache/shardingsphere/governance/core/config/ConfigCenter.java

@@ -21,16 +21,18 @@ import com.google.common.base.Joiner;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
+import com.google.common.eventbus.Subscribe;
 import org.apache.shardingsphere.encrypt.algorithm.config.AlgorithmProvidedEncryptRuleConfiguration;
 import org.apache.shardingsphere.encrypt.api.config.EncryptRuleConfiguration;
 import org.apache.shardingsphere.infra.auth.Authentication;
 import org.apache.shardingsphere.infra.auth.yaml.config.YamlAuthenticationConfiguration;
 import org.apache.shardingsphere.infra.auth.yaml.swapper.AuthenticationYamlSwapper;
-import org.apache.shardingsphere.infra.callback.governance.DataSourceCallback;
-import org.apache.shardingsphere.infra.callback.governance.RuleCallback;
-import org.apache.shardingsphere.infra.callback.governance.SchemaNameCallback;
 import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
 import org.apache.shardingsphere.infra.config.RuleConfiguration;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.DataSourceEvent;
+import org.apache.shardingsphere.infra.eventbus.event.RuleEvent;
+import org.apache.shardingsphere.infra.eventbus.event.SchemaNameEvent;
 import org.apache.shardingsphere.infra.yaml.config.YamlRootRuleConfigurations;
 import org.apache.shardingsphere.infra.yaml.engine.YamlEngine;
 import org.apache.shardingsphere.infra.yaml.swapper.YamlRuleConfigurationSwapperEngine;
@@ -66,9 +68,7 @@ public final class ConfigCenter {
     public ConfigCenter(final ConfigurationRepository repository) {
         node = new ConfigCenterNode();
         this.repository = repository;
-        DataSourceCallback.getInstance().register(this::persistDataSourceConfiguration);
-        RuleCallback.getInstance().register(this::persistRuleConfigurations);
-        SchemaNameCallback.getInstance().register(this::persistSchema);
+        ShardingSphereEventBus.getInstance().register(this);
     }
     
     /**
@@ -81,7 +81,7 @@ public final class ConfigCenter {
      */
     public void persistConfigurations(final String schemaName, final Map<String, DataSourceConfiguration> dataSourceConfigs,
                                       final Collection<RuleConfiguration> ruleConfigurations, final boolean isOverwrite) {
-        persistDataSourceConfiguration(schemaName, dataSourceConfigs, isOverwrite);
+        persistDataSourceConfigurations(schemaName, dataSourceConfigs, isOverwrite);
         persistRuleConfigurations(schemaName, ruleConfigurations, isOverwrite);
         // TODO Consider removing the following one.
         persistSchemaName(schemaName, isOverwrite);
@@ -99,14 +99,43 @@ public final class ConfigCenter {
         persistProperties(props, isOverwrite);
     }
     
-    private void persistDataSourceConfiguration(final String schemaName, final Map<String, DataSourceConfiguration> dataSourceConfigurations, final boolean isOverwrite) {
+    /**
+     * persist data source configurations.
+     * @param event Data source event.
+     */
+    @Subscribe
+    public synchronized void renew(final DataSourceEvent event) {
+        persistDataSourceConfigurations(event.getSchemaName(), event.getDataSourceConfigurations());
+    }
+    
+    /**
+     * Persist rule configurations.
+     * 
+     * @param event Rule event.
+     */
+    @Subscribe
+    public synchronized void renew(final RuleEvent event) {
+        persistRuleConfigurations(event.getSchemaName(), event.getRuleConfigurations());
+    }
+    
+    /**
+     * Persist schema name.
+     * 
+     * @param event Schema name event.
+     */
+    @Subscribe
+    public synchronized void renew(final SchemaNameEvent event) {
+        persistSchemaName(event.getSchemaName(), event.isOverwrite());
+    }
+    
+    private void persistDataSourceConfigurations(final String schemaName, final Map<String, DataSourceConfiguration> dataSourceConfigurations, final boolean isOverwrite) {
         if (dataSourceConfigurations.isEmpty() || !isOverwrite) {
             return;
         }
-        persistDataSourceConfiguration(schemaName, dataSourceConfigurations);
+        persistDataSourceConfigurations(schemaName, dataSourceConfigurations);
     }
     
-    private void persistDataSourceConfiguration(final String schemaName, final Map<String, DataSourceConfiguration> dataSourceConfigurations) {
+    private void persistDataSourceConfigurations(final String schemaName, final Map<String, DataSourceConfiguration> dataSourceConfigurations) {
         Preconditions.checkState(null != dataSourceConfigurations && !dataSourceConfigurations.isEmpty(), "No available data source in `%s` for governance.", schemaName);
         Map<String, YamlDataSourceConfiguration> yamlDataSourceConfigurations = dataSourceConfigurations.entrySet().stream()
                 .collect(Collectors.toMap(Entry::getKey, entry -> new DataSourceConfigurationYamlSwapper().swapToYamlConfiguration(entry.getValue())));

shardingsphere-governance/shardingsphere-governance-core/shardingsphere-governance-core-metadata/src/main/java/org/apache/shardingsphere/governance/core/metadata/MetaDataCenter.java

@@ -18,10 +18,12 @@
 package org.apache.shardingsphere.governance.core.metadata;
 
 import com.google.common.base.Strings;
-import org.apache.shardingsphere.infra.callback.governance.MetaDataCallback;
+import com.google.common.eventbus.Subscribe;
 import org.apache.shardingsphere.governance.repository.api.GovernanceRepository;
 import org.apache.shardingsphere.governance.core.metadata.yaml.RuleSchemaMetaDataYamlSwapper;
 import org.apache.shardingsphere.governance.core.metadata.yaml.YamlRuleSchemaMetaData;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.MetaDataEvent;
 import org.apache.shardingsphere.infra.metadata.schema.RuleSchemaMetaData;
 import org.apache.shardingsphere.infra.yaml.engine.YamlEngine;
 
@@ -39,7 +41,7 @@ public final class MetaDataCenter {
     public MetaDataCenter(final GovernanceRepository governanceRepository) {
         node = new MetaDataCenterNode();
         repository = governanceRepository;
-        MetaDataCallback.getInstance().register(this::persistMetaDataCenterNode);
+        ShardingSphereEventBus.getInstance().register(this);
     }
     
     /**
@@ -65,4 +67,14 @@ public final class MetaDataCenter {
         }
         return Optional.of(new RuleSchemaMetaDataYamlSwapper().swapToObject(YamlEngine.unmarshal(path, YamlRuleSchemaMetaData.class)));
     }
+    
+    /**
+     * Persist meta data.
+     * 
+     * @param event Meta data event.
+     */
+    @Subscribe
+    public synchronized void renew(final MetaDataEvent event) {
+        persistMetaDataCenterNode(event.getSchemaName(), event.getMetaData());
+    }
 }

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/callback/Callback.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback;
-
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.function.BiConsumer;
-
-/**
- * Callback.
- */
-// TODO rename schema based Callback
-public abstract class Callback<T> {
-    
-    private final List<BiConsumer<String, T>> consumers = new CopyOnWriteArrayList<>();
-    
-    /**
-     * Register consumer.
-     *
-     * @param consumer consumer
-     */
-    public void register(final BiConsumer<String, T> consumer) {
-        consumers.add(consumer);
-    }
-    
-    /**
-     * Run consumer.
-     *
-     * @param schemaName schema name
-     * @param arg another argument
-     */
-    public void run(final String schemaName, final T arg) {
-        for (BiConsumer<String, T> each : consumers) {
-            each.accept(schemaName, arg);
-        }
-    }
-}

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/callback/governance/RuleCallback.java → shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/eventbus/ShardingSphereEventBus.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback.governance;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.apache.shardingsphere.infra.callback.Callback;
-import org.apache.shardingsphere.infra.config.RuleConfiguration;
-
-import java.util.Collection;
-
-/**
- * Rule callback.
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public final class RuleCallback extends Callback<Collection<RuleConfiguration>> {
-    
-    private static final RuleCallback INSTANCE = new RuleCallback();
-    
-    /**
-     * Get instance.
-     *
-     * @return rule callback
-     */
-    public static RuleCallback getInstance() {
-        return INSTANCE;
-    }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus;
+
+import com.google.common.eventbus.EventBus;
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+/**
+ * ShardingSphere event bus.
+ */
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public final class ShardingSphereEventBus {
+    
+    /**
+     * Get instance of ShardingSphere event bus.
+     *
+     * @return instance of ShardingSphere event bus
+     */
+    public static EventBus getInstance() {
+        return ShardingSphereEventBusHolder.INSTANCE;
+    }
+    
+    private static final class ShardingSphereEventBusHolder {
+        private static final EventBus INSTANCE = new EventBus();
+    }
+}

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/callback/governance/DataSourceCallback.java → shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/eventbus/event/DataSourceEvent.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback.governance;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.apache.shardingsphere.infra.callback.Callback;
-import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
-
-import java.util.Map;
-
-/**
- * Data source callback.
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public final class DataSourceCallback extends Callback<Map<String, DataSourceConfiguration>> {
-    
-    private static final DataSourceCallback INSTANCE = new DataSourceCallback();
-    
-    /**
-     * Get instance.
-     *
-     * @return data source callback
-     */
-    public static DataSourceCallback getInstance() {
-        return INSTANCE;
-    }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
+
+import java.util.Map;
+
+/**
+ * Data source event.
+ */
+@RequiredArgsConstructor
+@Getter
+public class DataSourceEvent {
+    
+    private final String schemaName;
+    
+    private final Map<String, DataSourceConfiguration> dataSourceConfigurations;
+}

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/callback/governance/MetaDataCallback.java → shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/eventbus/event/MetaDataEvent.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback.governance;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.apache.shardingsphere.infra.callback.Callback;
-import org.apache.shardingsphere.infra.metadata.schema.RuleSchemaMetaData;
-
-/**
- * Meta data call back.
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public final class MetaDataCallback extends Callback<RuleSchemaMetaData> {
-    
-    private static final MetaDataCallback INSTANCE = new MetaDataCallback();
-    
-    /**
-     * Get instance.
-     *
-     * @return meta data call back
-     */
-    public static MetaDataCallback getInstance() {
-        return INSTANCE;
-    }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.metadata.schema.RuleSchemaMetaData;
+
+/**
+ * Meta data event.
+ */
+@RequiredArgsConstructor
+@Getter
+public class MetaDataEvent {
+    
+    private final String schemaName;
+    
+    private final RuleSchemaMetaData metaData;
+}

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/callback/governance/SchemaNameCallback.java → shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/eventbus/event/RuleEvent.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback.governance;
-
-import lombok.AccessLevel;
-import lombok.NoArgsConstructor;
-import org.apache.shardingsphere.infra.callback.Callback;
-
-/**
- * Schema name callback.
- */
-@NoArgsConstructor(access = AccessLevel.PRIVATE)
-public final class SchemaNameCallback extends Callback<Boolean> {
-    
-    private static final SchemaNameCallback INSTANCE = new SchemaNameCallback();
-    
-    /**
-     * Get instance.
-     *
-     * @return data source callback
-     */
-    public static SchemaNameCallback getInstance() {
-        return INSTANCE;
-    }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.infra.config.RuleConfiguration;
+
+import java.util.Collection;
+
+/**
+ * Rule event.
+ */
+@RequiredArgsConstructor
+@Getter
+public class RuleEvent {
+    
+    private final String schemaName;
+    
+    private final Collection<RuleConfiguration> ruleConfigurations;
+}

shardingsphere-infra/shardingsphere-infra-common/src/main/java/org/apache/shardingsphere/infra/eventbus/event/SchemaNameEvent.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Schema name event.
+ */
+@RequiredArgsConstructor
+@Getter
+public class SchemaNameEvent {
+    
+    private final String schemaName;
+    
+    private final boolean overwrite; 
+}

shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/callback/CallbackTest.java

-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.shardingsphere.infra.callback;
-
-import org.apache.shardingsphere.infra.callback.governance.DataSourceCallback;
-import org.apache.shardingsphere.infra.callback.governance.MetaDataCallback;
-import org.apache.shardingsphere.infra.callback.governance.RuleCallback;
-import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
-import org.apache.shardingsphere.infra.config.RuleConfiguration;
-import org.apache.shardingsphere.infra.metadata.schema.RuleSchemaMetaData;
-import org.junit.Test;
-
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.Map;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertThat;
-
-public final class CallbackTest {
-    
-    private static final String TEST = "test";
-    
-    @Test
-    public void assertRun() {
-        MetaDataCallback.getInstance().run(TEST, new RuleSchemaMetaData(null, null));
-        MetaDataCallback.getInstance().register((schemaName, ruleSchemaMetaData) -> {
-            assertThat(schemaName, is(TEST));
-            assertThat(ruleSchemaMetaData.getClass().getName(), is(RuleSchemaMetaData.class.getName()));
-        });
-        MetaDataCallback.getInstance().run(TEST, new RuleSchemaMetaData(null, null));
-        
-        DataSourceCallback.getInstance().run(TEST, new HashMap<>());
-        DataSourceCallback.getInstance().register((schemaName, map) -> {
-            assertThat(schemaName, is(TEST));
-            assertThat(map.size(), is(1));
-            map.forEach((k, v) -> assertThat(v.getClass().getName(), is(DataSourceConfiguration.class.getName())));
-        });
-        Map<String, DataSourceConfiguration> maps = new HashMap<>();
-        DataSourceConfiguration configuration = new DataSourceConfiguration("test");
-        maps.put(TEST, configuration);
-        DataSourceCallback.getInstance().run(TEST, maps);
-        
-        RuleCallback.getInstance().run(TEST, new ArrayList<>());
-        RuleCallback.getInstance().register((schemaName, ruleConfigurations) -> {
-            assertThat(schemaName, is(TEST));
-            assertFalse(ruleConfigurations.isEmpty());
-            ruleConfigurations.forEach(each -> assertThat(each.getClass().getName(), is(TestRuleConfiguration.class.getName())));
-        });
-        Collection<RuleConfiguration> ruleConfigurations = new ArrayList<>();
-        ruleConfigurations.add(new TestRuleConfiguration());
-        RuleCallback.getInstance().run(TEST, ruleConfigurations);
-    }
-    
-    static class TestRuleConfiguration implements RuleConfiguration {
-    }
-}

shardingsphere-infra/shardingsphere-infra-common/src/test/java/org/apache/shardingsphere/infra/eventbus/ShardingSphereEventBusTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.infra.eventbus;
+
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.junit.Assert.assertThat;
+
+public class ShardingSphereEventBusTest {
+    
+    @Test
+    public void assertInstance() {
+        assertThat(ShardingSphereEventBus.getInstance(), is(ShardingSphereEventBus.getInstance()));
+    }
+}

shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/executor/PreparedStatementExecutor.java

@@ -18,10 +18,11 @@
 package org.apache.shardingsphere.driver.executor;
 
 import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.callback.governance.MetaDataCallback;
 import org.apache.shardingsphere.infra.context.SchemaContext;
 import org.apache.shardingsphere.infra.context.SchemaContexts;
 import org.apache.shardingsphere.infra.database.DefaultSchema;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.MetaDataEvent;
 import org.apache.shardingsphere.infra.executor.kernel.InputGroup;
 import org.apache.shardingsphere.infra.executor.sql.ConnectionMode;
 import org.apache.shardingsphere.infra.executor.sql.QueryResult;
@@ -171,7 +172,7 @@ public final class PreparedStatementExecutor {
             refreshStrategy.get().refreshMetaData(schemaContext.getSchema().getMetaData(), schemaContexts.getDatabaseType(),
                     dataSourceMap, sqlStatementContext, tableName -> metaDataLoader.load(schemaContexts.getDatabaseType(),
                             dataSourceMap, tableName, schemaContexts.getProps()));
-            MetaDataCallback.getInstance().run(DefaultSchema.LOGIC_NAME, schemaContext.getSchema().getMetaData().getSchema());
+            ShardingSphereEventBus.getInstance().post(new MetaDataEvent(DefaultSchema.LOGIC_NAME, schemaContext.getSchema().getMetaData().getSchema()));
         }
     }
 }

shardingsphere-jdbc/shardingsphere-jdbc-core/src/main/java/org/apache/shardingsphere/driver/executor/StatementExecutor.java

@@ -18,10 +18,11 @@
 package org.apache.shardingsphere.driver.executor;
 
 import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.callback.governance.MetaDataCallback;
 import org.apache.shardingsphere.infra.context.SchemaContext;
 import org.apache.shardingsphere.infra.context.SchemaContexts;
 import org.apache.shardingsphere.infra.database.DefaultSchema;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.MetaDataEvent;
 import org.apache.shardingsphere.infra.executor.kernel.InputGroup;
 import org.apache.shardingsphere.infra.executor.sql.ConnectionMode;
 import org.apache.shardingsphere.infra.executor.sql.QueryResult;
@@ -246,7 +247,7 @@ public final class StatementExecutor {
             RuleSchemaMetaDataLoader metaDataLoader = new RuleSchemaMetaDataLoader(schemaContext.getSchema().getRules());
             refreshStrategy.get().refreshMetaData(schemaContext.getSchema().getMetaData(), schemaContexts.getDatabaseType(), dataSourceMap, sqlStatementContext,
                 tableName -> metaDataLoader.load(schemaContexts.getDatabaseType(), dataSourceMap, tableName, schemaContexts.getProps()));
-            MetaDataCallback.getInstance().run(DefaultSchema.LOGIC_NAME, schemaContext.getSchema().getMetaData().getSchema());
+            ShardingSphereEventBus.getInstance().post(new MetaDataEvent(DefaultSchema.LOGIC_NAME, schemaContext.getSchema().getMetaData().getSchema()));
         }
     }
     

shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/communication/jdbc/JDBCDatabaseCommunicationEngine.java

@@ -18,10 +18,11 @@
 package org.apache.shardingsphere.proxy.backend.communication.jdbc;
 
 import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.callback.governance.MetaDataCallback;
 import org.apache.shardingsphere.infra.config.exception.ShardingSphereConfigurationException;
 import org.apache.shardingsphere.infra.config.properties.ConfigurationPropertyKey;
 import org.apache.shardingsphere.infra.context.SchemaContext;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.MetaDataEvent;
 import org.apache.shardingsphere.infra.executor.sql.QueryResult;
 import org.apache.shardingsphere.infra.executor.sql.context.ExecutionContext;
 import org.apache.shardingsphere.infra.executor.sql.log.SQLLogger;
@@ -132,7 +133,7 @@ public final class JDBCDatabaseCommunicationEngine implements DatabaseCommunicat
         if (refreshStrategy.isPresent()) {
             refreshStrategy.get().refreshMetaData(schema.getSchema().getMetaData(), ProxySchemaContexts.getInstance().getSchemaContexts().getDatabaseType(),
                     schema.getSchema().getDataSources(), sqlStatementContext, this::loadTableMetaData);
-            MetaDataCallback.getInstance().run(schema.getName(), schema.getSchema().getMetaData().getSchema());
+            ShardingSphereEventBus.getInstance().post(new MetaDataEvent(schema.getName(), schema.getSchema().getMetaData().getSchema()));
         }
     }
     

shardingsphere-proxy/shardingsphere-proxy-backend/src/main/java/org/apache/shardingsphere/proxy/backend/text/admin/RDLBackendHandler.java

@@ -18,13 +18,14 @@
 package org.apache.shardingsphere.proxy.backend.text.admin;
 
 import lombok.RequiredArgsConstructor;
-import org.apache.shardingsphere.infra.callback.governance.DataSourceCallback;
-import org.apache.shardingsphere.infra.callback.governance.RuleCallback;
-import org.apache.shardingsphere.infra.callback.governance.SchemaNameCallback;
 import org.apache.shardingsphere.infra.config.DataSourceConfiguration;
 import org.apache.shardingsphere.infra.config.RuleConfiguration;
 import org.apache.shardingsphere.infra.context.impl.StandardSchemaContexts;
 import org.apache.shardingsphere.infra.database.type.DatabaseType;
+import org.apache.shardingsphere.infra.eventbus.ShardingSphereEventBus;
+import org.apache.shardingsphere.infra.eventbus.event.DataSourceEvent;
+import org.apache.shardingsphere.infra.eventbus.event.RuleEvent;
+import org.apache.shardingsphere.infra.eventbus.event.SchemaNameEvent;
 import org.apache.shardingsphere.infra.yaml.swapper.YamlRuleConfigurationSwapperEngine;
 import org.apache.shardingsphere.proxy.backend.communication.jdbc.connection.BackendConnection;
 import org.apache.shardingsphere.proxy.backend.exception.DBCreateExistsException;
@@ -78,8 +79,8 @@ public final class RDLBackendHandler implements TextProtocolBackendHandler {
         if (ProxySchemaContexts.getInstance().getSchemaNames().contains(context.getSqlStatement().getDatabaseName())) {
             return new ErrorResponse(new DBCreateExistsException(context.getSqlStatement().getDatabaseName()));
         }
-        SchemaNameCallback.getInstance().run(context.getSqlStatement().getDatabaseName(), false);
         // TODO Need to get the executed feedback from registry center for returning.
+        ShardingSphereEventBus.getInstance().post(new SchemaNameEvent(context.getSqlStatement().getDatabaseName(), true));
         UpdateResponse result = new UpdateResponse();
         result.setType("CREATE");
         return result;
@@ -89,8 +90,8 @@ public final class RDLBackendHandler implements TextProtocolBackendHandler {
         if (!ProxySchemaContexts.getInstance().getSchemaNames().contains(context.getSqlStatement().getDatabaseName())) {
             return new ErrorResponse(new DBCreateExistsException(context.getSqlStatement().getDatabaseName()));
         }
-        SchemaNameCallback.getInstance().run(context.getSqlStatement().getDatabaseName(), true);
         // TODO Need to get the executed feedback from registry center for returning.
+        ShardingSphereEventBus.getInstance().post(new SchemaNameEvent(context.getSqlStatement().getDatabaseName(), true));
         UpdateResponse result = new UpdateResponse();
         result.setType("DROP");
         return result;
@@ -100,7 +101,7 @@ public final class RDLBackendHandler implements TextProtocolBackendHandler {
         Map<String, YamlDataSourceParameter> parameters = new CreateDataSourcesStatementContextConverter().convert(context);
         Map<String, DataSourceConfiguration> dataSources = DataSourceConverter.getDataSourceConfigurationMap(DataSourceConverter.getDataSourceParameterMap2(parameters));
         // TODO Need to get the executed feedback from registry center for returning.
-        DataSourceCallback.getInstance().run(backendConnection.getSchema(), dataSources);
+        ShardingSphereEventBus.getInstance().post(new DataSourceEvent(backendConnection.getSchema(), dataSources));
         UpdateResponse result = new UpdateResponse();
         result.setType("CREATE");
         return result;
@@ -110,7 +111,7 @@ public final class RDLBackendHandler implements TextProtocolBackendHandler {
         YamlShardingRuleConfiguration configurations = new CreateShardingRuleStatementContextConverter().convert(context);
         Collection<RuleConfiguration> rules = new YamlRuleConfigurationSwapperEngine().swapToRuleConfigurations(Collections.singleton(configurations));
         // TODO Need to get the executed feedback from registry center for returning.
-        RuleCallback.getInstance().run(backendConnection.getSchema(), rules);
+        ShardingSphereEventBus.getInstance().post(new RuleEvent(backendConnection.getSchema(), rules));
         UpdateResponse result = new UpdateResponse();
         result.setType("CREATE");
         return result;